esim: TLV helpers, acquire_channel, and TiciLPA rework (#37776)

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