diff --git a/src/infuse_iot/rpc_wrappers/lte_state.py b/src/infuse_iot/rpc_wrappers/lte_state.py index 9cbd6cb..640c17d 100644 --- a/src/infuse_iot/rpc_wrappers/lte_state.py +++ b/src/infuse_iot/rpc_wrappers/lte_state.py @@ -54,26 +54,31 @@ def handle_response(self, return_code, response): or lte_state == reg_class.SEARCHING ) access_tech = z_lte.AccessTechnology(lte.access_technology) - if valid: - if lte.earfcn != 0: - freq_dl, freq_ul = z_lte.LteBands.earfcn_to_freq(lte.earfcn) - freq_string = f" (UL: {int(freq_ul)}MHz, DL: {int(freq_dl)}MHz)" - else: - freq_string = "" - country = z_lte.MobileCountryCodes.name_from_mcc(lte.mcc) - active_str = f"{lte.psm_active_time} s" if lte.psm_active_time != 65535 else "N/A" - edrx_interval_str = f"{lte.edrx_interval} s" if lte.edrx_interval != -1.0 else "N/A" - edrx_window_str = f"{lte.edrx_paging_window} s" if lte.edrx_paging_window != -1.0 else "N/A" - print(f"\t Access Tech: {access_tech}") - print(f"\t Country Code: {lte.mcc} ({country})") - print(f"\t Network Code: {lte.mnc}") - print(f"\t Cell ID: {lte.cell_id}") - print(f"\t Tracking Area: {lte.tac}") - print(f"\t TAU: {lte.tau} s") - print(f"\t EARFCN: {lte.earfcn}{freq_string}") - print(f"\t Band: {lte.band}") - print(f"\tPSM Active Time: {active_str}") - print(f"\t eDRX Interval: {edrx_interval_str}") - print(f"\t eDRX Window: {edrx_window_str}") - print(f"\t RSRP: {lte.rsrp} dBm") - print(f"\t RSRQ: {lte.rsrq} dB") + if not valid: + return + band = lte.band + if lte.earfcn != 0: + freq_dl, freq_ul = z_lte.LteBands.earfcn_to_freq(lte.earfcn) + freq_string = f" (UL: {int(freq_ul)}MHz, DL: {int(freq_dl)}MHz)" + if band == 0: + # Band not provided by device, derive from EARFCN + band = z_lte.LteBands.earfcn_to_band(lte.earfcn).band + else: + freq_string = "" + country = z_lte.MobileCountryCodes.name_from_mcc(lte.mcc) + active_str = f"{lte.psm_active_time} s" if lte.psm_active_time != 65535 else "N/A" + edrx_interval_str = f"{lte.edrx_interval} s" if lte.edrx_interval != -1.0 else "N/A" + edrx_window_str = f"{lte.edrx_paging_window} s" if lte.edrx_paging_window != -1.0 else "N/A" + print(f"\t Access Tech: {access_tech}") + print(f"\t Country Code: {lte.mcc} ({country})") + print(f"\t Network Code: {lte.mnc}") + print(f"\t Cell ID: {lte.cell_id}") + print(f"\t Tracking Area: {lte.tac}") + print(f"\t TAU: {lte.tau} s") + print(f"\t EARFCN: {lte.earfcn}{freq_string}") + print(f"\t Band: {band}") + print(f"\tPSM Active Time: {active_str}") + print(f"\t eDRX Interval: {edrx_interval_str}") + print(f"\t eDRX Window: {edrx_window_str}") + print(f"\t RSRP: {lte.rsrp} dBm") + print(f"\t RSRQ: {lte.rsrq} dB") diff --git a/src/infuse_iot/rpc_wrappers/lte_state_v2.py b/src/infuse_iot/rpc_wrappers/lte_state_v2.py index dea80d5..35c3881 100644 --- a/src/infuse_iot/rpc_wrappers/lte_state_v2.py +++ b/src/infuse_iot/rpc_wrappers/lte_state_v2.py @@ -54,31 +54,36 @@ def handle_response(self, return_code, response): or lte_state == reg_class.SEARCHING ) access_tech = z_lte.AccessTechnology(lte.access_technology) - if valid: - if lte.earfcn != 0: - freq_dl, freq_ul = z_lte.LteBands.earfcn_to_freq(lte.earfcn) - freq_string = f" (UL: {int(freq_ul)}MHz, DL: {int(freq_dl)}MHz)" - else: - freq_string = "" - country = z_lte.MobileCountryCodes.name_from_mcc(lte.mcc) - active_str = f"{lte.psm_active_time} s" if lte.psm_active_time != 65535 else "N/A" - edrx_interval_str = f"{lte.edrx_interval} s" if lte.edrx_interval != -1.0 else "N/A" - edrx_window_str = f"{lte.edrx_paging_window} s" if lte.edrx_paging_window != -1.0 else "N/A" - as_rai = defs.rpc_enum_support_status(lte.as_rai) - cp_rai = defs.rpc_enum_support_status(lte.cp_rai) + if not valid: + return + band = lte.band + if lte.earfcn != 0: + freq_dl, freq_ul = z_lte.LteBands.earfcn_to_freq(lte.earfcn) + freq_string = f" (UL: {int(freq_ul)}MHz, DL: {int(freq_dl)}MHz)" + if band == 0: + # Band not provided by device, derive from EARFCN + band = z_lte.LteBands.earfcn_to_band(lte.earfcn).band + else: + freq_string = "" + country = z_lte.MobileCountryCodes.name_from_mcc(lte.mcc) + active_str = f"{lte.psm_active_time} s" if lte.psm_active_time != 65535 else "N/A" + edrx_interval_str = f"{lte.edrx_interval} s" if lte.edrx_interval != -1.0 else "N/A" + edrx_window_str = f"{lte.edrx_paging_window} s" if lte.edrx_paging_window != -1.0 else "N/A" + as_rai = defs.rpc_enum_support_status(lte.as_rai) + cp_rai = defs.rpc_enum_support_status(lte.cp_rai) - print(f"\t Access Tech: {access_tech}") - print(f"\t Country Code: {lte.mcc} ({country})") - print(f"\t Network Code: {lte.mnc}") - print(f"\t Cell ID: {lte.cell_id}") - print(f"\t Tracking Area: {lte.tac}") - print(f"\t TAU: {lte.tau} s") - print(f"\t EARFCN: {lte.earfcn}{freq_string}") - print(f"\t Band: {lte.band}") - print(f"\tPSM Active Time: {active_str}") - print(f"\t eDRX Interval: {edrx_interval_str}") - print(f"\t eDRX Window: {edrx_window_str}") - print(f"\t RSRP: {lte.rsrp} dBm") - print(f"\t RSRQ: {lte.rsrq} dB") - print(f"\t AS-RAI: {as_rai.name}") - print(f"\t CP-RAI: {cp_rai.name}") + print(f"\t Access Tech: {access_tech}") + print(f"\t Country Code: {lte.mcc} ({country})") + print(f"\t Network Code: {lte.mnc}") + print(f"\t Cell ID: {lte.cell_id}") + print(f"\t Tracking Area: {lte.tac}") + print(f"\t TAU: {lte.tau} s") + print(f"\t EARFCN: {lte.earfcn}{freq_string}") + print(f"\t Band: {band}") + print(f"\tPSM Active Time: {active_str}") + print(f"\t eDRX Interval: {edrx_interval_str}") + print(f"\t eDRX Window: {edrx_window_str}") + print(f"\t RSRP: {lte.rsrp} dBm") + print(f"\t RSRQ: {lte.rsrq} dB") + print(f"\t AS-RAI: {as_rai.name}") + print(f"\t CP-RAI: {cp_rai.name}") diff --git a/src/infuse_iot/tools/annotate_events.py b/src/infuse_iot/tools/annotate_events.py index 992f1d6..a099318 100644 --- a/src/infuse_iot/tools/annotate_events.py +++ b/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) diff --git a/src/infuse_iot/tools/audio_record.py b/src/infuse_iot/tools/audio_record.py index 7d29067..4d040a1 100644 --- a/src/infuse_iot/tools/audio_record.py +++ b/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)" ) diff --git a/src/infuse_iot/tools/bt_log.py b/src/infuse_iot/tools/bt_log.py index 56553e1..9a93a67 100644 --- a/src/infuse_iot/tools/bt_log.py +++ b/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)" diff --git a/src/infuse_iot/tools/cloud.py b/src/infuse_iot/tools/cloud.py index dd5ea42..7dc9881 100644 --- a/src/infuse_iot/tools/cloud.py +++ b/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) @@ -637,6 +634,17 @@ def upload(self, client: Client): sys.exit(f"Unexpected internal type {type(application)}") ota_files = glob.glob(str(release.dir / "ota-*.bin")) + if len(ota_files) == 0: + # Old release folder, try and find the right file + app_folder = release.dir / release_app_meta["primary"] / "zephyr" + tfm_file = app_folder / "tfm_s_zephyr_ns_signed.bin" + std_file = app_folder / "zephyr.signed.bin" + if tfm_file.exists(): + ota_files = [str(tfm_file)] + elif std_file.exists(): + ota_files = [str(std_file)] + if len(ota_files) == 1 and not user_confirm(f"Use file {ota_files[0]} for upload?"): + sys.exit() if len(ota_files) != 1: sys.exit(f"Unexpected OTA file search result {ota_files}") diff --git a/src/infuse_iot/tools/ota_upgrade.py b/src/infuse_iot/tools/ota_upgrade.py index d51ceba..42a63d2 100644 --- a/src/infuse_iot/tools/ota_upgrade.py +++ b/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): diff --git a/src/infuse_iot/tools/provision.py b/src/infuse_iot/tools/provision.py index daa2460..3a6e8ee 100644 --- a/src/infuse_iot/tools/provision.py +++ b/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( diff --git a/src/infuse_iot/tools/rpc.py b/src/infuse_iot/tools/rpc.py index 8b08a33..f2e8d0d 100644 --- a/src/infuse_iot/tools/rpc.py +++ b/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)" diff --git a/src/infuse_iot/tools/rpc_cloud.py b/src/infuse_iot/tools/rpc_cloud.py index 631f495..6e6bb91 100644 --- a/src/infuse_iot/tools/rpc_cloud.py +++ b/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="", required=True) diff --git a/src/infuse_iot/tools/tdf_list.py b/src/infuse_iot/tools/tdf_list.py index ef11a55..e1469a4 100644 --- a/src/infuse_iot/tools/tdf_list.py +++ b/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") diff --git a/src/infuse_iot/util/argparse.py b/src/infuse_iot/util/argparse.py index 662e035..6590ddf 100644 --- a/src/infuse_iot/util/argparse.py +++ b/src/infuse_iot/util/argparse.py @@ -79,3 +79,13 @@ def to_ctype(cls, addr_type: rpc_enum_bt_le_addr_type, value: int) -> rpc_struct def integer_value(cls, string) -> int: """Integer value from address string""" return cast(int, cls(string)) + + +class InfuseDeviceId: + """Infuse-IoT Device ID""" + + def __new__(cls, string) -> int: # type: ignore + try: + return int(string, 16) + except ValueError as e: + raise argparse.ArgumentTypeError(f"{string} is not a valid hex ID") from e diff --git a/src/infuse_iot/zephyr/lte.py b/src/infuse_iot/zephyr/lte.py index 0362652..bd59baf 100644 --- a/src/infuse_iot/zephyr/lte.py +++ b/src/infuse_iot/zephyr/lte.py @@ -116,7 +116,7 @@ def __str__(self): class LteBand: - def __init__(self, band, freq_dl_low, offset_dl, freq_ul_low, offset_ul): + def __init__(self, band: int, freq_dl_low: float, offset_dl: int, freq_ul_low: float, offset_ul: int): self.band = band self.fdl_low = freq_dl_low self.ndl = offset_dl @@ -168,25 +168,27 @@ class LteBands: } @classmethod - def earfcn_to_freq(cls, earfcn_dl): + def earfcn_to_band(cls, earfcn_dl: int) -> LteBand: prev = None for b in cls.bands.values(): if prev is None: prev = b continue if earfcn_dl >= prev.ndl and earfcn_dl < b.ndl: - info = prev - - ul_offset = info.nul - info.ndl - earfcn_ul = earfcn_dl + ul_offset - - freq_dl = info.fdl_low + 0.1 * (earfcn_dl - info.ndl) - freq_ul = info.ful_low + 0.1 * (earfcn_ul - info.nul) - return (freq_dl, freq_ul) - + return prev prev = b raise ValueError(f"EARFCN {earfcn_dl} is invalid") + @classmethod + def earfcn_to_freq(cls, earfcn_dl: int): + info = cls.earfcn_to_band(earfcn_dl) + ul_offset = info.nul - info.ndl + earfcn_ul = earfcn_dl + ul_offset + + freq_dl = info.fdl_low + 0.1 * (earfcn_dl - info.ndl) + freq_ul = info.ful_low + 0.1 * (earfcn_ul - info.nul) + return (freq_dl, freq_ul) + class MobileCountryCodes: _names = { diff --git a/tests/util/test_argparse.py b/tests/util/test_argparse.py index 04fbf79..a3a8584 100644 --- a/tests/util/test_argparse.py +++ b/tests/util/test_argparse.py @@ -6,7 +6,7 @@ import pytest -from infuse_iot.util.argparse import BtLeAddress, ValidDir, ValidFile +from infuse_iot.util.argparse import BtLeAddress, InfuseDeviceId, ValidDir, ValidFile assert "TOXTEMPDIR" in os.environ, "you must run these tests using tox" @@ -46,3 +46,16 @@ def test_bt_le_address(): assert isinstance(addr, int) addr = BtLeAddress("123456aaFF4A") assert isinstance(addr, int) + + +def test_infuse_device_id(): + with pytest.raises(argparse.ArgumentTypeError): + InfuseDeviceId("NotHex") + with pytest.raises(argparse.ArgumentTypeError): + InfuseDeviceId("aabb::00") + assert InfuseDeviceId("0x00aa") == 0xAA + assert InfuseDeviceId("00aa") == 0xAA + assert InfuseDeviceId("0x99") == 0x99 + assert InfuseDeviceId("99") == 0x99 + assert InfuseDeviceId("0x1234aa43bc") == 0x1234AA43BC + assert InfuseDeviceId("1234aa43bc") == 0x1234AA43BC