Streams

SubscribeOpen an event, input, or log stream

All three are Device calls that send the box a subscribe request.

filters is one CatchFilter or an iterable of them, each an address (a CatchClass plus an id in it) with an optional direction and capture. The box holds a 32-entry table.

CatchFilter.watch(usage)                # one button, key, or media usage
CatchFilter.watch_axis(axis)            # one Axis
CatchFilter.watch_class(input_class)    # every usage in one Class
CatchFilter.watch_axes()                # X, Y, the wheel and AC Pan
CatchFilter.all_input()                 # a list: all four input classes
CatchFilter.traffic(traffic_class, id)  # one endpoint, interface, or EP number
CatchFilter.traffic_class(tc)           # every id in one TrafficClass
CatchFilter.everything()                # every class, every id, one table entry
  .with_direction(direction)            # a Direction, default BOTH
  .with_capture(n)                      # bytes kept per event, default 0 = whole packet
  .on_press() / .on_release()           # one edge of an input filter
  .inbound() / .outbound()              # one flow of a traffic filter

Classes and their ids are on CatchClass; matching order on The table.

ReceiveBlocking, polling, timed, and iterated reads

Every stream has these four reads plus is_connected and close(). Shown for EventStream (yielding CatchEvent); InputStream and LogStream yield InputEvent and LogLine.

import threading
from medius import CatchFilter, Device

def read(events):
    for ev in events:          # ends when the stream closes
        print(ev)

with Device.find() as dev:
    events = dev.catch_events(CatchFilter.everything())
    reader = threading.Thread(target=read, args=(events,))
    reader.start()
    ...
    events.close()             # the loop ends at once
    reader.join()

InputStream has no clone(); open a second one instead.

Event objectsDataclasses recv() returns

Every object here is a dataclass.

CatchEvent
 ├─ kind    : CatchEventKind       MOTION = 0 · USAGES = 1 · TRAFFIC = 2
 ├─ ts_us   : int                  box microseconds; wraps every ~71.6 min
 ├─ clock   : ClockDomain          HOST_CHIP = 0 (real device side) · DEVICE_CHIP = 1 (clone side)
 ├─ payload : MotionEvent | UsageSnapshot | TrafficEvent
 │
 ├─ .motion  → MotionEvent | None      None unless kind == MOTION
 │               dx, dy, dz, pan
 ├─ .usages  → UsageSnapshot | None    None unless kind == USAGES
 │               usages[], cls, direction, is_held(usage)
 └─ .traffic → TrafficEvent | None     None unless kind == TRAFFIC
                 catch_class, id, direction, flags,
                 true_len, bytes, truncated()

Field meanings are on Types & errors; held usage ids come from the HID usage tables. flags is class-specific, read by one accessor per meaning: zlp() on the packet classes, bulk_end_of_transfer() on VENDOR_BULK, control_status() on CONTROL, transfer_status() on CLIP_TRANSFER, bus_event() on BUS.

Consume loopSubscribe, iterate, react

from medius import (Device, CatchFilter, CatchEventKind, Usage, Button)

with Device.find() as dev:
    filters = [
        CatchFilter.watch_axes(),                     # cursor and wheel
        CatchFilter.watch(Button.LEFT),               # one button only
    ]
    with dev.catch_events(filters) as events:
        for ev in events:                      # ends when the link drops
            if ev.kind == CatchEventKind.MOTION:
                m = ev.motion
                print(f"moved {m.dx},{m.dy}  wheel {m.dz}")
            elif ev.kind == CatchEventKind.USAGES:
                if ev.usages.is_held(Usage.button(Button.LEFT)):
                    print("left held")
            if events.dropped:
                print("fell behind:", events.dropped, "dropped")

from medius import Device, CatchFilter, CatchEventKind, TrafficClass

VI = TrafficClass.VENDOR_INTERRUPT
filters = [
    CatchFilter.traffic_class(VI).with_capture(16),  # the rest, 16 bytes
    CatchFilter.traffic(VI, 0x83),                   # this one, whole packets
    CatchFilter.traffic_class(TrafficClass.BUS),     # resets, configures, detach
]

with Device.find() as dev:
    with dev.catch_events(filters) as events:
        for ev in events:
            if ev.kind != CatchEventKind.TRAFFIC:
                continue
            t = ev.traffic
            cut = " (cut)" if t.truncated() else ""
            print(f"{ev.ts_us:>10} {ev.clock.name:<11} {t.catch_class.name:<16} "
                  f"ep={t.id:#04x} {t.direction.name:<8} "
                  f"{len(t.bytes)}/{t.true_len} bytes{cut}  {t.bytes.hex(' ')}")

truncated() separates a clipped capture from a short packet: a 16-byte capture of a 64-byte report and a real 16-byte report differ only in true_len.

events = dev.catch_events(CatchFilter.everything().with_capture(16))
while running:
    ev = events.recv_timeout(50)   # wake every 50 ms to do other work
    if ev is None:
        continue
    handle(ev)

st = dev.query_catch()
if st.table_full:
    print("a filter was refused: the box's 32-entry table is full")
for e in st.entries:
    print(f"{e.filter!r}  dropped={e.dropped}")
print("box-wide dropped:", st.dropped, " clock age:", st.clock.age_ms)

Decoded inputPress and release edges, not snapshots

The box reports a held-usage snapshot per report; dev.input_events(filters) diffs each against the stream's held set and yields the edges.

Every filter must name an input class and cover both edges. A traffic class raises NotAnInputFilterError, everything() raises WildcardNotInputError, and one narrowed with on_press() raises HalfEdgeInputFilterError.

from medius import Class, CatchFilter, Device, InputKind

with Device.find() as dev:
    with dev.input_events(CatchFilter.all_input()) as inputs:
        for ev in inputs:
            if ev.kind == InputKind.MOTION:
                print(f"{ev.ts_us:>10}  move {ev.dx},{ev.dy} wheel {ev.dz}")
            else:
                edge = "down" if ev.is_press else "up"
                print(f"{ev.ts_us:>10}  {ev.usage!r} {edge}")
                print("   still held:", inputs.held(Class.KEY))

TimelinePut box stamps on this machine's clock

A catch stamp is microseconds on a chip's clock. It wraps every ~71.6 minutes and relates to no clock on this machine. Timeline maps it onto this machine's clock.

Feed every event in arrival order. Each domain is tracked separately; the mapping improves as it runs.

from medius import CatchFilter, Device, Timeline

with Device.find() as dev:
    with dev.catch_events(CatchFilter.watch_axes()) as events, Timeline() as time:
        for ev in events:
            at = time.observe(ev)
            print(f"{at.host_ns:>16} ns  box {at.box_us} us  +{at.excess_ns} jitter")

excess_ns is how much later than the measured floor the event arrived: jitter, not latency.

No asyncBuild async on the non-blocking reads