Skip to content

rpc_wrappers: lte_state_v2: added #98

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
Dec 2, 2025
Merged

rpc_wrappers: lte_state_v2: added #98

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
6fa0351
util: internal: add `importlib` submodule import
JordanYates Dec 2, 2025
4b96445
generated: regenerate definitions
JordanYates Nov 29, 2025
13078a0
rpc_wrappers: lte_state: use builtin defs
JordanYates Nov 29, 2025
82ef951
rpc_wrappers: lte_state_v2: added
JordanYates Nov 29, 2025
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
71 changes: 68 additions & 3 deletions src/infuse_iot/generated/kv_definitions.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,33 @@ class kv_utc_hms(VLACompatLittleEndianStruct):
]
_pack_ = 1

class kv_algorithm_logging(VLACompatLittleEndianStruct):
"""Algorithm logging configuration"""

_fields_ = [
("loggers", ctypes.c_uint8),
("tdf_mask", ctypes.c_uint8),
]
_pack_ = 1

class kv_algorithm_stationary_windowed_args(VLACompatLittleEndianStruct):
"""Arguments for 'Stationary Windowed' algorithm"""

_fields_ = [
("window_seconds", ctypes.c_uint32),
("std_dev_threshold_ug", ctypes.c_uint32),
]
_pack_ = 1

class kv_algorithm_tilt_args(VLACompatLittleEndianStruct):
"""Arguments for 'Tilt' algorithm"""

_fields_ = [
("iir_filter_alpha", ctypes.c_float),
("one_g_percent", ctypes.c_uint8),
]
_pack_ = 1


class slots:
class reboots(VLACompatLittleEndianStruct):
Expand Down Expand Up @@ -265,7 +292,7 @@ class lte_modem_esn(VLACompatLittleEndianStruct):
_pack_ = 1

class lte_modem_imei(VLACompatLittleEndianStruct):
"""'International Modem Equiment Identifier' as returned by AT+CGSN=1"""
"""'International Modem Equipment Identifier' as returned by AT+CGSN=1"""

NAME = "LTE_MODEM_IMEI"
BASE_ID = 43
Expand Down Expand Up @@ -349,7 +376,7 @@ class lora_config(VLACompatLittleEndianStruct):
_pack_ = 1

class bluetooth_throughput_limit(VLACompatLittleEndianStruct):
"""Request connected Bluetooth peers to limit throughtput"""
"""Request connected Bluetooth peers to limit throughput"""

NAME = "BLUETOOTH_THROUGHPUT_LIMIT"
BASE_ID = 52
Expand All @@ -371,6 +398,17 @@ class led_disable_daily_time_range(VLACompatLittleEndianStruct):
]
_pack_ = 1

class memfault_disable(VLACompatLittleEndianStruct):
"""Disable Memfault reporting at runtime"""

NAME = "MEMFAULT_DISABLE"
BASE_ID = 54
RANGE = 1
_fields_ = [
("disable", ctypes.c_uint8),
]
_pack_ = 1

class gravity_reference(VLACompatLittleEndianStruct):
"""Reference gravity vector for tilt calculations"""

Expand All @@ -396,6 +434,30 @@ class geofence(VLACompatLittleEndianStruct):
vla_field = ("points", 0 * structs.gcs_location)
_pack_ = 1

class alg_stationary_windowed_args(VLACompatLittleEndianStruct):
"""Configuration for the 'Stationary Windowed' algorithm"""

NAME = "ALG_STATIONARY_WINDOWED_ARGS"
BASE_ID = 200
RANGE = 1
_fields_ = [
("logging", structs.kv_algorithm_logging),
("args", structs.kv_algorithm_stationary_windowed_args),
]
_pack_ = 1

class alg_tilt_args(VLACompatLittleEndianStruct):
"""Configuration for the 'Tilt' algorithm"""

NAME = "ALG_TILT_ARGS"
BASE_ID = 201
RANGE = 1
_fields_ = [
("logging", structs.kv_algorithm_logging),
("args", structs.kv_algorithm_tilt_args),
]
_pack_ = 1

class task_schedules_default_id(VLACompatLittleEndianStruct):
"""Unique identifier for default schedule set"""

Expand All @@ -417,10 +479,10 @@ class task_schedules(VLACompatLittleEndianStruct):
("task_id", ctypes.c_uint8),
("validity", ctypes.c_uint8),
("periodicity_type", ctypes.c_uint8),
("boot_lockout_minutes", ctypes.c_uint8),
("timeout_s", ctypes.c_uint32),
("battery_start", structs.kv_range_u8),
("battery_terminate", structs.kv_range_u8),
("periodicity", ctypes.c_uint32),
]
vla_field = ("_remainder", 0 * ctypes.c_uint8)
_pack_ = 1
Expand Down Expand Up @@ -465,6 +527,7 @@ class secure_storage_reserved(VLACompatLittleEndianStruct):
51: lora_config,
52: bluetooth_throughput_limit,
53: led_disable_daily_time_range,
54: memfault_disable,
60: gravity_reference,
100: geofence,
101: geofence,
Expand All @@ -482,6 +545,8 @@ class secure_storage_reserved(VLACompatLittleEndianStruct):
113: geofence,
114: geofence,
115: geofence,
200: alg_stationary_windowed_args,
201: alg_tilt_args,
1000: task_schedules_default_id,
1001: task_schedules,
1002: task_schedules,
Expand Down
62 changes: 59 additions & 3 deletions src/infuse_iot/generated/rpc_definitions.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,6 +126,30 @@ class rpc_struct_lte_state(VLACompatLittleEndianStruct):
_pack_ = 1


class rpc_struct_lte_state_v2(VLACompatLittleEndianStruct):
"""LTE interface status"""

_fields_ = [
("registration_state", ctypes.c_uint8),
("access_technology", ctypes.c_uint8),
("mcc", ctypes.c_uint16),
("mnc", ctypes.c_uint16),
("cell_id", ctypes.c_uint32),
("tac", ctypes.c_uint32),
("tau", ctypes.c_int32),
("earfcn", ctypes.c_uint16),
("band", ctypes.c_uint8),
("psm_active_time", ctypes.c_int16),
("edrx_interval", ctypes.c_float),
("edrx_paging_window", ctypes.c_float),
("rsrp", ctypes.c_int16),
("rsrq", ctypes.c_int8),
("as_rai", ctypes.c_uint8),
("cp_rai", ctypes.c_uint8),
]
_pack_ = 1


class rpc_struct_wifi_scan_result(VLACompatLittleEndianStruct):
"""WiFi interface status"""

Expand Down Expand Up @@ -246,6 +270,14 @@ class rpc_enum_key_action(enum.IntEnum):
KEY_DELETE = 1


class rpc_enum_support_status(enum.IntEnum):
"""Field support status"""

UNSUPPORTED = 0
SUPPORTED = 1
UNKNOWN = 255


class RPCDefinitionBase:
NAME: str
HELP: str
Expand Down Expand Up @@ -377,11 +409,11 @@ class response(VLACompatLittleEndianStruct):


class kv_reflect_crcs(RPCDefinitionBase):
"""Read KV store CRC's"""
"""Read KV store CRCs"""

NAME = "kv_reflect_crcs"
HELP = "Read KV store CRC's"
DESCRIPTION = "Read KV store CRC's"
HELP = "Read KV store CRCs"
DESCRIPTION = "Read KV store CRCs"
COMMAND_ID = 7

class request(VLACompatLittleEndianStruct):
Expand Down Expand Up @@ -738,6 +770,26 @@ class response(VLACompatLittleEndianStruct):
_pack_ = 1


class lte_state_v2(RPCDefinitionBase):
"""Get current LTE interface state (with RAI information)"""

NAME = "lte_state_v2"
HELP = "Get current LTE interface state (with RAI information)"
DESCRIPTION = "Get current LTE interface state (with RAI information)"
COMMAND_ID = 23

class request(VLACompatLittleEndianStruct):
_fields_ = []
_pack_ = 1

class response(VLACompatLittleEndianStruct):
_fields_ = [
("common", rpc_struct_network_state),
("lte", rpc_struct_lte_state_v2),
]
_pack_ = 1


class coap_download(RPCDefinitionBase):
"""Download a file from a COAP server (Infuse-IoT DTLS protected)"""

Expand Down Expand Up @@ -1122,6 +1174,7 @@ class response(VLACompatLittleEndianStruct):
lte_at_cmd.COMMAND_ID: lte_at_cmd,
lte_state.COMMAND_ID: lte_state,
data_logger_read_available.COMMAND_ID: data_logger_read_available,
lte_state_v2.COMMAND_ID: lte_state_v2,
coap_download.COMMAND_ID: coap_download,
zperf_upload.COMMAND_ID: zperf_upload,
file_write_basic.COMMAND_ID: file_write_basic,
Expand Down Expand Up @@ -1150,6 +1203,7 @@ class response(VLACompatLittleEndianStruct):
"rpc_struct_network_state",
"rpc_struct_wifi_state",
"rpc_struct_lte_state",
"rpc_struct_lte_state_v2",
"rpc_struct_wifi_scan_result",
"rpc_struct_xyz_s16",
"rpc_struct_infuse_state",
Expand All @@ -1162,6 +1216,7 @@ class response(VLACompatLittleEndianStruct):
"rpc_enum_zperf_data_source",
"rpc_enum_key_id",
"rpc_enum_key_action",
"rpc_enum_support_status",
"reboot",
"fault",
"time_get",
Expand All @@ -1184,6 +1239,7 @@ class response(VLACompatLittleEndianStruct):
"lte_at_cmd",
"lte_state",
"data_logger_read_available",
"lte_state_v2",
"coap_download",
"zperf_upload",
"file_write_basic",
Expand Down
83 changes: 13 additions & 70 deletions src/infuse_iot/rpc_wrappers/lte_state.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
#!/usr/bin/env python3

import ctypes
import ipaddress

import infuse_iot.definitions.rpc as defs
Expand All @@ -10,66 +9,7 @@
from infuse_iot.zephyr.errno import errno


class interface_state(ctypes.LittleEndianStructure):
_fields_ = [
("_state", ctypes.c_uint8),
("_if_flags", ctypes.c_uint32),
("_l2_flags", ctypes.c_uint16),
("mtu", ctypes.c_uint16),
("ipv4", 4 * ctypes.c_uint8),
("ipv6", 16 * ctypes.c_uint8),
]
_pack_ = 1

@property
def state(self):
return z_nif.OperationalState(self._state)

@property
def if_flags(self):
return z_nif.InterfaceFlags(self._if_flags)

@property
def l2_flags(self):
return z_nif.L2Flags(self._l2_flags)


class lte_state_struct(ctypes.LittleEndianStructure):
_fields_ = [
("_state", ctypes.c_uint8),
("_act", ctypes.c_uint8),
("mcc", ctypes.c_uint16),
("mnc", ctypes.c_uint16),
("cell_id", ctypes.c_uint32),
("tac", ctypes.c_uint32),
("tau", ctypes.c_int32),
("earfcn", ctypes.c_uint16),
("band", ctypes.c_uint8),
("psm_active_time", ctypes.c_int16),
("edrx_interval", ctypes.c_float),
("edrx_window", ctypes.c_float),
("rsrp", ctypes.c_int16),
("rsrq", ctypes.c_int8),
]
_pack_ = 1

@property
def state(self):
return z_lte.RegistrationState(self._state)

@property
def access_technology(self):
return z_lte.AccessTechnology(self._act)


class lte_state(InfuseRpcCommand, defs.lte_state):
class response(ctypes.LittleEndianStructure):
_fields_ = [
("common", interface_state),
("lte", lte_state_struct),
]
_pack_ = 1

@classmethod
def add_parser(cls, parser):
return
Expand All @@ -95,22 +35,25 @@ def handle_response(self, return_code, response):
ipv4 = ipaddress.IPv4Address(bytes(common.ipv4))
ipv6 = ipaddress.IPv6Address(bytes(common.ipv6))

reg_class = z_lte.RegistrationState
lte_state = reg_class(lte.registration_state)

print("Interface State:")
print(f"\t State: {common.state.name}")
print(f"\t IF Flags: {common.if_flags}")
print(f"\t L2 Flags: {common.l2_flags}")
print(f"\t State: {z_nif.OperationalState(common.state).name}")
print(f"\t IF Flags: {z_nif.InterfaceFlags(common.if_flags)}")
print(f"\t L2 Flags: {z_nif.L2Flags(common.l2_flags)}")
print(f"\t MTU: {common.mtu}")
print(f"\t IPv4: {ipv4}")
print(f"\t IPv6: {ipv6}")
print("LTE State:")
print(f"\t Reg State: {lte.state}")
print(f"\t Reg State: {lte_state}")

reg_class = z_lte.RegistrationState
valid = (
lte.state == reg_class.REGISTERED_HOME
or lte.state == reg_class.REGISTERED_ROAMING
or lte.state == reg_class.SEARCHING
lte_state == reg_class.REGISTERED_HOME
or lte_state == reg_class.REGISTERED_ROAMING
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)
Expand All @@ -120,8 +63,8 @@ def handle_response(self, return_code, response):
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_window} s" if lte.edrx_window != -1.0 else "N/A"
print(f"\t Access Tech: {lte.access_technology}")
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}")
Expand Down
Loading