From 504f3c36114cb8fb23af511032c7c4c0ddf24108 Mon Sep 17 00:00:00 2001 From: Jordan Yates Date: Wed, 4 Feb 2026 10:13:14 +1000 Subject: [PATCH 1/2] socket_comms: communication check function Add a function for tools to check whether a gateway is running to handle requests. Signed-off-by: Jordan Yates --- src/infuse_iot/socket_comms.py | 50 +++++++++++++++++++++++++++++++ src/infuse_iot/tools/gateway.py | 9 ++++++ src/infuse_iot/tools/native_bt.py | 6 ++++ 3 files changed, 65 insertions(+) diff --git a/src/infuse_iot/socket_comms.py b/src/infuse_iot/socket_comms.py index 764c91f..59fdd9b 100644 --- a/src/infuse_iot/socket_comms.py +++ b/src/infuse_iot/socket_comms.py @@ -5,6 +5,7 @@ import socket import struct import sys +import time from collections.abc import Generator from contextlib import contextmanager from typing import cast @@ -28,6 +29,7 @@ class Type(enum.IntEnum): CONNECTION_CREATED = 2 CONNECTION_DROPPED = 3 KNOWN_DEVICES = 4 + COMMS_CHECK = 5 def to_json(self) -> dict: """Convert class to json dictionary""" @@ -47,6 +49,8 @@ def from_json(cls, values: dict) -> Self: return cast(Self, ClientNotificationConnectionDropped.from_json(values)) elif values["type"] == cls.Type.KNOWN_DEVICES: return cast(Self, ClientNotificationObservedDevices.from_json(values)) + elif values["type"] == cls.Type.COMMS_CHECK: + return cast(Self, ClientNotificationCommsCheck.from_json(values)) raise NotImplementedError(f"Unknown notification: {values}") @@ -82,6 +86,21 @@ def from_json(cls, values: dict) -> Self: return cls(decoded) +class ClientNotificationCommsCheck(ClientNotification): + TYPE = ClientNotification.Type.COMMS_CHECK + + def __init__(self): + pass + + def to_json(self) -> dict: + """Convert class to json dictionary""" + return {"type": int(self.TYPE)} + + @classmethod + def from_json(cls, _values: dict) -> Self: + return cls() + + class ClientNotificationConnection(ClientNotification): TYPE = 0 @@ -131,6 +150,7 @@ class Type(enum.IntEnum): CONNECTION_REQUEST = 1 CONNECTION_RELEASE = 2 KNOWN_DEVICES = 3 + COMMS_CHECK = 4 def to_json(self) -> dict: """Convert class to json dictionary""" @@ -147,6 +167,8 @@ def from_json(cls, values: dict) -> Self: return cast(Self, GatewayRequestConnectionRelease.from_json(values)) elif values["type"] == cls.Type.KNOWN_DEVICES: return cast(Self, GatewayRequestObservedDevices.from_json(values)) + elif values["type"] == cls.Type.COMMS_CHECK: + return cast(Self, GatewayRequestCommsCheck.from_json(values)) raise NotImplementedError(f"Unknown request: {values}") @@ -182,6 +204,22 @@ def from_json(cls, values: dict) -> Self: return cls() +class GatewayRequestCommsCheck(GatewayRequest): + """Request packet to be forwarded to device""" + + TYPE = GatewayRequest.Type.COMMS_CHECK + + def __init__(self): + pass + + def to_json(self) -> dict: + return {"type": int(self.TYPE)} + + @classmethod + def from_json(cls, _values: dict) -> Self: + return cls() + + class GatewayRequestConnection(GatewayRequest): TYPE = 0 @@ -289,6 +327,18 @@ def receive(self) -> ClientNotification | None: return None return ClientNotification.from_json(json.loads(data.decode("utf-8"))) + def comms_check(self, timeout: float = 0.5) -> bool: + expiry = time.time() + timeout + self.send(GatewayRequestCommsCheck()) + while time.time() < expiry: + rsp = self.receive() + if rsp is None: + continue + if not isinstance(rsp, ClientNotificationCommsCheck): + continue + return True + return False + def connection_create( self, infuse_id: int, data_types: GatewayRequestConnectionRequest.DataType, timeout_ms: int ) -> int: diff --git a/src/infuse_iot/tools/gateway.py b/src/infuse_iot/tools/gateway.py index 084d369..ba1ed58 100644 --- a/src/infuse_iot/tools/gateway.py +++ b/src/infuse_iot/tools/gateway.py @@ -39,11 +39,13 @@ from infuse_iot.serial_comms import PyOcdPort, RttPort, SerialFrame, SerialLike, SerialPort from infuse_iot.socket_comms import ( ClientNotification, + ClientNotificationCommsCheck, ClientNotificationConnectionCreated, ClientNotificationConnectionDropped, ClientNotificationConnectionFailed, ClientNotificationEpacketReceived, ClientNotificationObservedDevices, + GatewayRequestCommsCheck, GatewayRequestConnectionRelease, GatewayRequestConnectionRequest, GatewayRequestEpacketSend, @@ -471,6 +473,11 @@ def _handle_observed_devices(self): observed_devices[device] = info self._common.notification_broadcast(ClientNotificationObservedDevices(observed_devices)) + def _handle_comms_check(self): + if self._common.server is None: + raise RuntimeError + self._common.notification_broadcast(ClientNotificationCommsCheck()) + def _iter(self) -> None: if self._common.server is None: time.sleep(1.0) @@ -486,6 +493,8 @@ def _iter(self) -> None: self._handle_conn_release(req) elif isinstance(req, GatewayRequestObservedDevices): self._handle_observed_devices() + elif isinstance(req, GatewayRequestCommsCheck): + self._handle_comms_check() else: Console.log_error(f"Unhandled request {type(req)}") diff --git a/src/infuse_iot/tools/native_bt.py b/src/infuse_iot/tools/native_bt.py index a12ed19..9253a07 100644 --- a/src/infuse_iot/tools/native_bt.py +++ b/src/infuse_iot/tools/native_bt.py @@ -31,10 +31,12 @@ ) from infuse_iot.socket_comms import ( ClientNotification, + ClientNotificationCommsCheck, ClientNotificationConnectionCreated, ClientNotificationConnectionFailed, ClientNotificationEpacketReceived, GatewayRequest, + GatewayRequestCommsCheck, GatewayRequestConnection, GatewayRequestConnectionRelease, GatewayRequestConnectionRequest, @@ -166,6 +168,10 @@ def datagram_received(self, data: bytes, addr: tuple[str | Any, int]): loop = asyncio.get_event_loop() request = GatewayRequest.from_json(json.loads(data.decode("utf-8"))) + if isinstance(request, GatewayRequestCommsCheck): + self.wrapped_broadcast(ClientNotificationCommsCheck()) + return + # If not a connection request, attempt to forward to connection context if not isinstance(request, GatewayRequestConnectionRequest): if isinstance(request, GatewayRequestEpacketSend): From e9da94e6709aba838fd43c54a953fb72ec6a5d60 Mon Sep 17 00:00:00 2001 From: Jordan Yates Date: Wed, 4 Feb 2026 10:17:29 +1000 Subject: [PATCH 2/2] tools: check gateway comms before starting For tools that communicate through the gateway scripts, ensure the script is running before the tool attempts to start. Signed-off-by: Jordan Yates --- src/infuse_iot/tools/audio_record.py | 4 ++++ src/infuse_iot/tools/bt_log.py | 4 ++++ src/infuse_iot/tools/data_logger_sync.py | 4 ++++ src/infuse_iot/tools/localhost.py | 4 ++++ src/infuse_iot/tools/ota_upgrade.py | 3 +++ src/infuse_iot/tools/rpc.py | 4 ++++ src/infuse_iot/tools/serial_throughput.py | 4 ++++ src/infuse_iot/tools/tdf_csv.py | 4 ++++ src/infuse_iot/tools/tdf_list.py | 4 ++++ 9 files changed, 35 insertions(+) diff --git a/src/infuse_iot/tools/audio_record.py b/src/infuse_iot/tools/audio_record.py index 4c188f1..7d29067 100644 --- a/src/infuse_iot/tools/audio_record.py +++ b/src/infuse_iot/tools/audio_record.py @@ -5,6 +5,7 @@ __author__ = "Jordan Yates" __copyright__ = "Copyright 2024, Embeint Holdings Pty Ltd" +import sys import time import wave from contextlib import ExitStack @@ -107,6 +108,9 @@ def handle_connection(self): self.handle_channel("right", stack, tdf) def run(self): + if not self._client.comms_check(): + sys.exit("No communications gateway detected (infuse gateway/bt_native)") + try: types = GatewayRequestConnectionRequest.DataType.DATA Console.log_info(f"Connecting to 0x{self._id:016x}") diff --git a/src/infuse_iot/tools/bt_log.py b/src/infuse_iot/tools/bt_log.py index ace88cc..49a34bb 100644 --- a/src/infuse_iot/tools/bt_log.py +++ b/src/infuse_iot/tools/bt_log.py @@ -5,6 +5,7 @@ __author__ = "Jordan Yates" __copyright__ = "Copyright 2024, Embeint Holdings Pty Ltd" +import sys from infuse_iot.commands import InfuseCommand from infuse_iot.common import InfuseType @@ -41,6 +42,9 @@ def add_parser(cls, parser): ) def run(self): + if not self._client.comms_check(): + sys.exit("No communications gateway detected (infuse gateway/bt_native)") + try: types = GatewayRequestConnectionRequest.DataType.LOGGING if self._data: diff --git a/src/infuse_iot/tools/data_logger_sync.py b/src/infuse_iot/tools/data_logger_sync.py index 896e503..e08de7d 100644 --- a/src/infuse_iot/tools/data_logger_sync.py +++ b/src/infuse_iot/tools/data_logger_sync.py @@ -9,6 +9,7 @@ import glob import os import pathlib +import sys from rich.live import Live from rich.progress import ( @@ -176,6 +177,9 @@ def handle_sync(self, live: Live, device_id: int, state: DeviceState): self.state_update(live, "Scanning") def run(self): + if not self._client.comms_check(): + sys.exit("No communications gateway detected (infuse gateway/bt_native)") + with Live(self.progress_table(), refresh_per_second=4) as live: for source, announce in self._client.observe_announce(): self.state_update(live, "Scanning") diff --git a/src/infuse_iot/tools/localhost.py b/src/infuse_iot/tools/localhost.py index 08c5035..2a46ef4 100644 --- a/src/infuse_iot/tools/localhost.py +++ b/src/infuse_iot/tools/localhost.py @@ -7,6 +7,7 @@ import asyncio import pathlib +import sys import threading import time from typing import Any @@ -240,6 +241,9 @@ def recv_thread(self) -> None: self._data_lock.release() def run(self): + if not self._client.comms_check(): + sys.exit("No communications gateway detected (infuse gateway/bt_native)") + Console.init() app = web.Application() # Route for serving the HTML file diff --git a/src/infuse_iot/tools/ota_upgrade.py b/src/infuse_iot/tools/ota_upgrade.py index 2e6ec86..db0f512 100644 --- a/src/infuse_iot/tools/ota_upgrade.py +++ b/src/infuse_iot/tools/ota_upgrade.py @@ -199,6 +199,9 @@ def run_file_copy(self, live: Live, mtu: int, source: HopReceived): self._pending[source.infuse_id] = time.time() + 60 def run(self): + if not self._client.comms_check(): + sys.exit("No communications gateway detected (infuse gateway/bt_native)") + if self._single_diff: self.gateway_diff_load() diff --git a/src/infuse_iot/tools/rpc.py b/src/infuse_iot/tools/rpc.py index 0ebf6e8..c8542fe 100644 --- a/src/infuse_iot/tools/rpc.py +++ b/src/infuse_iot/tools/rpc.py @@ -9,6 +9,7 @@ import importlib import pkgutil import random +import sys import infuse_iot.rpc_wrappers as wrappers from infuse_iot.commands import InfuseCommand, InfuseRpcCommand @@ -74,6 +75,9 @@ def rx_handler(self, pkt: ClientNotification): print(pkt.epacket.payload.decode("utf-8"), end="") def run(self): + if not self._client.comms_check(): + sys.exit("No communications gateway detected (infuse gateway/bt_native)") + try: types = GatewayRequestConnectionRequest.DataType.COMMAND if self._args.conn_log: diff --git a/src/infuse_iot/tools/serial_throughput.py b/src/infuse_iot/tools/serial_throughput.py index e7590e4..7521f82 100644 --- a/src/infuse_iot/tools/serial_throughput.py +++ b/src/infuse_iot/tools/serial_throughput.py @@ -6,6 +6,7 @@ __copyright__ = "Copyright 2024, Embeint Holdings Pty Ltd" import random +import sys import time from infuse_iot.commands import InfuseCommand @@ -81,6 +82,9 @@ def run_send_test(self, num, size, queue_size): print(msg) def run(self): + if not self._client.comms_check(): + sys.exit("No communications gateway detected (infuse gateway/bt_native)") + # No queuing print(f"Averaged across {self._iterations} packets with no queuing:") self.run_send_test(self._iterations, 4, 1) diff --git a/src/infuse_iot/tools/tdf_csv.py b/src/infuse_iot/tools/tdf_csv.py index b4992dd..915f8fb 100644 --- a/src/infuse_iot/tools/tdf_csv.py +++ b/src/infuse_iot/tools/tdf_csv.py @@ -6,6 +6,7 @@ __copyright__ = "Copyright 2024, Embeint Holdings Pty Ltd" import os +import sys import time from infuse_iot.commands import InfuseCommand @@ -38,6 +39,9 @@ def __init__(self, args): self.args = args def run(self): + if not self._client.comms_check(): + sys.exit("No communications gateway detected (infuse gateway/bt_native)") + files = {} while True: diff --git a/src/infuse_iot/tools/tdf_list.py b/src/infuse_iot/tools/tdf_list.py index bf1ff79..2ac9eef 100644 --- a/src/infuse_iot/tools/tdf_list.py +++ b/src/infuse_iot/tools/tdf_list.py @@ -5,6 +5,7 @@ __author__ = "Jordan Yates" __copyright__ = "Copyright 2024, Embeint Holdings Pty Ltd" +import sys import time import tabulate @@ -97,6 +98,9 @@ def append_readings(self, table: list[tuple[str | None, str | None, str, str, st self.append_tdf(table, tdf_name, time_str, t) def run(self) -> None: + if not self._client.comms_check(): + sys.exit("No communications gateway detected (infuse gateway/bt_native)") + while True: msg = self._client.receive() if msg is None: