NexusBLESdk API¶

Shared Package¶

Client Module¶

class NexusBLESdk.client.GatewayClient(ser, *, client_name='nexus_ble_sdk', verbose=True)[source]¶

Bases: object

Parameters:

client_name (str)
verbose (bool)

request_id(prefix)[source]¶

Parameters:: prefix (str)
Return type:: str

send(obj)[source]¶

Parameters:: obj (dict[str, Any])

hello(protocol_version=1)[source]¶

Parameters:: protocol_version (int)

reset_session(timeout_s=5.0)[source]¶

Parameters:: timeout_s (float)

scan(timeout_ms, *, name_filter=None)[source]¶

Parameters:

timeout_ms (int)
name_filter (str | None)

Return type:

list[DiscoveredDevice]

connect(addresses, timeout_s)[source]¶

Parameters:

addresses (list[str])
timeout_s (float)

Return type:

list[SensorConnection]

subscribe(address, characteristic_uuid, timeout_s, *, binary_notifications=False)[source]¶

Parameters:

address (str)
characteristic_uuid (str)
timeout_s (float)
binary_notifications (bool)

subscribe_with_retry(address, characteristic_uuid, timeout_s, *, binary_notifications=False, attempts=2, retry_delay_s=0.3)[source]¶

Parameters:

address (str)
characteristic_uuid (str)
timeout_s (float)
binary_notifications (bool)
attempts (int)
retry_delay_s (float)

write_gatt(address, characteristic_uuid, payload_hex, timeout_s, *, without_response=False, allow_timeout=False)[source]¶

Parameters:

address (str)
characteristic_uuid (str)
payload_hex (str)
timeout_s (float)
without_response (bool)
allow_timeout (bool)

Return type:

float | None

write_gatt_nowait(address, characteristic_uuid, payload_hex, *, without_response=False)[source]¶

Parameters:

address (str)
characteristic_uuid (str)
payload_hex (str)
without_response (bool)

Return type:

float

read_gatt(address, characteristic_uuid, timeout_s)[source]¶

Parameters:

address (str)
characteristic_uuid (str)
timeout_s (float)

Return type:

bytes

disconnect(addresses, timeout_s, *, allow_timeout=False)[source]¶

Parameters:

addresses (list[str])
timeout_s (float)
allow_timeout (bool)

Return type:

list[str]

get_status_snapshot(timeout_s=10.0)[source]¶

Parameters:: timeout_s (float)
Return type:: dict[str, Any]

wait_for_request(request_id, success_type, timeout_s)[source]¶

Parameters:

request_id (str)
success_type (str)
timeout_s (float)

read_item(timeout_s=10.0)[source]¶

Parameters:: timeout_s (float)

read_json(timeout_s=10.0)[source]¶

Parameters:: timeout_s (float)
Return type:: dict[str, Any]

is_disconnected(address)[source]¶

Parameters:: address (str)
Return type:: bool

assert_connected(address, *, action)[source]¶

Parameters:

address (str)
action (str)

Monitoring Module¶

class NexusBLESdk.monitoring.StartupGateConfig(enabled: 'bool' = True, stability_window_seconds: 'float' = 5.0, packets_required: 'int' = 60, min_rate_hz: 'float' = 58.0, min_observation_seconds: 'float' = 2.0, max_gap_events: 'int' = 0, gap_grace_seconds: 'float' = 2.0)[source]¶

Bases: object

Parameters:

enabled (bool)
stability_window_seconds (float)
packets_required (int)
min_rate_hz (float)
min_observation_seconds (float)
max_gap_events (int)
gap_grace_seconds (float)

enabled: bool = True¶

stability_window_seconds: float = 5.0¶

packets_required: int = 60¶

min_rate_hz: float = 58.0¶

min_observation_seconds: float = 2.0¶

max_gap_events: int = 0¶

gap_grace_seconds: float = 2.0¶

class NexusBLESdk.monitoring.SensorStreamStats(address: 'str', sensor_id: 'int | None', label: 'str | None', expected_rate_hz: 'int', stream_start_command_time: 'float | None' = None, first_packet_time: 'float | None' = None, startup_first_sensor_timestamp: 'int | None' = None, startup_last_sensor_timestamp: 'int | None' = None, startup_packets_received: 'int' = 0, startup_gap_events: 'int' = 0, startup_estimated_dropped_packets: 'int' = 0, startup_gap_detection_start_sensor_timestamp: 'int | None' = None, startup_gate_first_sensor_timestamp: 'int | None' = None, startup_gate_last_sensor_timestamp: 'int | None' = None, startup_gate_packets_received: 'int' = 0, startup_gate_gap_events: 'int' = 0, startup_gate_estimated_dropped_packets: 'int' = 0, measurement_first_sensor_timestamp: 'int | None' = None, measurement_last_sensor_timestamp: 'int | None' = None, measurement_packets_received: 'int' = 0, measurement_gap_events: 'int' = 0, measurement_estimated_dropped_packets: 'int' = 0, host_parsed_frames: 'int' = 0)[source]¶

Bases: object

Parameters:

address (str)
sensor_id (int | None)
label (str | None)
expected_rate_hz (int)
stream_start_command_time (float | None)
first_packet_time (float | None)
startup_first_sensor_timestamp (int | None)
startup_last_sensor_timestamp (int | None)
startup_packets_received (int)
startup_gap_events (int)
startup_estimated_dropped_packets (int)
startup_gap_detection_start_sensor_timestamp (int | None)
startup_gate_first_sensor_timestamp (int | None)
startup_gate_last_sensor_timestamp (int | None)
startup_gate_packets_received (int)
startup_gate_gap_events (int)
startup_gate_estimated_dropped_packets (int)
measurement_first_sensor_timestamp (int | None)
measurement_last_sensor_timestamp (int | None)
measurement_packets_received (int)
measurement_gap_events (int)
measurement_estimated_dropped_packets (int)
host_parsed_frames (int)

address: str¶

sensor_id: int | None¶

label: str | None¶

expected_rate_hz: int¶

stream_start_command_time: float | None = None¶

first_packet_time: float | None = None¶

startup_first_sensor_timestamp: int | None = None¶

startup_last_sensor_timestamp: int | None = None¶

startup_packets_received: int = 0¶

startup_gap_events: int = 0¶

startup_estimated_dropped_packets: int = 0¶

startup_gap_detection_start_sensor_timestamp: int | None = None¶

startup_gate_first_sensor_timestamp: int | None = None¶

startup_gate_last_sensor_timestamp: int | None = None¶

startup_gate_packets_received: int = 0¶

startup_gate_gap_events: int = 0¶

startup_gate_estimated_dropped_packets: int = 0¶

measurement_first_sensor_timestamp: int | None = None¶

measurement_last_sensor_timestamp: int | None = None¶

measurement_packets_received: int = 0¶

measurement_gap_events: int = 0¶

measurement_estimated_dropped_packets: int = 0¶

host_parsed_frames: int = 0¶

property expected_delta_us: float¶

property startup_duration_seconds: float¶

property startup_observed_rate_hz: float¶

property startup_gate_duration_seconds: float¶

property startup_gate_rate_hz: float¶

property measurement_duration_seconds: float¶

property observed_rate_hz: float¶

property time_to_first_packet_ms: float | None¶

record_sample(timestamp, wall_time, *, measurement_active, startup_gap_grace_seconds)[source]¶

Parameters:

timestamp (int | None)
wall_time (float)
measurement_active (bool)
startup_gap_grace_seconds (float)

reset_measurement()[source]¶

class NexusBLESdk.monitoring.GenericStreamMonitor(*, connections, labels_by_address, expected_rate_hz, timestamp_parser, startup_gate, verbose=True)[source]¶

Bases: object

Parameters:

connections (list[SensorConnection])
labels_by_address (dict[str, str | None])
expected_rate_hz (int)
timestamp_parser (Callable[[bytes], int])
startup_gate (StartupGateConfig)
verbose (bool)

announce_startup_state()[source]¶

mark_stream_started(address, command_time)[source]¶

Parameters:

address (str)
command_time (float | None)

handle_stream_frame(frame, wall_time)[source]¶

Parameters:

frame (StreamFrame)
wall_time (float)

drain_after_stop(gateway_client, *, quiet_window_s=0.35, max_drain_s=2.0)[source]¶

Parameters:

quiet_window_s (float)
max_drain_s (float)

evaluate_startup_stability()[source]¶

Return type:: tuple[bool, list[str]]

activate_measurement()[source]¶

summary_lines(gateway_client)[source]¶

Return type:: list[str]

Models Module¶

class NexusBLESdk.models.StreamFrame(sensor_id: 'int', gateway_timestamp_us: 'int', payload: 'bytes')[source]¶

Bases: object

Parameters:

sensor_id (int)
gateway_timestamp_us (int)
payload (bytes)

sensor_id: int¶

gateway_timestamp_us: int¶

payload: bytes¶

class NexusBLESdk.models.DiscoveredDevice(address: 'str', name: 'str' = '', rssi: 'int | None' = None, service_uuids: 'tuple[str, ...]' = (), raw: 'dict' = <factory>)[source]¶

Bases: object

Parameters:

address (str)
name (str)
rssi (int | None)
service_uuids (tuple[str, ...])
raw (dict)

address: str¶

name: str = ''¶

rssi: int | None = None¶

service_uuids: tuple[str, ...] = ()¶

raw: dict¶

class NexusBLESdk.models.SensorConnection(address: 'str', sensor_id: 'int | None' = None)[source]¶

Bases: object

Parameters:

address (str)
sensor_id (int | None)

address: str¶

sensor_id: int | None = None¶

Transport Module¶

NexusBLESdk.transport.json_objects_from_line(line)[source]¶

Parameters:: line (str)

NexusBLESdk.transport.open_gateway_serial(port='/dev/serial/by-id/usb-SEGGER_J-Link_001057755524-if02', baudrate=1000000)[source]¶

Parameters:

port (str)
baudrate (int)

Return type:

Iterator[Serial]