Workers

workers.WorkerTransport

The async surface a sidecar exposes: a wire method call, plus shutdown.

Usage

Source

workers.WorkerTransport()

workers.WorkerSpec

How to spawn a sidecar: the command, an environment overlay, and a working directory.

Usage

Source

workers.WorkerSpec(command, env=(), cwd=None)

Parameter Attributes

command: tuple[str, …]
env: tuple[tuple[str, str], …] = ()
cwd: Path | None = None

Example

>>> WorkerSpec(("uvx", "--python", "3.13", "athome-ocr-paddle"))

workers.PipeWorker

A lazily-spawned sidecar subprocess speaking length-prefixed wire frames over stdio.

Usage

Source

workers.PipeWorker(
    spec,
    lock=Lock(),
    process=None,
    stdout=None,
    fingerprint=dict(),
    stderr_tail=(lambda: deque(maxlen=STDERR_TAIL_CHUNKS))(),
    stderr_thread=None
)

The child is spawned on the first call(); its first frame is a handshake carrying the wire version and a fingerprint. A wire-version mismatch raises HandshakeMismatch before that first call returns. Round-trips are serialized by one lock — a single stdin/stdout pipe pair carries one request/reply at a time, and interleaved writes deadlock the pipe. The child’s stderr is teed to the parent’s stderr and its tail retained for crash diagnostics.

Parameter Attributes

spec: WorkerSpec
lock: Lock = Lock()
process: Process | None = None
stdout: BufferedByteReceiveStream | None = None
fingerprint: dict[str, Wire] = dict()
stderr_tail: deque[bytes] = (lambda: deque(maxlen=STDERR_TAIL_CHUNKS))()
stderr_thread: threading.Thread | None = None

Example

>>> worker = PipeWorker(WorkerSpec(("uvx", "athome-ocr-paddle")))
>>> await worker.call("read", image)

workers.WorkerPool

A fixed pool of PipeWorker sidecars with digest-keyed prefetch and lease affinity.

Usage

Source

workers.WorkerPool(spec, *, size)

Each lease() hands out one worker for exclusive use. A lease keyed to a string prefers the worker that prefetch()-ed that key while it is free, falling back to any free worker. prefetch() warms a free worker with one call and records the affinity, so the follow-up keyed lease lands on the already-warm process.

Example

>>> pool = WorkerPool(WorkerSpec(("uvx", "athome-ocr-paddle")), size=4)
>>> await pool.prefetch(digest, "read", image)
>>> async with pool.lease(digest) as worker:
...     await worker.call("read", image)

workers.serve()

Run a sidecar’s frame loop until the parent closes the pipe.

Usage

Source

workers.serve(handler)

Emits the handshake frame (wire version plus handler.fingerprint() when defined), then reads request frames and dispatches each to the named handler method, replying {"ok": ...} on success or {"err": <traceback>} on any exception. Returns when stdin reaches EOF.

Example

>>> class Echo:
...     def echo(self, payload): return payload
>>> serve(Echo())

workers.WorkerError

A sidecar reported an error through the wire, or its process failed.

Usage

Source

workers.WorkerError()

workers.WorkerCrashed

A sidecar process exited before replying; carries its return code and stderr tail.

Usage

Source

workers.WorkerCrashed(returncode, stderr_tail)

workers.HandshakeMismatch

A sidecar announced a wire version the parent does not speak.

Usage

Source

workers.HandshakeMismatch(expected, got)

wire.validate()

Structurally verify that obj is a Wire value, normalized to exact wire types.

Usage

Source

wire.validate(obj)

Walks containers recursively, coercing every primitive to its exact builtin so the types the wire accepts are the types a frame can carry: a primitive subclass — ocrmac’s objc.pyobjc_unicode, a NumPy scalar, a str enum — pickles as a global that RestrictedUnpickler then refuses, so a sender that skipped the coercion would encode a frame no receiver can load. Dict keys must be str. Raises WireError on any other type (a set, a custom object, a non-str mapping key).

wire.encode()

Validate obj and serialize it into a length-prefixed wire frame.

Usage

Source

wire.encode(obj)

The frame is a 4-byte big-endian body length followed by a pickle (protocol 5) of the validated value.

wire.decode()

Decode a length-prefixed wire frame produced by encode() back into a Wire.

Usage

Source

wire.decode(frame)

Raises WireError if the declared length disagrees with the body or the payload is not a valid Wire.

wire.WireError

A value is not a valid Wire, or a frame cannot be decoded.

Usage

Source

wire.WireError()

Rooted at Exception, deliberately not at athome.errors.AthomeError: wire must import with only the standard library so sidecar distributions that do not depend on athome can vendor it.