Notifications

Everything the device pushes rather than answers arrives as a notification: UART bytes, GPIO interrupts, I3C in-band interrupts and hot-joins, the target bus events of all three protocols, and physical plug/unplug on the I3C connector.

Subscribing

Device.subscribe takes the notification class you want, a callback, and an optional where= predicate. It returns a Subscription, which detaches on unsubscribe() (idempotent) or on leaving a with block:

from pycosmicsdk import UartRxNotification

with dev.subscribe(UartRxNotification, lambda n: print(n.data)) as sub:
    ...

AsyncDevice.subscribe is the same, except the callback is a coroutine function: registration stays synchronous, and the coroutine is scheduled onto the running loop from the dispatcher thread. The async API also offers AsyncDevice.iter_notifications, a pull-based async iterator with no blocking equivalent.

Blocking callbacks run on the dispatcher’s own thread. Keep them short: there is one dispatcher thread per device, and every notification and every legacy deliverable is delivered on it in order, so a slow callback delays everything behind it. Hand long work to a queue or a thread of your own.

Calling back into the device from inside a callback is explicitly safe, and is not something to design around. The fan-out loop releases its subscription lock before invoking any user code, so a callback may issue a device operation, or unsubscribe() itself, without deadlocking. What it must not do is block for long, for the ordering reason above — and a callback that waits on something only another notification can deliver waits forever, because that notification is queued behind it.

Warning

where= predicates run on the dispatcher thread, where an exception is logged and swallowed — a raising predicate silently drops the event. Compare raw notification fields against enum members directly; never convert with SomeEnum(raw_value) inside a predicate. See the GPIO guide for the concrete trap.

Exceptions from your callbacks are logged on the pycosmicsdk.dispatcher logger; exceptions caught at the C++ to Python bridge go to pycosmicsdk.backend. Neither propagates into C++.

Raw field values

Notification fields that come from a firmware-populated struct are plain int, not enums — no enum survives the crossing. That is deliberate: a firmware value this release does not name still arrives instead of raising. The enums those fields are meant to be compared against are arithmetic (IntEnum/IntFlag) precisely so the comparison works as written; see Enums for which ones are and are not.

The rule holds without exception on this page’s subject: I3cTargetBusEventNotification.kind is a plain int and is never upgraded, so a subscribe callback that writes n.kind is I3cTargetBusEventKind.WRITE_TRANSFER is comparing an int against a plain enum.Enum member and gets False every time. Convert first, in the callback rather than in a where= predicate:

from pycosmicsdk import I3cTargetBusEventKind

def on_bus_event(n) -> None:
    try:
        kind = I3cTargetBusEventKind(n.kind)
    except ValueError:
        return  # a kind this release does not name
    if kind is I3cTargetBusEventKind.WRITE_TRANSFER:
        ...

Important

Do not confuse it with I3cTargetBusEvent, which is a different type reached a different way. That one is the dataclass I3cTarget.wait_for_event returns, and its kind is upgraded where possible — an I3cTargetBusEventKind member when the SDK knows the event kind, a plain int otherwise — because the upgrade is this SDK’s own re-wrap on the wait_for_event path, not something that survives the C++ crossing. Same words, two types, opposite handling. The I3C guide has the side-by-side.

The three target bus events are not the same shape

One per protocol, and the differences are load-bearing rather than cosmetic.

	`I2cTargetBusEventNotification`	`SpiTargetBusEventNotification`	`I3cTargetBusEventNotification`
Kind field	`event`	`event`	`kind`
Second field	`target_address` — the adapter’s own configured address	`chip_select` — which CS line framed it	`target_address` — the adapter’s own dynamic address
`memory_address`	real — the arena byte offset the phase started at	always `0`	register index
`payload_length`	bytes moved	bytes clocked	(absent — read `payload`)
`payload`	the whole raw frame (inherited)	the whole raw frame (inherited)	the transferred bytes (shadows the base)
Emitted	one per transfer phase, all flushed at the STOP — one per completed transaction, since an ordinary transaction is one phase; a `non_stop` transfer emits none until something issues the STOP	only on chip-select de-assert, and only when at least one byte moved — one per transaction	on transfer complete, CCC handled, and dynamic-address change

Warning

On the I2C and SPI target events, payload is the whole raw notification frame — not the transferred bytes, and not empty either. Neither class defines a payload of its own, so the name resolves to the base Notification’s, which the SDK fills from the entire packet. The I3C event is the one of the three that shadows the base with a property returning the bytes that moved on the bus, because it is the only one firmware appends data to.

The trap is that there are two unrelated length fields, both spelled payloadLength upstream, and only one of them survives into the typed record. The one that sizes the raw frame stays behind; payload_length on the event is the count of bytes that moved on the bus. So for a well-formed I2C or SPI event len(payload) is a constant twelve — header plus parameters — while payload_length is whatever the transaction moved. Reading payload expecting the data gets you a frame whose length happens to look plausible. That constant twelve is now a bench measurement rather than a reading of the source: on the I2C target it held for every event driven while payload_length ranged from 0 to 1024.

So on I2C payload_length is the field to trust for “how many bytes did that transaction move”, and it stays meaningful after the bytes themselves are gone. To get the bytes, read them out of the target’s arena — accepting that the arena is live, so nothing stops the external controller overwriting that region between the notification and your read. On I2C the event’s memory_address tells you where to read; on SPI it does not, for the reason below.

On SPI it is not a data length at all. The table above says “bytes clocked” for a reason: the SPI count covers the whole framed transaction, command and address phases included, so a FLASH read of N bytes reports N + 4 at the default three-byte address width. The overhead varies with the model and the opcode and the event carries neither, so nothing — here or in this SDK — can recover N from it. SpiTargetBusEvent.frame_length is the same number under an honest name and is the one to read on the typed record. See Target mode.

Caution

A malformed target bus-event frame is not a catchable error — it is a subscriber that silently never fires. The typed accessor for each of these events returns nothing when the frame is shorter than its twelve-byte header, and the notification builder then falls through to UnknownNotification without raising. So a truncated frame is still delivered, but to a different subscriber than the one you registered: code waiting on I2cTargetBusEventNotification sees nothing at all, and there is no exception anywhere to catch. If target-mode events stop arriving, subscribe to UnknownNotification alongside and log n.payload.hex() — that is the only way such a frame becomes visible.

Warning

The SPI event’s memory_address is always 0. It is documented upstream as the arena offset the transaction started at, but both firmware emit sites pass zero regardless of where the transaction landed — the FLASH and REGISTER_MAP models track their address internally and never report it. The field is carried rather than dropped so the record matches the I2C one and so it starts working on its own if firmware ever populates it. Until then, where a transaction landed is something you know from the protocol you are emulating.

Two more asymmetries worth naming. Only the I3C event’s firmware status is computed — the I2C and SPI ones hard-code it to success in the shared notification header — and this SDK surfaces no status field on any of the three, so a bus event is not the place to look for a transfer-level failure. And I3C’s target_address is always the adapter’s own dynamic address for every event kind, not the controller’s: an I3C target has no way to learn who addressed it.

The kind field follows the raw-int rule above on every one of the three. SpiTargetBusEventKind has exactly one member today, which makes the raw-int arm matter more there than anywhere else: it is the only thing that will make a future second kind visible instead of dropping the event.

All three targets also offer a pull-based alternative, and each hands back a re-wrapped record rather than the notification: I2cTarget.wait_for_event → I2cTargetBusEvent, SpiTarget.wait_for_event → SpiTargetBusEvent, and I3cTarget.wait_for_event → I3cTargetBusEvent. On every one of the three the record’s kind is upgraded to an enum member where the SDK knows the value — as is chip_select on the SPI record — which is the same “same words, two types, opposite handling” split as the I3C pair above. Async twins exist for all three.

When an event arrives as `UnknownNotification`

UnknownNotification is the fallback the C++ notification builder returns when it cannot give a frame a type. It is not an error, and nothing is dropped to produce one — it is the same “surface it, do not drop it” policy as the raw int fields above, applied one level up: to the whole frame rather than to a field.

Two things produce one:

The frame’s type is one this release does not name. A firmware newer than your wheel pushed an event that did not exist when the SDK was built.
The type is one it names, but the typed payload was absent. The builder asks upstream for the matching struct and gets a null pointer back, so there is nothing to copy into a typed object.

It carries nothing but the base class’s raw bytes: Notification.payload, exactly as received. There are no decoded fields, because deciding what the fields are is the step that did not happen.

Subscriber matching is isinstance, which settles how you see one:

from pycosmicsdk import Notification, UnknownNotification

# Only the frames the SDK could not type.
sub = dev.subscribe(UnknownNotification, lambda n: print(n.payload.hex(" ")))

# Every notification, typed or not -- UnknownNotification included.
everything = dev.subscribe(Notification, lambda n: print(type(n).__name__))

There is no handling to write for one, only reporting: log n.payload.hex() together with dev.info.fw_version and dev.get_version(). An event you were expecting that arrives this way usually means the adapter’s firmware is ahead of this SDK, in which case the fix is a package upgrade rather than anything in your code.

unknown_notification synthesises one with a payload you choose, so the branch is testable without a device or an out-of-date adapter — see Testing without hardware.

Reference

class pycosmicsdk.Subscription(*, notification_type: type[N], callback: Callable[[N], None], where: Callable[[N], bool] | None, on_unsubscribe: Callable[[Subscription[Any]], None] | None)

Bases: Generic[N]

Type-narrowed handle returned by Device.subscribe / AsyncDevice.subscribe.

Use unsubscribe() to detach, or scope the subscription to a with block using context-manager semantics. unsubscribe() is idempotent.

Notes

Constructor is package-private: user code receives Subscription instances only from Device.subscribe / AsyncDevice.subscribe, never by direct instantiation. Internally, callers go through _make_subscription.

unsubscribe() → None

Detach from the dispatcher. Idempotent.

The first call flips the internal active flag and invokes the registered on_unsubscribe hook (if any) exactly once. Subsequent calls are no-ops.

property event_filter: Callable[[Notification], bool]

Read-only combined predicate (isinstance + optional where).

Returns True for notifications that should be delivered to callback. The dispatcher’s fan-out loop calls this once per notification per subscription.

property callback: Callable[[N], None]: Read-only user callback to be invoked on matching notifications.

property active: bool: True until unsubscribe() runs (or the context manager exits).

class pycosmicsdk.Notification(*args, **kwargs)

Bases: object

property payload: (self) -> bytes

class pycosmicsdk.UartRxNotification(*args, **kwargs)

Bases: Notification

property data: (self) -> bytes

class pycosmicsdk.GpioInterruptNotification(*args, **kwargs)

Bases: Notification

property logic_level: (self) -> int

property pin_number: (self) -> int

property triggered_edge: (self) -> int

class pycosmicsdk.I3cIbiNotification(*args, **kwargs)

Bases: Notification

property address: (self) -> int

property payload: (self) -> bytes

class pycosmicsdk.I3cHotJoinNotification(*args, **kwargs)

Bases: Notification

property bcr: (self) -> int

property dcr: (self) -> int

property dynamic_address: (self) -> int

property pid: (self) -> bytes

class pycosmicsdk.I2cTargetBusEventNotification(*args, **kwargs)

Bases: Notification

property event: (self) -> int

property memory_address: (self) -> int

property payload_length: (self) -> int

property target_address: (self) -> int

class pycosmicsdk.SpiTargetBusEventNotification(*args, **kwargs)

Bases: Notification

property chip_select: (self) -> int

property event: (self) -> int

property memory_address: (self) -> int

property payload_length: (self) -> int

class pycosmicsdk.I3cTargetBusEventNotification(*args, **kwargs)

Bases: Notification

property kind: (self) -> int

property memory_address: (self) -> int

property payload: (self) -> bytes

property target_address: (self) -> int

class pycosmicsdk.I3cConnectorEventNotification(*args, **kwargs)

Bases: Notification

property connector_type: (self) -> int

property event: (self) -> int

property port: (self) -> int

class pycosmicsdk.UnknownNotification(*args, **kwargs): Bases: Notification

Notifications

Subscribing

Raw field values

The three target bus events are not the same shape

When an event arrives as UnknownNotification

Reference

When an event arrives as `UnknownNotification`