Skip to content

rpc_wrappers: lte_state: derive band from EARFCN #149

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JordanYates merged 4 commits into from
Jul 7, 2026
Merged

rpc_wrappers: lte_state: derive band from EARFCN #149

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
d432694
rpc_wrappers: lte_state: derive band from EARFCN
JordanYates Jul 6, 2026
5e53d2c
tools: cloud: apps: upload support old release folders
JordanYates Jul 7, 2026
c555b56
util: argparse: device ID helper class
JordanYates Jul 7, 2026
b6ce5e2
tools: use common `InfuseDeviceId` type
JordanYates Jul 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 28 additions & 23 deletions src/infuse_iot/rpc_wrappers/lte_state.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
59 changes: 32 additions & 27 deletions src/infuse_iot/rpc_wrappers/lte_state_v2.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}")
110 changes: 61 additions & 49 deletions src/infuse_iot/tools/annotate_events.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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"
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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:
Expand All @@ -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.
Expand Down Expand Up @@ -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}")
Expand All @@ -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()
Expand All @@ -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)
3 changes: 2 additions & 1 deletion src/infuse_iot/tools/audio_record.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand All @@ -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)"
)
Expand Down
3 changes: 2 additions & 1 deletion src/infuse_iot/tools/bt_log.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand All @@ -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)"
Expand Down
Loading
Loading