241 lines
7.5 KiB
Python
241 lines
7.5 KiB
Python
import os
|
|
import subprocess
|
|
from pathlib import Path
|
|
from smbus2 import SMBus
|
|
|
|
from cereal import log
|
|
from selfdrive.hardware.base import HardwareBase, ThermalConfig
|
|
|
|
NM = 'org.freedesktop.NetworkManager'
|
|
NM_CON_ACT = NM + '.Connection.Active'
|
|
NM_DEV_WL = NM + '.Device.Wireless'
|
|
NM_AP = NM + '.AccessPoint'
|
|
DBUS_PROPS = 'org.freedesktop.DBus.Properties'
|
|
|
|
MM = 'org.freedesktop.ModemManager1'
|
|
MM_MODEM = MM + ".Modem"
|
|
MM_MODEM_SIMPLE = MM + ".Modem.Simple"
|
|
MM_SIM = MM + ".Sim"
|
|
|
|
MM_MODEM_STATE_CONNECTED = 11
|
|
|
|
TIMEOUT = 0.1
|
|
|
|
NetworkType = log.DeviceState.NetworkType
|
|
NetworkStrength = log.DeviceState.NetworkStrength
|
|
|
|
# https://developer.gnome.org/ModemManager/unstable/ModemManager-Flags-and-Enumerations.html#MMModemAccessTechnology
|
|
MM_MODEM_ACCESS_TECHNOLOGY_UMTS = 1 << 5
|
|
MM_MODEM_ACCESS_TECHNOLOGY_LTE = 1 << 14
|
|
|
|
AMP_I2C_BUS = 0
|
|
AMP_ADDRESS = 0x10
|
|
|
|
def write_amplifier_reg(reg, val, offset, mask):
|
|
with SMBus(AMP_I2C_BUS) as bus:
|
|
v = bus.read_byte_data(AMP_ADDRESS, reg, force=True)
|
|
v = (v & (~mask)) | ((val << offset) & mask)
|
|
bus.write_byte_data(AMP_ADDRESS, reg, v, force=True)
|
|
|
|
class Tici(HardwareBase):
|
|
def __init__(self):
|
|
import dbus # pylint: disable=import-error
|
|
|
|
self.bus = dbus.SystemBus()
|
|
self.nm = self.bus.get_object(NM, '/org/freedesktop/NetworkManager')
|
|
self.mm = self.bus.get_object(MM, '/org/freedesktop/ModemManager1')
|
|
|
|
def get_os_version(self):
|
|
with open("/VERSION") as f:
|
|
return f.read().strip()
|
|
|
|
def get_device_type(self):
|
|
return "tici"
|
|
|
|
def get_sound_card_online(self):
|
|
return (os.path.isfile('/proc/asound/card0/state') and
|
|
open('/proc/asound/card0/state').read().strip() == 'ONLINE')
|
|
|
|
def reboot(self, reason=None):
|
|
subprocess.check_output(["sudo", "reboot"])
|
|
|
|
def uninstall(self):
|
|
Path("/data/__system_reset__").touch()
|
|
os.sync()
|
|
self.reboot()
|
|
|
|
def get_serial(self):
|
|
return self.get_cmdline()['androidboot.serialno']
|
|
|
|
def get_network_type(self):
|
|
try:
|
|
primary_connection = self.nm.Get(NM, 'PrimaryConnection', dbus_interface=DBUS_PROPS, timeout=TIMEOUT)
|
|
primary_connection = self.bus.get_object(NM, primary_connection)
|
|
tp = primary_connection.Get(NM_CON_ACT, 'Type', dbus_interface=DBUS_PROPS, timeout=TIMEOUT)
|
|
|
|
if tp in ['802-3-ethernet', '802-11-wireless']:
|
|
return NetworkType.wifi
|
|
elif tp in ['gsm']:
|
|
modem = self.get_modem()
|
|
access_t = modem.Get(MM_MODEM, 'AccessTechnologies', dbus_interface=DBUS_PROPS, timeout=TIMEOUT)
|
|
if access_t >= MM_MODEM_ACCESS_TECHNOLOGY_LTE:
|
|
return NetworkType.cell4G
|
|
elif access_t >= MM_MODEM_ACCESS_TECHNOLOGY_UMTS:
|
|
return NetworkType.cell3G
|
|
else:
|
|
return NetworkType.cell2G
|
|
except Exception:
|
|
pass
|
|
|
|
return NetworkType.none
|
|
|
|
def get_modem(self):
|
|
objects = self.mm.GetManagedObjects(dbus_interface="org.freedesktop.DBus.ObjectManager", timeout=TIMEOUT)
|
|
modem_path = list(objects.keys())[0]
|
|
return self.bus.get_object(MM, modem_path)
|
|
|
|
def get_wlan(self):
|
|
wlan_path = self.nm.GetDeviceByIpIface('wlan0', dbus_interface=NM, timeout=TIMEOUT)
|
|
return self.bus.get_object(NM, wlan_path)
|
|
|
|
def get_sim_info(self):
|
|
modem = self.get_modem()
|
|
sim_path = modem.Get(MM_MODEM, 'Sim', dbus_interface=DBUS_PROPS, timeout=TIMEOUT)
|
|
|
|
if sim_path == "/":
|
|
return {
|
|
'sim_id': '',
|
|
'mcc_mnc': None,
|
|
'network_type': ["Unknown"],
|
|
'sim_state': ["ABSENT"],
|
|
'data_connected': False
|
|
}
|
|
else:
|
|
sim = self.bus.get_object(MM, sim_path)
|
|
return {
|
|
'sim_id': str(sim.Get(MM_SIM, 'SimIdentifier', dbus_interface=DBUS_PROPS, timeout=TIMEOUT)),
|
|
'mcc_mnc': str(sim.Get(MM_SIM, 'OperatorIdentifier', dbus_interface=DBUS_PROPS, timeout=TIMEOUT)),
|
|
'network_type': ["Unknown"],
|
|
'sim_state': ["READY"],
|
|
'data_connected': modem.Get(MM_MODEM, 'State', dbus_interface=DBUS_PROPS, timeout=TIMEOUT) == MM_MODEM_STATE_CONNECTED,
|
|
}
|
|
|
|
def get_subscriber_info(self):
|
|
return ""
|
|
|
|
def get_imei(self, slot):
|
|
if slot != 0:
|
|
return ""
|
|
|
|
return str(self.get_modem().Get(MM_MODEM, 'EquipmentIdentifier', dbus_interface=DBUS_PROPS, timeout=TIMEOUT))
|
|
|
|
def get_network_info(self):
|
|
modem = self.get_modem()
|
|
try:
|
|
info = modem.Command("AT+QNWINFO", int(TIMEOUT * 1000), dbus_interface=MM_MODEM, timeout=TIMEOUT)
|
|
except Exception:
|
|
return None
|
|
|
|
if info and info.startswith('+QNWINFO: '):
|
|
info = info.replace('+QNWINFO: ', '').replace('"', '').split(',')
|
|
|
|
if len(info) != 4:
|
|
return None
|
|
|
|
technology, operator, band, channel = info
|
|
|
|
return({
|
|
'technology': technology,
|
|
'operator': operator,
|
|
'band': band,
|
|
'channel': int(channel)
|
|
})
|
|
else:
|
|
return None
|
|
|
|
def parse_strength(self, percentage):
|
|
if percentage < 25:
|
|
return NetworkStrength.poor
|
|
elif percentage < 50:
|
|
return NetworkStrength.moderate
|
|
elif percentage < 75:
|
|
return NetworkStrength.good
|
|
else:
|
|
return NetworkStrength.great
|
|
|
|
def get_network_strength(self, network_type):
|
|
network_strength = NetworkStrength.unknown
|
|
|
|
try:
|
|
if network_type == NetworkType.none:
|
|
pass
|
|
elif network_type == NetworkType.wifi:
|
|
wlan = self.get_wlan()
|
|
active_ap_path = wlan.Get(NM_DEV_WL, 'ActiveAccessPoint', dbus_interface=DBUS_PROPS, timeout=TIMEOUT)
|
|
if active_ap_path != "/":
|
|
active_ap = self.bus.get_object(NM, active_ap_path)
|
|
strength = int(active_ap.Get(NM_AP, 'Strength', dbus_interface=DBUS_PROPS, timeout=TIMEOUT))
|
|
network_strength = self.parse_strength(strength)
|
|
else: # Cellular
|
|
modem = self.get_modem()
|
|
strength = int(modem.Get(MM_MODEM, 'SignalQuality', dbus_interface=DBUS_PROPS, timeout=TIMEOUT)[0])
|
|
network_strength = self.parse_strength(strength)
|
|
except Exception:
|
|
pass
|
|
|
|
return network_strength
|
|
|
|
# We don't have a battery, so let's use some sane constants
|
|
def get_battery_capacity(self):
|
|
return 100
|
|
|
|
def get_battery_status(self):
|
|
return ""
|
|
|
|
def get_battery_current(self):
|
|
return 0
|
|
|
|
def get_battery_voltage(self):
|
|
return 0
|
|
|
|
def get_battery_charging(self):
|
|
return True
|
|
|
|
def set_battery_charging(self, on):
|
|
pass
|
|
|
|
def get_usb_present(self):
|
|
# Not sure if relevant on tici, but the file exists
|
|
return self.read_param_file("/sys/class/power_supply/usb/present", lambda x: bool(int(x)), False)
|
|
|
|
def get_current_power_draw(self):
|
|
return (self.read_param_file("/sys/class/hwmon/hwmon1/power1_input", int) / 1e6)
|
|
|
|
def shutdown(self):
|
|
# Note that for this to work and have the device stay powered off, the panda needs to be in UsbPowerMode::CLIENT!
|
|
os.system("sudo poweroff")
|
|
|
|
def get_thermal_config(self):
|
|
return ThermalConfig(cpu=((1, 2, 3, 4, 5, 6, 7, 8), 1000), gpu=((48,49), 1000), mem=(15, 1000), bat=(None, 1), ambient=(65, 1000))
|
|
|
|
def set_screen_brightness(self, percentage):
|
|
try:
|
|
with open("/sys/class/backlight/panel0-backlight/brightness", "w") as f:
|
|
f.write(str(int(percentage * 10.23)))
|
|
except Exception:
|
|
pass
|
|
|
|
def set_power_save(self, enabled):
|
|
# amplifier, 100mW at idle
|
|
write_amplifier_reg(0x51, 0b0 if enabled else 0b1, 7, 0b10000000)
|
|
|
|
# offline big cluster, leave core 4 online for boardd
|
|
for i in range(5, 8):
|
|
# TODO: fix permissions with udev
|
|
val = "0" if enabled else "1"
|
|
os.system(f"sudo su -c 'echo {val} > /sys/devices/system/cpu/cpu{i}/online'")
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
Tici().set_power_save(bool(int(sys.argv[1])))
|