Skip to main content

Python

Subscribe to a stream, publish to one, and read what a stream announced about itself.

from ranvier import schema_id_of, subscribe

with subscribe("127.0.0.1", 7000, "pointer") as pointer:
    print(schema_id_of(pointer.announcement()))
    for sample in pointer:
        handle(sample.values)

sample.values is bytes this package never opens. What they mean is the schema the stream announced, and decoding them belongs to whatever compiled that schema — np.frombuffer(sample.values, "<f4") if the payload is a packed block, a generated protobuf class if it is a message.

Install​

pip install ranvier

There is no runtime dependency. Wheels are built against CPython's stable ABI from 3.11, so one wheel per platform covers every Python version from 3.11 onward rather than needing one per version.

What you can do​

Read a stream another process publishes​

from ranvier import subscribe

pointer = subscribe("127.0.0.1", 7000, "pointer", connect_timeout=5.0)

subscribe dials immediately and raises OSError if nothing is listening — a subscription that will never yield anything is worth being told about rather than discovering later. Constructing Subscription(host, port, name) directly does the opposite: it dials on the first next, retrying until your timeout, which is what you want when a publisher and a subscriber start in no particular order.

Timeouts are in seconds. next(timeout=0.3) gives up after 300 milliseconds and answers None; next() waits indefinitely. None is an answer rather than an error, and it means one of three things told apart by what you asked for: the wait ran out, the subscription was closed, or the publisher is not there and reconnect is off.

Ctrl-C works. Every wait is a quarter-second poll with the interpreter released, so an interrupt is felt within 250 milliseconds and other Python threads keep running while you block.

For a frame loop or a GUI callback that cannot block at all, try_next() answers whatever is already waiting and None otherwise, without waiting.

Ask what the stream said about itself​

info = pointer.announcement()
if info is not None and info.has_schema:
    identifier = info.schema.schema_id      # "ranvier.example.v1.PointerSample"
    descriptor = info.schema.descriptor     # a serialized FileDescriptorSet

announcement() is None until a publisher has spoken, which is a real state rather than an error. Several streams may publish under one name, so announcements() gives all of them and announcement(stream_ref) asks about one — and sample.stream_ref is the only thing that says which stream a sample came from.

An announcement also carries who was operating and on what authority, when the publisher stated them: info.operator.reference, info.operator.asserted_by, info.authorisation.reference and its validity window. has_operator and has_authorisation separate unstated from empty, which matters because Ranvier records the identity it was given and never verifies it.

Count what went missing​

counts = pointer.counts()
print(counts.received, counts.missing, counts.breaks, counts.dropped)

missing is what the sequence numbers say never arrived. breaks is how many separate runs those were, and it is the number that separates two very different problems sharing one total: a hundred consecutive losses is a disconnection, a hundred scattered singles is a subscriber marginally too slow. dropped is what this consumer's own queue discarded because it was full, and undecodable above zero means a publisher and this subscriber disagree about the contract.

Publish​

from ranvier import Origin, Runtime

runtime = Runtime()
stream = runtime.node("analysis").declare_stream(
    "band-power",
    schema_id="lab.v1.BandPower",
    descriptor=descriptor_set_bytes,
    origin=Origin.DERIVED,
    source_clock_domain="device.eeg",
)

serving = runtime.serve("127.0.0.1:0")
print(f"listening on {serving.address}")

stream.publish(encoded, source_timestamp_ns=instrument_time)

Announce a descriptor if you have one. It is what lets something never compiled against your schema still read the stream — a viewer, a converter, a tool written years from now.

source_timestamp_ns=None is an answer. A synthetic event, a device with no clock, or a handshake that has not produced one are all honestly undated, and saying so is different from claiming a time of zero.

Declare source_clock_domain if your samples carry instrument times. A stream that named no clock cannot carry one, and a timestamp handed to such a stream is dropped rather than sent — a time on a clock nobody named is a number nothing can compare against. It is dropped quietly, which is the sharpest edge on this surface, so stream.source_timestamp_is_carried exists to be asked rather than discovered.

A port of 0 lets the operating system choose, and serving.address reports what it chose in the exact form subscribe takes.

Publish and subscribe in one process​

runtime = Runtime()
node = runtime.node("analysis")
stream = node.declare_stream("band-power")
local = node.subscribe("band-power")

No socket is involved, and a subscriber cannot tell the difference from a remote one.

What this package does not do​

No asyncio. Every call blocks the calling thread, and the omission is deliberate rather than pending. A native binding cannot make a socket read into an awaitable without either running the wait on a thread pool — which is what asyncio.to_thread already is, in your code where you can see it — or reimplementing the transport against an event loop, which would be a second implementation of the thing this package exists to have only one of. If you want a coroutine:

sample = await asyncio.to_thread(pointer.next, 1.0)

No redialling a connection that dropped. reconnect and reconnect_delay cover a publisher that is not up yet, which is the ordinary case at startup: the first next keeps dialling until your timeout. They cannot cover a connection that was established and then broke, because nothing in the C ABI reports that it broke — a link exposes counters that stop moving, which a quiet publisher also does. Redialling on a stalled counter was considered and rejected: a merely slow stream would be redialled underneath you, and samples lost across the swap would be counted nowhere. A silent loss is worse than an absent feature.

No hard real-time. Delivery goes through a bounded queue with a background receive thread. If you need microsecond-class latency guarantees, this is not the shape of tool that provides them.

No decoding. By design. See the top of this file.

Types​

The package ships py.typed and a complete stub, so mypy --strict sees real signatures rather than Any.

Migrating from the package that spoke the wire in Python​

The subscribing surface is unchanged: subscribe, Subscription and its methods, Sample, Counts, StreamInfo and schema_id_of all keep their names and signatures. Four things moved.

counts() answers eight numbers rather than four. The four you had keep their names. Read dropped first: a slow consumer used to apply backpressure through the kernel receive buffer, and its loss surfaced as missing, attributed to the publisher. It now surfaces as dropped, attributed to you. Monitoring written against missing alone will read a slow consumer as a quiet one.

missing is credited back when a late sample turns up, where the old counter never credited back. Ranvier's transports do not reorder — TCP does not, and in-process delivery is ordered under one lock — so the two definitions produce the same number today, and reordered says when that stops being true.

The framing primitives are gone. FrameReader, write_frame, MAX_PAYLOAD, PREFIX and FramingError are not here. The wire belongs to the ABI now, and a second copy of it in Python is exactly what this package was rewritten to remove. Nothing raises FramingError any more: a frame that arrives whole and cannot be read is counted in counts().undecodable, and a stream that misaligns closes its connection.

announcement() answers this package's own StreamInfo, not a generated protobuf message. The field paths are the same — info.name, info.schema.schema_id, info.schema.descriptor, info.host_id — so most code does not notice, and the protobuf runtime is not a dependency of this package. If you were calling protobuf methods on the result (SerializeToString, CopyFrom, HasField) those are gone; has_schema and friends replace HasField.

What it is underneath​

A PyO3 binding over Ranvier's C ABI, which is the one boundary every non-Rust binding sits on. The framing, the protobuf decoding, the sequence accounting and the loss policy are the same code the Rust runtime uses and the same code the other language bindings use.

That is the whole argument for the rewrite. The package this replaces was 740 lines of Python implementing a contract that Rust also implemented, and two implementations of one contract disagree eventually — which is how two hand-written runtimes came to declare four of six refusal reasons with nothing positioned to notice. This binding declares none of them: it loops the ABI's own enumeration functions at import time, so list(RefuseReason) cannot be a subset of what the contract says.