jacdac

Subpackages

Submodules

Package Contents

Classes

JDPacket

A Jacdac packet

Transport

A base class for packet transports

EventEmitter

Bus

A Jacdac bus that managed devices, service client, registers.

InPipe

Incoming pipe

OutPipe

Out going packet pipe

RawRegisterClient

A Jacdac register client

Server

SensorServer

ControlServer

A server for the control service, used internally by the bus.

LoggerServer

UniqueBrainServer

A server for the unique brain service, used internally by the bus

DeviceWrapper

RoleBinding

ServerBindings

RoleManagerServer

A server for the role manager service

Client

Base class to define service clients.

SensorClient

A client for a sensor service

Device

A device on the bus

BufferClient

Base class to define service clients.

JDPacket

A Jacdac packet

Functions

jdpack(→ bytes)

jdunpack(→ PackTuple)

now()

short_id(longid)

jdpack(→ bytes)

jdunpack(→ PackTuple)

_service_matches(dev, serv)

rand_u64()

is_raspberrypi()

Attributes

PackType

PackTuple

PackType

EV_CHANGE

EV_DEVICE_CONNECT

EV_DEVICE_CHANGE

EV_DEVICE_ANNOUNCE

EV_SELF_ANNOUNCE

EV_PACKET_PROCESS

EV_REPORT_RECEIVE

EV_REPORT_UPDATE

EV_RESTART

EV_PACKET_RECEIVE

EV_EVENT

EV_STATUS_EVENT

EV_IDENTIFY

EV_CONNECTED

EV_DISCONNECTED

RegType

HandlerFn

EventHandlerFn

UnsubscribeFn

jacdac.PackType
jacdac.jdpack(fmt: str, *args: PackType) → bytes
jacdac.jdunpack(buf: bytes, fmt: str) → PackTuple
class jacdac.JDPacket(*, cmd: int = None, size: int = 0, frombytes: bytes = None, data: bytes = None, sender: Any = None)

A Jacdac packet

static packed(cmd: int, fmt: str, *args: jacdac.pack.PackType)
unpack(fmt: str)
property service_command
property device_id → str
property packet_flags
property multicommand_class
property size
property requires_ack
property service_index
property is_regular_service
property crc
property is_event
property event_code
property event_counter
property is_reg_set
property is_reg_get
property reg_code
property data
property is_command
property is_report
to_string()
__str__()

Return str(self).

not_implemented()
class jacdac.Transport

A base class for packet transports

on_receive :Optional[Callable[[bytes], None]]
send(pkt: bytes) → None
jacdac.now()
jacdac.short_id(longid: Union[bytes, str])
jacdac.PackTuple
jacdac.PackType
jacdac.jdpack(fmt: str, *args: PackType) → bytes
jacdac.jdunpack(buf: bytes, fmt: str) → PackTuple
jacdac.EV_CHANGE = change
jacdac.EV_DEVICE_CONNECT = deviceConnect
jacdac.EV_DEVICE_CHANGE = deviceChange
jacdac.EV_DEVICE_ANNOUNCE = deviceAnnounce
jacdac.EV_SELF_ANNOUNCE = selfAnnounce
jacdac.EV_PACKET_PROCESS = packetProcess
jacdac.EV_REPORT_RECEIVE = reportReceive
jacdac.EV_REPORT_UPDATE = reportUpdate
jacdac.EV_RESTART = restart
jacdac.EV_PACKET_RECEIVE = packetReceive
jacdac.EV_EVENT = packetEvent
jacdac.EV_STATUS_EVENT = statusEvent
jacdac.EV_IDENTIFY = identify
jacdac.EV_CONNECTED = connected
jacdac.EV_DISCONNECTED = disconnected
jacdac.RegType
jacdac.HandlerFn
jacdac.EventHandlerFn
jacdac.UnsubscribeFn
class jacdac.EventEmitter(bus: Bus)
emit(id: str, *args: object)
_init_emitter()
on(id: str, fn: HandlerFn) → UnsubscribeFn

Subscribes an event to a handler. Returns a callback to unsubscribe.

Args:

id (str): event identifier fn (HandlerFn): event callback

Returns: callback to unsubscribe

once(id: str, fn: HandlerFn)

Subscribes an event to run once; then get unsubscribed

Args:

id (str): event identifier fn (HandlerFn): event callback

off(id: str, fn: HandlerFn)

Unsubscribes a handler from an event

Args:

id (str): event identifier fn (HandlerFn): event callback

async event(id: str)
wait_for(id: str)
_log_report_prefix() → str
_add_log_report(priority: int, text: str, *args: object)
info(text: str, *args: object)
warn(text: str, *args: object)
debug(text: str, *args: object)
error(text: str, *args: object)
jacdac._service_matches(dev: Device, serv: bytearray)
jacdac.rand_u64()
jacdac.is_raspberrypi()
class jacdac.Bus(*, transports: jacdac.packet.Optional[List[jacdac.transport.Transport]] = None, device_id: jacdac.packet.Optional[str] = None, product_identifier: jacdac.packet.Optional[int] = None, firmware_version: jacdac.packet.Optional[str] = None, device_description: jacdac.packet.Optional[str] = None, disable_logger: bool = False, disable_role_manager: bool = False, disable_settings: bool = False, disable_brain: bool = False, disable_dev_tools: bool = False, spi: jacdac.packet.Optional[bool] = None, hf2_portname: jacdac.packet.Optional[str] = None, transport_cmd: jacdac.packet.Optional[str] = None, default_logger_min_priority: jacdac.packet.Optional[int] = None, storage_dir: jacdac.packet.Optional[str] = None, logger_name: jacdac.packet.Optional[str] = None)

Bases: EventEmitter

A Jacdac bus that managed devices, service client, registers.

run(cb: Callable[Ellipsis, None], *args: jacdac.packet.Any)
_sender()
_process_task()
process_frame(sender: jacdac.transport.Transport, frame: bytes)
force_jd_thread()
force_non_jd_thread()
debug_dump()
lookup_server(service_class: int) → jacdac.packet.Optional[Server]
_gc_devices()
_queue_core(pkt: jacdac.packet.JDPacket)
_send_core(pkt: jacdac.packet.JDPacket)
clear_attach_cache()
mk_event_cmd(ev_code: int)
_reattach(dev: Device)
process_packet(pkt: jacdac.packet.JDPacket)
add_pipe(pipe: InPipe)
remove_pipe(pipe: InPipe)
class jacdac.InPipe(bus: Bus)

Bases: EventEmitter

Incoming pipe

open_command(cmd: int)
bytes_available() → int
read() → jacdac.packet.Optional[bytearray]
_close()
close()
meta(pkt: jacdac.packet.JDPacket)
handle_packet(pkt: jacdac.packet.JDPacket)
read_list() → List[bytearray]
class jacdac.OutPipe(bus: Bus, pkt: jacdac.packet.JDPacket)

Bases: EventEmitter

Out going packet pipe

property open
write_ex(buf: bytearray, flags: int)
write(buf: bytearray)
write_and_close(buf: bytearray)
close()
write_meta(buf: bytearray)
class jacdac.RawRegisterClient(client: Client, code: int, pack_format: jacdac.packet.Optional[str])

Bases: EventEmitter

A Jacdac register client

clear()
current(refresh_ms: int = -1)
values() → jacdac.packet.Optional[jacdac.pack.PackTuple]
set_values(*args: jacdac.pack.PackType)
value(default_value: jacdac.packet.Any = None) → jacdac.packet.Optional[jacdac.packet.Any]

Extracts the value of the first field.

bool_value(default_value: jacdac.packet.Optional[bool] = None) → jacdac.packet.Optional[bool]

Extracts the value of the first field as a boolean.

float_value(default_value: jacdac.packet.Optional[float] = None, scale: int = 1) → jacdac.packet.Optional[float]
_query()
refresh()
query(refresh_ms: int = 500)
async query_async(refresh_ms: int = 500)
query_no_wait(refresh_ms: int = -1)
handle_packet(pkt: jacdac.packet.JDPacket)
class jacdac.Server(bus: Bus, service_class: int, *, instance_name: str = None)

Bases: EventEmitter

status_code()
set_status_code(code: int, vendor_code: int)
handle_packet_outer(pkt: jacdac.packet.JDPacket)
handle_packet(pkt: jacdac.packet.JDPacket)
send_report(pkt: jacdac.packet.JDPacket)
send_event(event_code: int, data: bytes = None)
send_change_event()
handle_status_code(pkt: jacdac.packet.JDPacket)
handle_reg_u8(pkt: jacdac.packet.JDPacket, register: int, current: int)
handle_reg_u32(pkt: jacdac.packet.JDPacket, register: int, current: int)
handle_reg_i32(pkt: jacdac.packet.JDPacket, register: int, current: int)
handle_reg(pkt: jacdac.packet.JDPacket, register: int, fmt: str, current: RegType) → RegType
_handle_instance_name(pkt: jacdac.packet.JDPacket)
_log_report_prefix() → str
class jacdac.SensorServer(bus: Bus, service_class: int, streaming_interval: int, *, instance_name: str = None, streaming_preferred_interval: int = None)

Bases: Server

abstract send_reading()
_start_streaming()
async _stream()
handle_packet(pkt: jacdac.packet.JDPacket)
_handle_streaming_samples(pkt: jacdac.packet.JDPacket)
_handle_streaming_interval(pkt: jacdac.packet.JDPacket)
_handle_streaming_preferred_interval(pkt: jacdac.packet.JDPacket)
class jacdac.ControlServer(bus: Bus)

Bases: Server

A server for the control service, used internally by the bus.

queue_announce()
handle_packet(pkt: jacdac.packet.JDPacket)
class jacdac.LoggerServer(bus: Bus)

Bases: Server

handle_packet(pkt: jacdac.packet.JDPacket)
report(priority: int, msg: str, *args: object)
class jacdac.UniqueBrainServer(bus: Bus)

Bases: Server

A server for the unique brain service, used internally by the bus

class jacdac.DeviceWrapper(device: Device)
class jacdac.RoleBinding(role_manager: RoleManagerServer, role: str, service_class: int)
host() → str
select(devwrap: DeviceWrapper, service_idx: int)
class jacdac.ServerBindings(host: str)
property fully_bound → bool
score_for(devwrap: DeviceWrapper, select: jacdac.packet.Optional[bool] = False)

candidate devices are ordered by [numBound, numPossible, device_id] where numBound is number of clients already bound to this device and numPossible is number of clients that can possibly be additionally bound

class jacdac.RoleManagerServer(bus: Bus)

Bases: Server

A server for the role manager service

handle_packet(pkt: jacdac.packet.JDPacket)
handle_list_roles(pkt: jacdac.packet.JDPacket)
handle_all_roles_allocated(pkt: jacdac.packet.JDPacket)
handle_clear_all_roles(pkt: jacdac.packet.JDPacket)
handle_set_role(pkt: jacdac.packet.JDPacket)
set_role(role: str, device: Device, service_idx: int)
is_match_role(role: str, device: Device, service_idx: int)
_binding_hash()
_check_changes()
bind_roles()
class jacdac.Client(bus: Bus, service_class: int, pack_formats: Dict[int, str], role: str)

Bases: EventEmitter

Base class to define service clients.

__str__() → str

Return str(self).

_lookup_register(code: int)
property connected → bool

Indicates if the client is a connected to a server

register(code: int)

Retreives the register by code

_lookup_packformat(code: int) → jacdac.packet.Optional[str]
handle_packet(pkt: jacdac.packet.JDPacket)
handle_packet_outer(pkt: jacdac.packet.JDPacket)
send_cmd(pkt: jacdac.packet.JDPacket)

Sends a command packet to the server

send_cmd_packed(cmd: int, *args: jacdac.pack.PackType)
_attach(dev: Device, service_idx: int)
_detach()
on_connect(handler: EventHandlerFn) → UnsubscribeFn

Registers an event handler when the client connects to a server

Args:

handler (EventHandlerFn): function to run with client connects

Returns:

UnsubscribeFn: function to call to unregister handler

on_disconnect(handler: EventHandlerFn) → UnsubscribeFn

Registers an event handler when the client disconnects from a server

Args:

handler (EventHandlerFn): function to run with client connects

Returns:

UnsubscribeFn: function to call to unregister handler

on_event(code: int, handler: EventHandlerFn) → UnsubscribeFn

Registers an event handler for the given event code

Args:

code (int): event identifier code handler (EventHandlerFn): function to run with decoded event data and packet

Returns:

UnsubscribeFn: function to call to unregister handler

_log_report_prefix() → str
class jacdac.SensorClient(bus: Bus, service_class: int, pack_formats: Dict[int, str], role: str, *, preferred_interval: int = None)

Bases: Client

A client for a sensor service

property streaming_samples → jacdac.packet.Optional[int]

Queries the current estimated streaming samples value

property streaming_interval → jacdac.packet.Optional[int]
property streaming_preferred_interval → jacdac.packet.Optional[int]
refresh_reading() → None
_lookup_packformat(code: int) → jacdac.packet.Optional[str]
_should_refresh_streaming_samples() → bool
_reading_interval() → int

Resolves the best refresh interval for streaming

class jacdac.Device(bus: Bus, device_id: str, services: bytearray)

Bases: EventEmitter

A device on the bus

property ctrl_client
property announce_flags
property reset_count
property packet_count
property is_connected
property short_id
__str__() → str

Return str(self).

debug_info()
service_class_at(idx: int)
matches_role_at(role: str, service_idx: int)
property num_service_classes
_destroy()
_log_report_prefix() → str
process_packet(pkt: jacdac.packet.JDPacket)
class jacdac.BufferClient(bus: Bus, service_class: int, pack_formats: Dict[int, str], role: str)

Bases: Client

Base class to define service clients.

_value :bytearray
_dirty :bool

A client that handles a double-buffer bytes buffer

property value → bytearray

Cached reading value

property dirty → bool
set_dirty() → None
refresh_value() → None
update_value_length(length: jacdac.packet.Optional[int]) → None
class jacdac.JDPacket(*, cmd: int = None, size: int = 0, frombytes: bytes = None, data: bytes = None, sender: Any = None)

A Jacdac packet

static packed(cmd: int, fmt: str, *args: jacdac.pack.PackType)
unpack(fmt: str)
property service_command
property device_id → str
property packet_flags
property multicommand_class
property size
property requires_ack
property service_index
property is_regular_service
property crc
property is_event
property event_code
property event_counter
property is_reg_set
property is_reg_get
property reg_code
property data
property is_command
property is_report
to_string()
__str__()

Return str(self).

not_implemented()