tools: use common InfuseDeviceId type

Use the common `utils.argparse.InfuseDeviceId` class for device IDs so
that behaviour is standardised across all tools and supports the IDs
without the `0x` prefix.

Signed-off-by: Jordan Yates <jordan@embeint.com>

Author: JordanYates

src/infuse_iot/tools/annotate_events.py

@@ -28,7 +28,7 @@
     default_multicast_address,
 )
 from infuse_iot.time import InfuseTime
-from infuse_iot.util.argparse import ValidFile
+from infuse_iot.util.argparse import InfuseDeviceId, ValidFile
 from infuse_iot.util.console import choose_one
 from infuse_iot.zephyr.errno import errno
 
@@ -37,12 +37,14 @@ class LabelType(enum.Enum):
     CUSTOM = "custom"
     MANUAL = "manual"
 
+
 class TimeCheckType(enum.Enum):
     NONE = "none"
     FORCE = "force"
     AUTO = "auto"
     DEFAULT = "default"
 
+
 class SubCommand(InfuseCommand):
     NAME = "annotate_events"
     HELP = "Annotate events on Infuse Tags"
@@ -58,44 +60,66 @@ class SubCommand(InfuseCommand):
     def add_parser(cls, parser):
         # Logger Selection parameters.
         logger_parser = parser.add_mutually_exclusive_group(required=True)
-        logger_parser.add_argument("--onboard", dest="logger", action="store_const",
-                              const=rpc_enum_data_logger.FLASH_ONBOARD)
-        logger_parser.add_argument("--external", dest="logger", action="store_const",
-                              const=rpc_enum_data_logger.FLASH_REMOVABLE)
-        logger_parser.add_argument("--logger", "-l", type=annotate_wrapper.parse_logger,
-                              help="TDF Data Logger to write the event to")
+        logger_parser.add_argument(
+            "--onboard", dest="logger", action="store_const", const=rpc_enum_data_logger.FLASH_ONBOARD
+        )
+        logger_parser.add_argument(
+            "--external", dest="logger", action="store_const", const=rpc_enum_data_logger.FLASH_REMOVABLE
+        )
+        logger_parser.add_argument(
+            "--logger", "-l", type=annotate_wrapper.parse_logger, help="TDF Data Logger to write the event to"
+        )
 
         # Label selection parameters.
         label_group = parser.add_mutually_exclusive_group(required=True)
         label_group.add_argument(
-            "--preset-labels", "-p", dest="labels", type=ValidFile,
-            help="JSON file containing labels"
+            "--preset-labels", "-p", dest="labels", type=ValidFile, help="JSON file containing labels"
         )
         label_group.add_argument(
-            "--custom-labels", "-c", dest="labels", action="store_const", const=LabelType.CUSTOM,
-            help="Specify custom labels at runtime"
+            "--custom-labels",
+            "-c",
+            dest="labels",
+            action="store_const",
+            const=LabelType.CUSTOM,
+            help="Specify custom labels at runtime",
         )
         label_group.add_argument(
-            "--manual-labels", "-m", dest="labels", action="store_const", const=LabelType.MANUAL,
-            help="Manually enter labels for each event"
+            "--manual-labels",
+            "-m",
+            dest="labels",
+            action="store_const",
+            const=LabelType.MANUAL,
+            help="Manually enter labels for each event",
         )
 
         # Time sync parameters.
         time_group = parser.add_mutually_exclusive_group()
         time_group.add_argument(
-            "--force-time", "-f", dest="time", action="store_const", const=TimeCheckType.FORCE,
-            help="Forcibly update the tag's time before writing annotations"
+            "--force-time",
+            "-f",
+            dest="time",
+            action="store_const",
+            const=TimeCheckType.FORCE,
+            help="Forcibly update the tag's time before writing annotations",
         )
         time_group.add_argument(
-            "--auto-time", "-a", dest="time", action="store_const", const=TimeCheckType.AUTO,
-            help="Automatically update the tag's time if it is not current"
+            "--auto-time",
+            "-a",
+            dest="time",
+            action="store_const",
+            const=TimeCheckType.AUTO,
+            help="Automatically update the tag's time if it is not current",
         )
         time_group.add_argument(
-            "--skip-time", "-s", dest="time", action="store_const", const=TimeCheckType.NONE,
-            help="Do not update the tag's time before writing annotations"
+            "--skip-time",
+            "-s",
+            dest="time",
+            action="store_const",
+            const=TimeCheckType.NONE,
+            help="Do not update the tag's time before writing annotations",
         )
 
-        parser.add_argument("--id", type=lambda x: int(x, 0), help="Device to log events to")
+        parser.add_argument("--id", type=InfuseDeviceId, help="Device to log events to")
 
     def __init__(self, args):
         self._label_type = args.labels
@@ -147,18 +171,14 @@ def load_tag_time(self):
         sync_request_sent = datetime.now()
         assert self.rpc_client is not None
         hdr, rsp = self.rpc_client.run_standard_cmd(
-            time_get.COMMAND_ID,
-            Auth.DEVICE,
-            bytes(params),
-            time_get.response.from_buffer_copy
+            time_get.COMMAND_ID, Auth.DEVICE, bytes(params), time_get.response.from_buffer_copy
         )
         sync_response_received = datetime.now()
 
         if hdr is None:
             raise RuntimeError("Failed to get time from tag")
         if hdr.return_code != 0:
-            raise RuntimeError(f"Error getting time from tag ({hdr.return_code}): "
-                               f"{errno.strerror(-hdr.return_code)}")
+            raise RuntimeError(f"Error getting time from tag ({hdr.return_code}): {errno.strerror(-hdr.return_code)}")
 
         assert isinstance(rsp, time_get.response)
         time_response: time_get.response = rsp
@@ -179,7 +199,7 @@ def check_tag_needs_sync(self) -> bool:
                         f"Tag's clock is out of sync. Update the tag's time?\n"
                         f"Tag:    {tag_datetime_now}\n"
                         f"System: {self._time_of_sync}",
-                        ["Yes", "No"]
+                        ["Yes", "No"],
                     )
                     update = not bool(selection)
                 except IndexError:
@@ -190,25 +210,19 @@ def check_tag_needs_sync(self) -> bool:
     def sync_tag_time(self):
         # Update the tag's time to the current time.
         now = datetime.now().timestamp()
-        params = time_set.request(
-            InfuseTime.epoch_time_from_unix(now)
-        )
+        params = time_set.request(InfuseTime.epoch_time_from_unix(now))
 
         sync_request_sent = datetime.now()
         assert self.rpc_client is not None
         hdr, _ = self.rpc_client.run_standard_cmd(
-            time_set.COMMAND_ID,
-            Auth.DEVICE,
-            bytes(params),
-            time_set.response.from_buffer_copy
+            time_set.COMMAND_ID, Auth.DEVICE, bytes(params), time_set.response.from_buffer_copy
         )
 
         sync_response_received = datetime.now()
         if hdr is None:
             raise RuntimeError("Failed to set time on tag")
         if hdr.return_code != 0:
-            raise RuntimeError(f"Error setting time on tag ({hdr.return_code}): "
-                               f"{errno.strerror(-hdr.return_code)}")
+            raise RuntimeError(f"Error setting time on tag ({hdr.return_code}): {errno.strerror(-hdr.return_code)}")
 
         # Update sync point to reflect new time on tag, assuming the tag's time doesn't change for
         # the duration of the connection.
@@ -249,8 +263,11 @@ def connection_listener(self):
             evt = self._client.receive()
             if evt is None:
                 continue
-            if isinstance(evt, ClientNotificationConnectionDropped) and \
-                    evt.infuse_id == self._device_id and not self.complete:
+            if (
+                isinstance(evt, ClientNotificationConnectionDropped)
+                and evt.infuse_id == self._device_id
+                and not self.complete
+            ):
                 # Ensure the connection wasn't caused by the script existing.
                 print("\n" * (len(self._labels)))  # Clear any pending input lines
                 print(f"Lost connection to {self._device_id:016x}")
@@ -267,10 +284,10 @@ def run(self):
         cl.start()
 
         while not self.complete:
-            with Live(self.draw_connecting(), refresh_per_second=4) as live, \
-                self._client.connection(
-                    self._device_id, GatewayRequestConnectionRequest.DataType.COMMAND
-                ) as mtu:
+            with (
+                Live(self.draw_connecting(), refresh_per_second=4) as live,
+                self._client.connection(self._device_id, GatewayRequestConnectionRequest.DataType.COMMAND) as mtu,
+            ):
                 self.connected = True
                 live.transient = True
                 live.stop()
@@ -293,15 +310,10 @@ def run(self):
                     params = annotate_wrapper.annotate_factory(self._logger, timestamp, label)
 
                     hdr, _ = self.rpc_client.run_standard_cmd(
-                        annotate.COMMAND_ID,
-                        Auth.DEVICE,
-                        bytes(params),
-                        annotate.response.from_buffer_copy
+                        annotate.COMMAND_ID, Auth.DEVICE, bytes(params), annotate.response.from_buffer_copy
                     )
 
                     if hdr is None:
                         print("Failed to send annotation event to tag")
                         continue
-                    annotate_wrapper.handle_response_generic(
-                        hdr.return_code, self._logger, now, label
-                    )
+                    annotate_wrapper.handle_response_generic(hdr.return_code, self._logger, now, label)

src/infuse_iot/tools/audio_record.py

@@ -22,6 +22,7 @@
     default_multicast_address,
 )
 from infuse_iot.tdf import TDF
+from infuse_iot.util.argparse import InfuseDeviceId
 from infuse_iot.util.console import Console
 
 
@@ -47,7 +48,7 @@ def __init__(self, args):
     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")
+        addr_group.add_argument("--id", type=InfuseDeviceId, 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)"
         )

src/infuse_iot/tools/bt_log.py

@@ -18,6 +18,7 @@
     default_multicast_address,
 )
 from infuse_iot.tdf import TDF
+from infuse_iot.util.argparse import InfuseDeviceId
 from infuse_iot.util.console import Console
 
 
@@ -35,7 +36,7 @@ def __init__(self, args):
 
     @classmethod
     def add_parser(cls, parser):
-        parser.add_argument("--id", type=lambda x: int(x, 0), required=True, help="Infuse ID to receive logs for")
+        parser.add_argument("--id", type=InfuseDeviceId, required=True, help="Infuse ID to receive logs for")
         parser.add_argument("--data", action="store_true", help="Subscribe to the data characteristic as well")
         parser.add_argument(
             "--conn-timeout", type=int, default=10000, help="Timeout to wait for a connection to the device (ms)"

src/infuse_iot/tools/cloud.py

@@ -49,7 +49,7 @@
 from infuse_iot.api_client.types import File, Unset
 from infuse_iot.commands import InfuseCommand
 from infuse_iot.credentials import get_api_key
-from infuse_iot.util.argparse import ValidRelease
+from infuse_iot.util.argparse import InfuseDeviceId, ValidRelease
 from infuse_iot.util.console import choose_one, user_confirm, user_response
 from infuse_iot.util.version import Version
 
@@ -188,11 +188,11 @@ def add_parser(cls, parser):
 
         info_parser = tool_parser.add_parser("info", help="General device information")
         info_parser.set_defaults(command_fn=cls.info)
-        info_parser.add_argument("--id", type=str, required=True, help="Infuse-IoT device ID")
+        info_parser.add_argument("--id", type=InfuseDeviceId, required=True, help="Infuse-IoT device ID")
 
         kv_parser = tool_parser.add_parser("kv_state", help="Key-Value device state")
         kv_parser.set_defaults(command_fn=cls.kv_state)
-        kv_parser.add_argument("--id", type=str, required=True, help="Infuse-IoT device ID")
+        kv_parser.add_argument("--id", type=InfuseDeviceId, required=True, help="Infuse-IoT device ID")
         kv_parser.add_argument("--schedules", action="store_true", help="Display task schedules")
         kv_display = kv_parser.add_mutually_exclusive_group()
         kv_display.add_argument("--hex", action="store_true", help="Display values as hex strings instead of decoding")
@@ -202,7 +202,7 @@ def add_parser(cls, parser):
 
         dfu_parser = tool_parser.add_parser("dfu", help="Manage device firmware upgrades")
         dfu_parser.set_defaults(command_fn=cls.dfu)
-        dfu_parser.add_argument("--id", type=str, required=True, help="Infuse-IoT device ID")
+        dfu_parser.add_argument("--id", type=InfuseDeviceId, required=True, help="Infuse-IoT device ID")
         dfu_action = dfu_parser.add_mutually_exclusive_group(required=True)
         dfu_action.add_argument("--schedule", type=str, help="Release ID to upgrade to")
         dfu_action.add_argument("--status", action="store_true", help="Check DFU status")
@@ -212,8 +212,7 @@ def run(self):
             self.args.command_fn(self, client)
 
     def info(self, client: Client):
-        id_int = int(self.args.id, 0)
-        id_str = f"{id_int:016x}"
+        id_str = f"{self.args.id:016x}"
         info = get_device_by_device_id.sync(client=client, device_id=id_str)
         if info is None:
             sys.exit(f"No device with Infuse-IoT ID {id_str} found")
@@ -300,8 +299,7 @@ def _kv_display(self, table: list[tuple[str, str, Any]], key_val: str, name_base
             key_val = ""
 
     def kv_state(self, client: Client):
-        id_int = int(self.args.id, 0)
-        id_str = f"{id_int:016x}"
+        id_str = f"{self.args.id:016x}"
 
         kv_state = get_device_kv_entries_by_device_id.sync(client=client, device_id=id_str)
         if not isinstance(kv_state, list):
@@ -336,8 +334,7 @@ def kv_state(self, client: Client):
         print(tabulate(table))
 
     def dfu(self, client: Client):
-        id_int = int(self.args.id, 0)
-        id_str = f"{id_int:016x}"
+        id_str = f"{self.args.id:016x}"
 
         if self.args.schedule:
             body = models.NewDeviceApplicationUpdate(self.args.schedule)

src/infuse_iot/tools/ota_upgrade.py

@@ -30,7 +30,7 @@
     LocalClient,
     default_multicast_address,
 )
-from infuse_iot.util.argparse import ValidFile, ValidRelease
+from infuse_iot.util.argparse import InfuseDeviceId, ValidFile, ValidRelease
 from infuse_iot.util.crc import crc16_ccitt
 from infuse_iot.zephyr.errno import errno
 
@@ -104,7 +104,7 @@ def add_parser(cls, parser):
             "--conn-timeout", type=int, default=10000, help="Timeout to wait for a connection to the device (ms)"
         )
         explicit = parser.add_mutually_exclusive_group()
-        explicit.add_argument("--id", type=lambda x: int(x, 0), help="Single device to upgrade")
+        explicit.add_argument("--id", type=InfuseDeviceId, help="Single device to upgrade")
         explicit.add_argument("--list", type=ValidFile, help="File containing a list of IDs to upgrade")
 
     def progress_table(self):

src/infuse_iot/tools/provision.py

@@ -20,6 +20,7 @@
 from infuse_iot.api_client.models import Board, Device, DeviceMetadata, Error, NewDevice
 from infuse_iot.commands import InfuseCommand
 from infuse_iot.credentials import get_api_key
+from infuse_iot.util.argparse import InfuseDeviceId
 from infuse_iot.util.console import choose_one
 from infuse_iot.util.soc import nrf, soc, stm
 
@@ -49,7 +50,7 @@ def add_parser(cls, parser):
         parser.add_argument(
             "--id",
             "-i",
-            type=lambda x: int(x, 0),
+            type=InfuseDeviceId,
             help="Infuse device ID to provision as",
         )
         parser.add_argument(

src/infuse_iot/tools/rpc.py

@@ -22,6 +22,7 @@
     LocalClient,
     default_multicast_address,
 )
+from infuse_iot.util.argparse import InfuseDeviceId
 
 
 class SubCommand(InfuseCommand):
@@ -33,7 +34,7 @@ class SubCommand(InfuseCommand):
     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")
+        addr_group.add_argument("--id", type=InfuseDeviceId, help="Infuse ID to run command on")
         parser.add_argument("--conn-log", action="store_true", help="Request logs from remote device")
         parser.add_argument(
             "--conn-timeout", type=int, default=10000, help="Timeout to wait for a connection to the device (ms)"

src/infuse_iot/tools/rpc_cloud.py

@@ -22,6 +22,7 @@
 from infuse_iot.commands import InfuseCommand, InfuseRpcCommand, wrapper_from_command_id
 from infuse_iot.credentials import get_api_key
 from infuse_iot.definitions.rpc import id_type_mapping
+from infuse_iot.util.argparse import InfuseDeviceId
 from infuse_iot.zephyr.errno import errno
 
 
@@ -36,7 +37,7 @@ def add_parser(cls, parser):
 
         parser_queue = subparser.add_parser("queue", help="Queue a RPC to be sent")
         parser_queue.set_defaults(_tool_action="queue")
-        parser_queue.add_argument("--id", required=True, type=lambda x: int(x, 0), help="Infuse ID to run command on")
+        parser_queue.add_argument("--id", required=True, type=InfuseDeviceId, help="Infuse ID to run command on")
         parser_queue.add_argument("--queue-timeout", type=int, default=600, help="Timeout to send command in seconds")
         parser_queue.add_argument("--print-params", action="store_true", help="Print queued RPC request")
         command_list_parser = parser_queue.add_subparsers(title="commands", metavar="<command>", required=True)

src/infuse_iot/tools/tdf_list.py

@@ -20,6 +20,7 @@
 )
 from infuse_iot.tdf import TDF
 from infuse_iot.time import InfuseTime
+from infuse_iot.util.argparse import InfuseDeviceId
 
 
 class SubCommand(InfuseCommand):
@@ -31,7 +32,7 @@ class SubCommand(InfuseCommand):
     def add_parser(cls, parser):
         parser.add_argument("--array-all", action="store_true", help="Display all array values, not just the last")
         parser.add_argument(
-            "--id", type=lambda x: int(x, 0), action="append", default=[], help="Limit displayed TDFs by device ID"
+            "--id", type=InfuseDeviceId, action="append", default=[], help="Limit displayed TDFs by device ID"
         )
         parser.add_argument("--min-rssi", type=int, help="Minimum RSSI to display TDF")