From 54a2d31e97542373d7a5bc6b3646792fff17f088 Mon Sep 17 00:00:00 2001 From: Trey Moen <50057480+greatgitsby@users.noreply.github.com> Date: Tue, 7 Apr 2026 18:14:45 -0700 Subject: [PATCH] esim: TLV helpers, acquire_channel, and TiciLPA rework (#37776) --- system/hardware/tici/lpa.py | 108 ++++++++++++++++++++++++++---------- 1 file changed, 79 insertions(+), 29 deletions(-) diff --git a/system/hardware/tici/lpa.py b/system/hardware/tici/lpa.py index 93cbca247..ca93b3e41 100644 --- a/system/hardware/tici/lpa.py +++ b/system/hardware/tici/lpa.py @@ -2,6 +2,7 @@ import atexit import base64 +import fcntl import math import os import serial @@ -10,7 +11,9 @@ import sys import termios import time -from collections.abc import Generator +from collections.abc import Callable, Generator +from contextlib import contextmanager +from typing import Any from openpilot.system.hardware.base import LPABase, Profile @@ -27,21 +30,35 @@ OPEN_ISDR_RETRIES = 10 OPEN_ISDR_RETRY_DELAY_S = 0.25 OPEN_ISDR_RESET_ATTEMPT = 5 SEND_APDU_RETRIES = 3 +LOCK_FILE = '/dev/shm/modem_lpa.lock' DEBUG = os.environ.get("DEBUG") == "1" # TLV Tags TAG_ICCID = 0x5A +TAG_STATUS = 0x80 TAG_PROFILE_INFO_LIST = 0xBF2D +TAG_OK = 0xA0 STATE_LABELS = {0: "disabled", 1: "enabled", 255: "unknown"} ICON_LABELS = {0: "jpeg", 1: "png", 255: "unknown"} CLASS_LABELS = {0: "test", 1: "provisioning", 2: "operational", 255: "unknown"} +# TLV tag -> (field_name, decoder) +FieldMap = dict[int, tuple[str, Callable[[bytes], Any]]] + def b64e(data: bytes) -> str: return base64.b64encode(data).decode("ascii") +def base64_trim(s: str) -> str: + return "".join(c for c in s if c not in "\n\r \t") + + +def b64d(s: str) -> bytes: + return base64.b64decode(base64_trim(s)) + + class AtClient: def __init__(self, device: str, baud: int, timeout: float) -> None: self.channel: str | None = None @@ -222,12 +239,37 @@ def find_tag(data: bytes, target: int) -> bytes | None: return next((v for t, v in iter_tlv(data) if t == target), None) +def require_tag(data: bytes, target: int, label: str = "") -> bytes: + v = find_tag(data, target) + if v is None: + raise RuntimeError(f"Missing {label or f'tag 0x{target:X}'}") + return v + + def tbcd_to_string(raw: bytes) -> str: return "".join(str(n) for b in raw for n in (b & 0x0F, b >> 4) if n <= 9) -# Profile field decoders: TLV tag -> (field_name, decoder) -_PROFILE_FIELDS = { +def string_to_tbcd(s: str) -> bytes: + digits = [int(c) for c in s if c.isdigit()] + return bytes(digits[i] | ((digits[i + 1] if i + 1 < len(digits) else 0xF) << 4) for i in range(0, len(digits), 2)) + + +def encode_tlv(tag: int, value: bytes) -> bytes: + tag_bytes = bytes([(tag >> 8) & 0xFF, tag & 0xFF]) if tag > 255 else bytes([tag]) + vlen = len(value) + if vlen <= 127: + return tag_bytes + bytes([vlen]) + value + length_bytes = vlen.to_bytes((vlen.bit_length() + 7) // 8, "big") + return tag_bytes + bytes([0x80 | len(length_bytes)]) + length_bytes + value + + +def int_bytes(n: int) -> bytes: + """Encode a positive integer as minimal big-endian bytes (at least 1 byte).""" + return n.to_bytes((n.bit_length() + 7) // 8 or 1, "big") + + +PROFILE: FieldMap = { TAG_ICCID: ("iccid", tbcd_to_string), 0x4F: ("isdpAid", lambda v: v.hex().upper()), 0x9F70: ("profileState", lambda v: STATE_LABELS.get(v[0], "unknown")), @@ -240,11 +282,11 @@ _PROFILE_FIELDS = { } -def _decode_profile_fields(data: bytes) -> dict: - """Parse known profile metadata TLV fields into a dict.""" - result = {} +def decode_struct(data: bytes, field_map: FieldMap) -> dict[str, Any]: + """Parse TLV data using a {tag: (field_name, decoder)} map into a dict.""" + result: dict[str, Any] = {name: None for name, _ in field_map.values()} for tag, value in iter_tlv(data): - if (field := _PROFILE_FIELDS.get(tag)): + if (field := field_map.get(tag)): result[field[0]] = field[1](value) return result @@ -277,14 +319,11 @@ def es10x_command(client: AtClient, data: bytes) -> bytes: # --- Profile operations --- def decode_profiles(blob: bytes) -> list[dict]: - root = find_tag(blob, TAG_PROFILE_INFO_LIST) - if root is None: - raise RuntimeError("Missing ProfileInfoList") - list_ok = find_tag(root, 0xA0) + root = require_tag(blob, TAG_PROFILE_INFO_LIST, "ProfileInfoList") + list_ok = find_tag(root, TAG_OK) if list_ok is None: return [] - defaults = {name: None for name, _ in _PROFILE_FIELDS.values()} - return [{**defaults, **_decode_profile_fields(value)} for tag, value in iter_tlv(list_ok) if tag == 0xE3] + return [decode_struct(value, PROFILE) for tag, value in iter_tlv(list_ok) if tag == 0xE3] def list_profiles(client: AtClient) -> list[dict]: @@ -292,29 +331,40 @@ def list_profiles(client: AtClient) -> list[dict]: class TiciLPA(LPABase): - _instance = None - - def __new__(cls): - if cls._instance is None: - cls._instance = super().__new__(cls) - return cls._instance - def __init__(self): if hasattr(self, '_client'): return self._client = AtClient(DEFAULT_DEVICE, DEFAULT_BAUD, DEFAULT_TIMEOUT) atexit.register(self._client.close) + @contextmanager + def _acquire_channel(self): + fd = os.open(LOCK_FILE, os.O_CREAT | os.O_RDWR) + try: + fcntl.flock(fd, fcntl.LOCK_EX) + self._client.open_isdr() + yield + finally: + if self._client.channel: + try: + self._client.query(f"AT+CCHC={self._client.channel}") + except (RuntimeError, TimeoutError): + pass + self._client.channel = None + fcntl.flock(fd, fcntl.LOCK_UN) + os.close(fd) + def list_profiles(self) -> list[Profile]: - return [ - Profile( - iccid=p.get("iccid", ""), - nickname=p.get("profileNickname") or "", - enabled=p.get("profileState") == "enabled", - provider=p.get("serviceProviderName") or "", - ) - for p in list_profiles(self._client) - ] + with self._acquire_channel(): + return [ + Profile( + iccid=p.get("iccid", ""), + nickname=p.get("profileNickname") or "", + enabled=p.get("profileState") == "enabled", + provider=p.get("serviceProviderName") or "", + ) + for p in list_profiles(self._client) + ] def get_active_profile(self) -> Profile | None: return None