replace smbus2 package with minimal implementation (#37061)

* replace smbus2 package with minimal implementation

* cleanup

* fix up

Author: adeebshihadeh

common/i2c.py

@@ -0,0 +1,81 @@
+import os
+import fcntl
+import ctypes
+
+# I2C constants from /usr/include/linux/i2c-dev.h
+I2C_SLAVE = 0x0703
+I2C_SLAVE_FORCE = 0x0706
+I2C_SMBUS = 0x0720
+
+# SMBus transfer types
+I2C_SMBUS_READ = 1
+I2C_SMBUS_WRITE = 0
+I2C_SMBUS_BYTE_DATA = 2
+I2C_SMBUS_I2C_BLOCK_DATA = 8
+
+I2C_SMBUS_BLOCK_MAX = 32
+
+
+class _I2cSmbusData(ctypes.Union):
+  _fields_ = [
+    ("byte", ctypes.c_uint8),
+    ("word", ctypes.c_uint16),
+    ("block", ctypes.c_uint8 * (I2C_SMBUS_BLOCK_MAX + 2)),
+  ]
+
+
+class _I2cSmbusIoctlData(ctypes.Structure):
+  _fields_ = [
+    ("read_write", ctypes.c_uint8),
+    ("command", ctypes.c_uint8),
+    ("size", ctypes.c_uint32),
+    ("data", ctypes.POINTER(_I2cSmbusData)),
+  ]
+
+
+class SMBus:
+  def __init__(self, bus: int):
+    self._fd = os.open(f'/dev/i2c-{bus}', os.O_RDWR)
+
+  def __enter__(self) -> 'SMBus':
+    return self
+
+  def __exit__(self, *args) -> None:
+    self.close()
+
+  def close(self) -> None:
+    if hasattr(self, '_fd') and self._fd >= 0:
+      os.close(self._fd)
+      self._fd = -1
+
+  def _set_address(self, addr: int, force: bool = False) -> None:
+    ioctl_arg = I2C_SLAVE_FORCE if force else I2C_SLAVE
+    fcntl.ioctl(self._fd, ioctl_arg, addr)
+
+  def _smbus_access(self, read_write: int, command: int, size: int, data: _I2cSmbusData) -> None:
+    ioctl_data = _I2cSmbusIoctlData(read_write, command, size, ctypes.pointer(data))
+    fcntl.ioctl(self._fd, I2C_SMBUS, ioctl_data)
+
+  def read_byte_data(self, addr: int, register: int, force: bool = False) -> int:
+    self._set_address(addr, force)
+    data = _I2cSmbusData()
+    self._smbus_access(I2C_SMBUS_READ, register, I2C_SMBUS_BYTE_DATA, data)
+    return int(data.byte)
+
+  def write_byte_data(self, addr: int, register: int, value: int, force: bool = False) -> None:
+    self._set_address(addr, force)
+    data = _I2cSmbusData()
+    data.byte = value & 0xFF
+    self._smbus_access(I2C_SMBUS_WRITE, register, I2C_SMBUS_BYTE_DATA, data)
+
+  def read_i2c_block_data(self, addr: int, register: int, length: int, force: bool = False) -> list[int]:
+    self._set_address(addr, force)
+    if not (0 <= length <= I2C_SMBUS_BLOCK_MAX):
+      raise ValueError(f"length must be 0..{I2C_SMBUS_BLOCK_MAX}")
+
+    data = _I2cSmbusData()
+    data.block[0] = length
+    self._smbus_access(I2C_SMBUS_READ, register, I2C_SMBUS_I2C_BLOCK_DATA, data)
+    read_len = int(data.block[0]) or length
+    read_len = min(read_len, length)
+    return [int(b) for b in data.block[1 : read_len + 1]]

pyproject.toml

@@ -17,9 +17,6 @@ dependencies = [
   "crcmod-plus",  # cars + qcomgpsd
   "tqdm",         # cars (fw_versions.py) on start + many one-off uses
 
-  # hardwared
-  "smbus2", # configuring amp
-
   # core
   "cffi",
   "scons",

system/hardware/tici/amplifier.py

@@ -1,8 +1,9 @@
 #!/usr/bin/env python3
 import time
-from smbus2 import SMBus
 from collections import namedtuple
 
+from openpilot.common.i2c import SMBus
+
 # https://datasheets.maximintegrated.com/en/ds/MAX98089.pdf
 
 AmpConfig = namedtuple('AmpConfig', ['name', 'value', 'register', 'offset', 'mask'])

system/sensord/sensors/i2c_sensor.py

@@ -1,9 +1,10 @@
 import time
-import smbus2
 import ctypes
 from collections.abc import Iterable
 
 from cereal import log
+from openpilot.common.i2c import SMBus
+
 
 class Sensor:
   class SensorException(Exception):
@@ -13,7 +14,7 @@ class DataNotReady(SensorException):
     pass
 
   def __init__(self, bus: int) -> None:
-    self.bus = smbus2.SMBus(bus)
+    self.bus = SMBus(bus)
     self.source = log.SensorEventData.SensorSource.velodyne  # unknown
     self.start_ts = 0.