Medius - BindingsFirst program

First program

Connect, move, click, read one event

One file that finds the box, moves the cursor, clicks, and reads one physical event. Install with pip (pip install medius, see Install). What each call does lives in the Rust Library and Native API.

Fire-and-forget calls return the moment they're queued (the injection model); Blocks calls wait for the box's reply. Either kind raises a MediusError on failure.

The program

The full listing
import time
from medius import Device, Usage, Button, CatchFilter, NotFoundError

try:
    with Device.find() as dev:                       # open first box + handshake
        v = dev.query_version()                      # blocks for one reply
        print(f"firmware {v.fw_major}.{v.fw_minor}.{v.fw_patch}, proto {v.proto_ver}")

        dev.move_rel(100, 0)                         # nudge cursor 100 right
        dev.press(Usage.button(Button.LEFT))         # hold left down
        time.sleep(0.02)
        dev.soft_release(Usage.button(Button.LEFT))  # let it back up

        with dev.catch_events(CatchFilter.everything()) as stream:  # subscribe to everything
            event = stream.recv_timeout(5000)                      # one event, or None after 5 s
            if event is None:
                print("no physical input within 5 s")
            elif event.motion:
                m = event.motion
                print(f"motion  dx={m.dx} dy={m.dy} wheel={m.dz}")
            elif event.usages and event.usages.is_held(Usage.button(Button.LEFT)):
                print("left button held")
            elif event.traffic:
                t = event.traffic
                print(f"traffic {t.catch_class.name} id={t.id:#04x} "
                      f"{len(t.bytes)} of {t.true_len} bytes")
            else:
                print(f"event  {event.kind.name}")
        # stream + link are closed here, on block exit
except NotFoundError:
    raise SystemExit("no medius box found: check the control-port cable")

The cursor won't move and the click won't land on this machine: injection only reaches the game PC through the clone port, not the control PC running this script. See Hardware for the port map.

Walkthrough

What each call is and where its meaning lives
CallKindDoes (follow the link)
Device.find()BlocksOpen the first box and run the handshake. See Connection.
dev.query_version()BlocksRead a Version. See Requests.
dev.move_rel(100, 0)Fire-and-forgetRelative cursor move. See MOVE / Move.
dev.press(Usage.button(Button.LEFT))Fire-and-forgetHold a button. See the injection model / Inject.
dev.soft_release(Usage.button(Button.LEFT))Fire-and-forgetRelease, unless the user is physically holding it. See Injection.
dev.catch_events(CatchFilter.everything())Fire-and-forgetSubscribe with one CatchFilter (or an iterable); returns an EventStream. See Catch.
stream.recv_timeout(5000)BlocksWait up to 5 s for one CatchEvent, or None. More on Streams.

Button ids are on Usage IDs; the full call surface is the API index.

Run it

One command, expected output
python first.py
# firmware 3.2.0, proto 5
# motion  dx=8 dy=-3 wheel=0

The motion line needs a real mouse move or click inside the 5-second window. CatchFilter.everything() subscribes to every class, so a traffic line can arrive first on a device with busy vendor endpoints; narrow it with CatchFilter.watch_axes().

Reading events

Four ways to pull from a stream

The program uses recv_timeout so it can't hang.

ReaderEmpty queueLink dropped
stream.recv()blocks foreverraises DisconnectedError
stream.recv_timeout(ms)returns None after msreturns None (same as a timeout)
stream.try_recv()returns None at oncereturns None (same as empty)
for ev in stream:blocks for eachloop ends cleanly

A CatchEvent carries one of .motion / .usages / .traffic (the other two are None), plus .ts_us and the .clock domain that stamped it. A UsageSnapshot has is_held(usage) for any built Usage; a TrafficEvent has truncated(). Full payload shapes on Streams.

To skip diffing snapshots yourself, subscribe with dev.input_events() instead: it yields press and release edges directly.

Errors

Failures raise; they're never a return code

Every Device and stream call raises on failure. Each exception is a MediusError subclass carrying .status, .message, and .proto_ver.

WhenRaises
find() with no box presentNotFoundError
a query_* gets no reply in timeQueryTimeoutError
the box speaks a different protocolBadProtoVerError (check .proto_ver)
a stream read after the link dropsDisconnectedError
any other failurea MediusError subclass

Catch the base MediusError to handle them all at once. Full list and the Status codes on Types & errors; patterns on Calls & errors.

Closing

A with-block, close(), or garbage collection

A Device and each stream hold a live connection. Close it three ways, all safe to combine:

MechanismFrees when
with Device.find() as dev:the context manager exits (used twice above, for the link and the stream)
dev.close()you call it; idempotent, a second call is a no-op
garbage collectionthe object is collected, via __del__, if you forgot

dev.clone() returns another handle to the same link; the connection lives until the last handle is freed. See Lifecycle and the keepalive thread.