Sync wrapper¶

anyserial is async-first. The anyserial.sync.SerialPort class is a thin blocking wrapper around anyserial.SerialPort for scripts and test benches that do not want to run an event loop themselves.

Quickstart¶

from anyserial.sync import SerialPort

with SerialPort.open("/dev/ttyUSB0", baudrate=115200) as port:
    port.send(b"ping\n")
    reply = port.receive(1024, timeout=1.0)

Every method that blocks on I/O accepts an optional timeout= keyword; snapshot-style calls (input_waiting, output_waiting, property accesses) do not dispatch through the portal and return immediately.

How it works¶

Every sync port in the process shares one anyio.from_thread.BlockingPortalProvider, which owns a single background thread running an AnyIO event loop. The provider is refcounted — the first SerialPort.open(...) spawns the event-loop thread; the last close() tears it down. Opening multiple sync ports does not spawn multiple event-loop threads.

Each sync method forwards to its async counterpart via portal.call(coroutine, *args); optional timeouts wrap the coroutine in anyio.fail_after.

       ┌──────────────────── caller thread(s) ─────────────────────┐
       │  port.send(b"x", timeout=1.0)                             │
       └────────────────┬───────────────────────┬──────────────────┘
                        │ portal.call           │ portal.call
                        ▼                       ▼
       ┌────────── portal event-loop thread (one) ─────────────────┐
       │  async_port.send(b"x")  ←─ AnyIO coroutine                │
       │  anyio.wait_writable(fd) / os.write(fd, ...)              │
       └───────────────────────────────────────────────────────────┘

Choosing the AnyIO backend¶

The default is plain asyncio. To use uvloop, winloop, or Trio, call anyserial.sync.configure_portal before opening the first sync port:

from anyserial.sync import configure_portal, SerialPort

configure_portal(backend="asyncio", backend_options={"use_uvloop": True})

with SerialPort.open("/dev/ttyUSB0") as port:
    ...

configure_portal raises RuntimeError if called after the portal thread has started.

Per-call timeouts¶

Timeouts are per-call keyword arguments on every blocking method. TimeoutError (the stdlib one, which AnyIO's fail_after raises directly) surfaces unchanged when they fire; the port stays usable for subsequent calls.

try:
    data = port.receive(64, timeout=0.5)
except TimeoutError:
    ...  # port is still open and healthy

Some methods take both a behavioural-timing parameter and the portal-level timeout — for example send_break(duration=0.25, timeout=1.0): duration is how long the BREAK condition is asserted, timeout bounds the whole call at the portal.

Concurrency¶

The portal's event loop is single-threaded, so calls from multiple OS threads dispatch through one serialization point. Full-duplex send+ receive from different threads works cleanly; concurrent send calls from multiple threads are not guaranteed to serialize — the async port's ResourceGuard can still fire (anyio.BusyResourceError) if two coroutines enter the same guarded section. Coordinate writes from a single thread if you need ordering guarantees.

Context managers and lifecycle¶

SerialPort supports the stdlib with protocol and is idempotently closeable:

port = SerialPort.open("/dev/ttyUSB0")
port.close()
port.close()  # no-op

Leaking an open port triggers a ResourceWarning during garbage collection, matching the async port's contract. Always close explicitly or use with.

SerialConnectable¶

The deferred-open data class mirrors anyserial.SerialConnectable but does not implement anyio.abc.ByteStreamConnectable — that Protocol requires an async connect, which is incompatible with sync call sites. Sync code that needs AnyIO connectable polymorphism should use the async variant.

from anyserial.sync import SerialConnectable

recipe = SerialConnectable(path="/dev/ttyUSB0", config=SerialConfig(baudrate=115200))
with recipe.connect() as port:
    ...

API parity¶

Every non-lifecycle method and property on the async SerialPort has a sync counterpart with a matching signature (minus async, plus an optional timeout= on portal-dispatched methods). The parity is regression-tested in tests/unit/test_sync_parity.py.

The only intentional differences:

When to use which¶

The sync wrapper is a convenience; code that expects to grow into an async application should start with the async API.

Performance expectations¶

Every sync call pays one portal.call(coroutine, *args) hop — the caller's thread submits the coroutine to the event-loop thread and blocks until the result returns. The hop is measured on real hardware at ~470 µs per call on a modern laptop (reference). Two consequences worth internalizing:

• Fine for setup, one-shot scripts, and I/O-bound calls. The hop is dwarfed by any wait on the serial port (milliseconds for USB adapters).
• Visible on tight request/response loops. A 500-iteration poll loop using the sync wrapper takes roughly 250 ms longer than the same loop in pure async code. For that workload, drop into an anyio.run(...) block or structure the caller as async.

Cancellation shows the same shape: async cancellation hits p99 < 1 ms on real USB; the sync wrapper's timeout= lands at p99 ≈ 2.6 ms because the cancellation event itself crosses the portal. Again, fine for most code — just know the number so you don't be surprised.