Documentation

Peripheral Device

A Peripheral Device (PD) is the follower on the bus: it answers a Control Panel's commands and reports events such as card reads and key presses. In Python that role is the PeripheralDevice class. A PD is described by one PDInfo and a set of capabilities that tell the CP what it can do.

Lifecycle ​

  1. Construct — PeripheralDevice(pd_info, pd_cap, ...), optionally passing a command handler and an event-completion handler.
  2. Register handlers (optional) — or set them later with set_command_handler / set_event_completion_handler.
  3. Start — start() spawns the background thread that services the bus and keeps the library refreshed within OSDP's 50ms window.
  4. Wait for the link — sc_wait() blocks until the secure channel is up (or returns once online for an unkeyed link).
  5. Run — answer commands in the handler and report events with submit_event().
  6. Tear down — teardown().
pd = PeripheralDevice(pd_info, pd_cap, command_handler=command_handler)
pd.start()
pd.sc_wait()
...
pd.teardown()
Callbacks run on the library thread

The command handler and event-completion handler are invoked from the background refresh thread. Keep them short — blocking in the command handler delays the reply to the CP — and guard any state shared with your application.

The command-handler contract ​

The command handler is the heart of a PD. LibOSDP calls it for each command the CP sends, and your return value decides the reply:

  • Return None — accept the command (the library ACKs it).
  • Return a command — answer inline. A commands.Status query, for example, is answered by returning a commands.Status carrying the report.
  • Raise NakError — decline the command with a NakCode.
from osdp import Command, NakCode, NakError, StatusReportType, commands

def command_handler(cmd: Command) -> Command | None:
    match cmd:
        case commands.Status(type=StatusReportType.Input):
            return commands.Status(type=StatusReportType.Input, report=read_inputs())
        case commands.LED():
            drive_led(cmd)
        case commands.BioRead():
            raise NakError(NakCode.BioType)   # not supported on this PD
    return None

You can also pull commands off a queue instead of (or alongside) the handler with get_command(timeout=...), which returns the next command or None.

PeripheralDevice ​

classPeripheralDevice

Answers a CP on one address.

Runs a background thread that services the bus; nothing happens until `start()` is called.

Example:

def on_command(cmd: Command) -> Command | None:
    match cmd:
        case commands.Status(type=t):
            return commands.Status(type=t, report=read_inputs())
        case commands.BioRead():
            raise NakError(NakCode.BioType)
    return None

pd = PeripheralDevice(info, caps, command_handler=on_command)
pd.start()
Methods
__init__(self, PDInfo pd_info, PDCapabilities pd_cap, LogLevel log_level=LogLevel.Info, CommandHandler|None command_handler=None, EventCompletionHandler|None event_completion_handler=None)
ctx(self) -> "_sys.PeripheralDevice"

The library context. Raises once this PD has been torn down.

set_command_handler(self, CommandHandler|None handler)

Install the handler called for every command from the CP.

set_event_completion_handler(self, EventCompletionHandler|None handler)

Install the handler called when a submitted event settles.

submit_event(self, Event event) -> bool

Queue an event for the CP. Returns False if it could not be queued.

Submitting from inside the command handler sends the event as an inline reply to that command. Submitting later sends it as a poll response, and the command is acknowledged on its own.

flush_events(self) -> int

Drop every queued event. Returns how many were dropped.

Each one is still reported to the completion handler, as Flushed.

get_command(self, float timeout=5) -> Command|None

Pop the next command from the CP, or None if none arrives in time.

A negative timeout does not block.

is_online(self) -> bool

Whether the CP is polling us.

is_sc_active(self) -> bool

Whether the secure channel is up.

sc_wait(self, float timeout=8) -> bool

Block until the secure channel comes up, or the timeout elapses.

register_file_ops(self, FileOps fops) -> bool

Supply the sink a file transfer from the CP will write into.

get_file_tx_status(self) -> FileTxStatus|None

How far the running transfer has got, or None if none is running.

get_metrics(self) -> Metrics|None

Packet and command counters.

start(self)

Start servicing the bus.

stop(self)

Stop servicing the bus.

teardown(self)

Stop the bus and release the library context.

The PD cannot be used afterwards.

Reporting events ​

A PD reports asynchronous events — card reads, key presses, biometric results — by submitting an event dataclass:

pd.submit_event(events.CardRead(reader_no=0, format=CardFormat.Wiegand, data=card, bits=26))

Each submitted event settles once. Register an event-completion handler to learn its outcome as a CompletionStatus — for instance Flushed if flush_events() dropped it before it reached the wire:

def on_event_complete(event, status):
    if status == CompletionStatus.Flushed:
        print("event removed by flush")

pd.set_event_completion_handler(on_event_complete)

Install mode and secure channel ​

Constructing PDInfo with scbk=None puts the PD in install mode: it will accept a Secure Channel Base Key handed to it by the CP over the initial link. Pass a 16-byte scbk instead to fix the key ahead of time. See Secure Channel for the full keying model.

See also ​