tools: audio_record: added

src/infuse_iot/generated/kv_definitions.py

@@ -139,6 +139,17 @@ class infuse_application_id(VLACompatLittleEndianStruct):
         ]
         _pack_ = 1
 
+    class application_active(VLACompatLittleEndianStruct):
+        """Control STATE_APPLICATION_ACTIVE"""
+
+        NAME = "APPLICATION_ACTIVE"
+        BASE_ID = 6
+        RANGE = 1
+        _fields_ = [
+            ("active", ctypes.c_uint8),
+        ]
+        _pack_ = 1
+
     class fixed_location(VLACompatLittleEndianStruct):
         """Fixed global location of the device"""
 
@@ -423,6 +434,7 @@ class secure_storage_reserved(VLACompatLittleEndianStruct):
         3: bluetooth_ctlr_version,
         4: device_name,
         5: infuse_application_id,
+        6: application_active,
         10: fixed_location,
         20: wifi_ssid,
         21: wifi_psk,

src/infuse_iot/generated/rpc_definitions.py

@@ -232,6 +232,20 @@ class rpc_enum_zperf_data_source(enum.IntEnum):
     ENCRYPT = 128
 
 
+class rpc_enum_key_id(enum.IntEnum):
+    """Infuse security key identifier"""
+
+    NETWORK_KEY = 0
+    SECONDARY_NETWORK_KEY = 1
+
+
+class rpc_enum_key_action(enum.IntEnum):
+    """Infuse security key action"""
+
+    KEY_WRITE = 0
+    KEY_DELETE = 1
+
+
 class RPCDefinitionBase:
     NAME: str
     HELP: str
@@ -1004,6 +1018,29 @@ class response(VLACompatLittleEndianStruct):
         _pack_ = 1
 
 
+class security_key_update(RPCDefinitionBase):
+    """Update key material"""
+
+    NAME = "security_key_update"
+    HELP = "Update key material"
+    DESCRIPTION = "Update key material"
+    COMMAND_ID = 30001
+
+    class request(VLACompatLittleEndianStruct):
+        _fields_ = [
+            ("key_id", ctypes.c_uint8),
+            ("key_action", ctypes.c_uint8),
+            ("key_global_identifier", ctypes.c_uint32),
+            ("key_bitstream", 32 * ctypes.c_uint8),
+            ("reboot_delay", ctypes.c_uint8),
+        ]
+        _pack_ = 1
+
+    class response(VLACompatLittleEndianStruct):
+        _fields_ = []
+        _pack_ = 1
+
+
 class data_sender(RPCDefinitionBase):
     """Send multiple INFUSE_RPC_DATA packets"""
 
@@ -1096,6 +1133,7 @@ class response(VLACompatLittleEndianStruct):
     bt_mcumgr_reboot.COMMAND_ID: bt_mcumgr_reboot,
     gravity_reference_update.COMMAND_ID: gravity_reference_update,
     security_state.COMMAND_ID: security_state,
+    security_key_update.COMMAND_ID: security_key_update,
     data_sender.COMMAND_ID: data_sender,
     data_receiver.COMMAND_ID: data_receiver,
     echo.COMMAND_ID: echo,
@@ -1122,6 +1160,8 @@ class response(VLACompatLittleEndianStruct):
     "rpc_enum_infuse_bt_characteristic",
     "rpc_enum_data_logger",
     "rpc_enum_zperf_data_source",
+    "rpc_enum_key_id",
+    "rpc_enum_key_action",
     "reboot",
     "fault",
     "time_get",
@@ -1155,6 +1195,7 @@ class response(VLACompatLittleEndianStruct):
     "bt_mcumgr_reboot",
     "gravity_reference_update",
     "security_state",
+    "security_key_update",
     "data_sender",
     "data_receiver",
     "echo",

src/infuse_iot/generated/tdf_definitions.py

@@ -1535,6 +1535,89 @@ class battery_soc(TdfReadingBase):
             "soc": "{}",
         }
 
+    class state_event_set(TdfReadingBase):
+        """Infuse-IoT application state transitioned from cleared to set"""
+
+        ID = 55
+        NAME = "STATE_EVENT_SET"
+        _fields_ = [
+            ("state", ctypes.c_uint8),
+        ]
+        _pack_ = 1
+        _postfix_ = {
+            "state": "",
+        }
+        _display_fmt_ = {
+            "state": "{}",
+        }
+
+    class state_event_cleared(TdfReadingBase):
+        """Infuse-IoT application state transitioned from set to cleared"""
+
+        ID = 56
+        NAME = "STATE_EVENT_CLEARED"
+        _fields_ = [
+            ("state", ctypes.c_uint8),
+        ]
+        _pack_ = 1
+        _postfix_ = {
+            "state": "",
+        }
+        _display_fmt_ = {
+            "state": "{}",
+        }
+
+    class state_duration(TdfReadingBase):
+        """Duration an Infuse-IoT application state was asserted for"""
+
+        ID = 57
+        NAME = "STATE_DURATION"
+        _fields_ = [
+            ("state", ctypes.c_uint8),
+            ("duration", ctypes.c_uint32),
+        ]
+        _pack_ = 1
+        _postfix_ = {
+            "state": "",
+            "duration": "",
+        }
+        _display_fmt_ = {
+            "state": "{}",
+            "duration": "{}",
+        }
+
+    class pcm_16bit_chan_left(TdfReadingBase):
+        """16bit PCM (Audio) data for the left channel"""
+
+        ID = 58
+        NAME = "PCM_16BIT_CHAN_LEFT"
+        _fields_ = [
+            ("val", ctypes.c_int16),
+        ]
+        _pack_ = 1
+        _postfix_ = {
+            "val": "",
+        }
+        _display_fmt_ = {
+            "val": "{}",
+        }
+
+    class pcm_16bit_chan_right(TdfReadingBase):
+        """Duration an Infuse-IoT application state was asserted for"""
+
+        ID = 59
+        NAME = "PCM_16BIT_CHAN_RIGHT"
+        _fields_ = [
+            ("val", ctypes.c_int16),
+        ]
+        _pack_ = 1
+        _postfix_ = {
+            "val": "",
+        }
+        _display_fmt_ = {
+            "val": "{}",
+        }
+
 
 id_type_mapping: dict[int, type[TdfReadingBase]] = {
     readings.announce.ID: readings.announce,
@@ -1588,6 +1671,11 @@ class battery_soc(TdfReadingBase):
     readings.exception_stack_frame.ID: readings.exception_stack_frame,
     readings.battery_voltage.ID: readings.battery_voltage,
     readings.battery_soc.ID: readings.battery_soc,
+    readings.state_event_set.ID: readings.state_event_set,
+    readings.state_event_cleared.ID: readings.state_event_cleared,
+    readings.state_duration.ID: readings.state_duration,
+    readings.pcm_16bit_chan_left.ID: readings.pcm_16bit_chan_left,
+    readings.pcm_16bit_chan_right.ID: readings.pcm_16bit_chan_right,
 }
 
 __all__ = [

src/infuse_iot/tools/audio_record.py

@@ -0,0 +1,126 @@
+#!/usr/bin/env python3
+
+"""Connect to remote Bluetooth device serial logs"""
+
+__author__ = "Jordan Yates"
+__copyright__ = "Copyright 2024, Embeint Holdings Pty Ltd"
+
+import time
+import wave
+from contextlib import ExitStack
+
+from infuse_iot.commands import InfuseCommand
+from infuse_iot.common import InfuseID, InfuseType
+from infuse_iot.definitions import tdf as tdf_defs
+from infuse_iot.epacket import interface
+from infuse_iot.socket_comms import (
+    ClientNotificationConnectionDropped,
+    ClientNotificationEpacketReceived,
+    GatewayRequestConnectionRequest,
+    LocalClient,
+    default_multicast_address,
+)
+from infuse_iot.tdf import TDF
+from infuse_iot.util.console import Console
+
+
+class SubCommand(InfuseCommand):
+    NAME = "audio_record"
+    HELP = "Record audio data to a file from TDF"
+    DESCRIPTION = "Record audio data to a file from TDF"
+
+    def __init__(self, args):
+        self._client = LocalClient(default_multicast_address(), 1.0)
+        self._decoder = TDF()
+        if args.gateway:
+            self._id = InfuseID.GATEWAY
+        else:
+            self._id = args.id
+        self._file_prefix = f"{args.name}_" if args.name else ""
+        self._conn_timeout = args.conn_timeout
+        self._freq: int | None = None
+        self._left: wave.Wave_write | None = None
+        self._right: wave.Wave_write | None = None
+
+    @classmethod
+    def add_parser(cls, parser):
+        addr_group = parser.add_mutually_exclusive_group(required=True)
+        addr_group.add_argument("--gateway", action="store_true", help="Run command on local gateway")
+        addr_group.add_argument("--id", type=lambda x: int(x, 0), help="Infuse ID to run command on")
+        parser.add_argument(
+            "--conn-timeout", type=int, default=10000, help="Timeout to wait for a connection to the device (ms)"
+        )
+        parser.add_argument("--name", type=str, help="Filename prefix")
+
+    def handle_channel(self, channel: str, stack: ExitStack, tdf: TDF.Reading):
+        if channel == "left":
+            chan = self._left
+        else:
+            chan = self._right
+
+        if chan is None:
+            filename = f"{self._file_prefix}{int(time.time())}_{channel}.wav"
+            Console.log_info(f"Opening '{filename}'")
+            chan = stack.enter_context(wave.open(filename, "wb"))  # noqa: SIM115
+            chan.setnchannels(1)
+            chan.setsampwidth(2)
+            assert self._freq
+            chan.setframerate(float(self._freq))
+
+            if channel == "left":
+                self._left = chan
+            else:
+                self._right = chan
+
+        samples = b"".join([x.val.to_bytes(2, "little", signed=True) for x in tdf.data])
+        chan.writeframes(samples)
+
+    def handle_connection(self):
+        with ExitStack() as stack:
+            Console.log_info("Waiting for frequency information...")
+            while evt := self._client.receive():
+                if evt is None:
+                    continue
+                if isinstance(evt, ClientNotificationConnectionDropped):
+                    Console.log_error(f"Connection to {self._id:016x} lost")
+                    break
+                if not isinstance(evt, ClientNotificationEpacketReceived):
+                    continue
+                source = evt.epacket.route[0]
+                if source.infuse_id != self._id:
+                    continue
+                if source.interface != interface.ID.BT_CENTRAL:
+                    continue
+                if evt.epacket.ptype != InfuseType.TDF:
+                    continue
+                for tdf in self._decoder.decode(evt.epacket.payload):
+                    if self._freq is None:
+                        if tdf.id == tdf_defs.readings.idx_array_freq.ID:
+                            self._freq = tdf.data[0].frequency
+                            Console.log_info(f"Audio frequency is {self._freq} Hz")
+                        else:
+                            # Don't write until metadata is known
+                            continue
+                    if tdf.id == tdf_defs.readings.pcm_16bit_chan_left.ID:
+                        self.handle_channel("left", stack, tdf)
+                    elif tdf.id == tdf_defs.readings.pcm_16bit_chan_right.ID:
+                        self.handle_channel("right", stack, tdf)
+
+    def run(self):
+        try:
+            types = GatewayRequestConnectionRequest.DataType.DATA
+            Console.log_info(f"Connecting to 0x{self._id:016x}")
+            with self._client.connection(self._id, types, self._conn_timeout) as _:
+                self.handle_connection()
+
+        except KeyboardInterrupt:
+            Console.log_error(f"Disconnecting from {self._id:016x}")
+        except ConnectionRefusedError:
+            Console.log_error(f"Unable to connect to {self._id:016x}")
+
+        if self._left:
+            assert self._freq
+            Console.log_text(f"Left Channel Recorded: {self._left.getnframes() / self._freq} Seconds")
+        if self._right:
+            assert self._freq
+            Console.log_text(f"Right Channel Recorded: {self._right.getnframes() / self._freq} Seconds")

src/infuse_iot/tools/bt_log.py

@@ -66,9 +66,9 @@ def run(self):
                             t = tdf.data[-1]
                             t_str = f"{tdf.time:.3f}" if tdf.time else "N/A"
                             if len(tdf.data) > 1:
-                                print(f"{t_str} TDF: {t.name}[{len(tdf.data)}]")
+                                print(f"{t_str} TDF: {t.NAME}[{len(tdf.data)}]")
                             else:
-                                print(f"{t_str} TDF: {t.name}")
+                                print(f"{t_str} TDF: {t.NAME}")
 
         except KeyboardInterrupt:
             print(f"Disconnecting from {self._id:016x}")

src/infuse_iot/util/console.py

@@ -60,7 +60,7 @@ def log(timestamp: datetime.datetime, colour, string: str):
         """Log colourised string to terminal"""
         ts = timestamp.strftime("%H:%M:%S.%f")[:-3]
         with _lock:
-            print(f"[{ts}]{colour} {string}")
+            print(f"[{ts}]{colour} {string}{colorama.Fore.RESET}")
 
 
 def choose_one(title: str, options: list[str]) -> tuple[int, str]: