Merge branch commaai/openpilot:master into branch sunnypilot/sunnypilot:master-new

This commit is contained in:
discountchubbs
2025-05-25 09:09:02 -07:00
committed by Jason Wen
207 changed files with 4167 additions and 1948 deletions
+1 -1
View File
@@ -565,7 +565,7 @@ def getNetworks():
@dispatcher.add_method
def takeSnapshot() -> str | dict[str, str] | None:
from openpilot.system.camerad.snapshot.snapshot import jpeg_write, snapshot
from openpilot.system.camerad.snapshot import jpeg_write, snapshot
ret = snapshot()
if ret is not None:
def b64jpeg(x):
View File
-27
View File
@@ -1,27 +0,0 @@
#!/usr/bin/env python3
# type: ignore
import cereal.messaging as messaging
all_sockets = ['roadCameraState', 'driverCameraState', 'wideRoadCameraState']
prev_id = [None,None,None]
this_id = [None,None,None]
dt = [None,None,None]
num_skipped = [0,0,0]
if __name__ == "__main__":
sm = messaging.SubMaster(all_sockets)
while True:
sm.update()
for i in range(len(all_sockets)):
if not sm.updated[all_sockets[i]]:
continue
this_id[i] = sm[all_sockets[i]].frameId
if prev_id[i] is None:
prev_id[i] = this_id[i]
continue
dt[i] = this_id[i] - prev_id[i]
if dt[i] != 1:
num_skipped[i] += dt[i] - 1
print(all_sockets[i] ,dt[i] - 1, num_skipped[i])
prev_id[i] = this_id[i]
@@ -1,24 +0,0 @@
#!/usr/bin/env python3
import argparse
import os
from tqdm import tqdm
from openpilot.tools.lib.logreader import LogReader
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("route", help="The route name")
args = parser.parse_args()
out_path = os.path.join("jpegs", f"{args.route.replace('|', '_').replace('/', '_')}")
os.makedirs(out_path, exist_ok=True)
lr = LogReader(args.route)
for msg in tqdm(lr):
if msg.which() == 'thumbnail':
with open(os.path.join(out_path, f"{msg.thumbnail.frameId}.jpg"), 'wb') as f:
f.write(msg.thumbnail.thumbnail)
elif msg.which() == 'navThumbnail':
with open(os.path.join(out_path, f"nav_{msg.navThumbnail.frameId}.jpg"), 'wb') as f:
f.write(msg.navThumbnail.thumbnail)
+1 -1
View File
@@ -3,7 +3,7 @@ import numpy as np
import pytest
from openpilot.selfdrive.test.helpers import with_processes
from openpilot.system.camerad.snapshot.snapshot import get_snapshots
from openpilot.system.camerad.snapshot import get_snapshots
TEST_TIME = 45
REPEAT = 5
+43
View File
@@ -6,6 +6,19 @@ from cereal import log
NetworkType = log.DeviceState.NetworkType
class LPAError(RuntimeError):
pass
class LPAProfileNotFoundError(LPAError):
pass
@dataclass
class Profile:
iccid: str
nickname: str
enabled: bool
provider: str
@dataclass
class ThermalZone:
# a zone from /sys/class/thermal/thermal_zone*
@@ -33,6 +46,7 @@ class ThermalZone:
class ThermalConfig:
cpu: list[ThermalZone] | None = None
gpu: list[ThermalZone] | None = None
dsp: ThermalZone | None = None
pmic: list[ThermalZone] | None = None
memory: ThermalZone | None = None
intake: ThermalZone | None = None
@@ -50,6 +64,31 @@ class ThermalConfig:
ret[f.name + "TempC"] = v.read()
return ret
class LPABase(ABC):
@abstractmethod
def list_profiles(self) -> list[Profile]:
pass
@abstractmethod
def get_active_profile(self) -> Profile | None:
pass
@abstractmethod
def delete_profile(self, iccid: str) -> None:
pass
@abstractmethod
def download_profile(self, qr: str, nickname: str | None = None) -> None:
pass
@abstractmethod
def nickname_profile(self, iccid: str, nickname: str) -> None:
pass
@abstractmethod
def switch_profile(self, iccid: str) -> None:
pass
class HardwareBase(ABC):
@staticmethod
def get_cmdline() -> dict[str, str]:
@@ -104,6 +143,10 @@ class HardwareBase(ABC):
def get_sim_info(self):
pass
@abstractmethod
def get_sim_lpa(self) -> LPABase:
pass
@abstractmethod
def get_network_strength(self, network_type):
pass
+38
View File
@@ -0,0 +1,38 @@
#!/usr/bin/env python3
import argparse
from openpilot.system.hardware import HARDWARE
if __name__ == '__main__':
parser = argparse.ArgumentParser(prog='esim.py', description='manage eSIM profiles on your comma device', epilog='comma.ai')
parser.add_argument('--backend', choices=['qmi', 'at'], default='qmi', help='use the specified backend, defaults to qmi')
parser.add_argument('--switch', metavar='iccid', help='switch to profile')
parser.add_argument('--delete', metavar='iccid', help='delete profile (warning: this cannot be undone)')
parser.add_argument('--download', nargs=2, metavar=('qr', 'name'), help='download a profile using QR code (format: LPA:1$rsp.truphone.com$QRF-SPEEDTEST)')
parser.add_argument('--nickname', nargs=2, metavar=('iccid', 'name'), help='update the nickname for a profile')
args = parser.parse_args()
lpa = HARDWARE.get_sim_lpa()
if args.switch:
lpa.switch_profile(args.switch)
elif args.delete:
confirm = input('are you sure you want to delete this profile? (y/N) ')
if confirm == 'y':
lpa.delete_profile(args.delete)
print('deleted profile, please restart device to apply changes')
else:
print('cancelled')
exit(0)
elif args.download:
lpa.download_profile(args.download[0], args.download[1])
elif args.nickname:
lpa.nickname_profile(args.nickname[0], args.nickname[1])
else:
parser.print_help()
profiles = lpa.list_profiles()
print(f'\n{len(profiles)} profile{"s" if len(profiles) > 1 else ""}:')
for p in profiles:
print(f'- {p.iccid} (nickname: {p.nickname or "<none provided>"}) (provider: {p.provider}) - {"enabled" if p.enabled else "disabled"}')
+4 -2
View File
@@ -1,12 +1,11 @@
import random
from cereal import log
from openpilot.system.hardware.base import HardwareBase
from openpilot.system.hardware.base import HardwareBase, LPABase
NetworkType = log.DeviceState.NetworkType
NetworkStrength = log.DeviceState.NetworkStrength
class Pc(HardwareBase):
def get_os_version(self):
return None
@@ -41,6 +40,9 @@ class Pc(HardwareBase):
'data_connected': False
}
def get_sim_lpa(self) -> LPABase:
raise NotImplementedError("SIM LPA not implemented for PC")
def get_network_strength(self, network_type):
return NetworkStrength.unknown
+26 -26
View File
@@ -1,25 +1,25 @@
[
{
"name": "xbl",
"url": "https://commadist.azureedge.net/agnosupdate/xbl-468f1ad6ab55e198647ff9191f91bd2918db9c0a3e27bae5673b4c5575c1254c.img.xz",
"hash": "468f1ad6ab55e198647ff9191f91bd2918db9c0a3e27bae5673b4c5575c1254c",
"hash_raw": "468f1ad6ab55e198647ff9191f91bd2918db9c0a3e27bae5673b4c5575c1254c",
"url": "https://commadist.azureedge.net/agnosupdate/xbl-6710967ca9701f205d7ab19c3a9b0dd2f547e65b3d96048b7c2b03755aafa0f1.img.xz",
"hash": "6710967ca9701f205d7ab19c3a9b0dd2f547e65b3d96048b7c2b03755aafa0f1",
"hash_raw": "6710967ca9701f205d7ab19c3a9b0dd2f547e65b3d96048b7c2b03755aafa0f1",
"size": 3282256,
"sparse": false,
"full_check": true,
"has_ab": true,
"ondevice_hash": "d35a86e7b8ddd9279b513a6f27da1521aa0f89fb93987ea74d57d0f0bbbbd247"
"ondevice_hash": "003a17ab1be68a696f7efe4c9938e8be511d4aacfc2f3211fc896bdc1681d089"
},
{
"name": "xbl_config",
"url": "https://commadist.azureedge.net/agnosupdate/xbl_config-92b675dc2862ed15c732d91d9eb307d7e852e349217db8bee8f8829db543686b.img.xz",
"hash": "92b675dc2862ed15c732d91d9eb307d7e852e349217db8bee8f8829db543686b",
"hash_raw": "92b675dc2862ed15c732d91d9eb307d7e852e349217db8bee8f8829db543686b",
"url": "https://commadist.azureedge.net/agnosupdate/xbl_config-63922cfbfdf4ab87986c4ba8f3a4df5bf28414b3f71a29ec5947336722215535.img.xz",
"hash": "63922cfbfdf4ab87986c4ba8f3a4df5bf28414b3f71a29ec5947336722215535",
"hash_raw": "63922cfbfdf4ab87986c4ba8f3a4df5bf28414b3f71a29ec5947336722215535",
"size": 98124,
"sparse": false,
"full_check": true,
"has_ab": true,
"ondevice_hash": "623f1568072ee2d687ba8449a3d894c1c83dc4131b2e79eff35696885f70a419"
"ondevice_hash": "2a855dd636cc94718b64bea83a44d0a31741ecaa8f72a63613ff348ec7404091"
},
{
"name": "abl",
@@ -34,50 +34,50 @@
},
{
"name": "aop",
"url": "https://commadist.azureedge.net/agnosupdate/aop-f0fcf7611d0890a72984f15a516dd37fa532dfcb70d428a8406838cf74ce23d5.img.xz",
"hash": "f0fcf7611d0890a72984f15a516dd37fa532dfcb70d428a8406838cf74ce23d5",
"hash_raw": "f0fcf7611d0890a72984f15a516dd37fa532dfcb70d428a8406838cf74ce23d5",
"url": "https://commadist.azureedge.net/agnosupdate/aop-21370172e590bd4ea907a558bcd6df20dc7a6c7d38b8e62fdde18f4a512ba9e9.img.xz",
"hash": "21370172e590bd4ea907a558bcd6df20dc7a6c7d38b8e62fdde18f4a512ba9e9",
"hash_raw": "21370172e590bd4ea907a558bcd6df20dc7a6c7d38b8e62fdde18f4a512ba9e9",
"size": 184364,
"sparse": false,
"full_check": true,
"has_ab": true,
"ondevice_hash": "bf74feca486f650589f6b7c90eab73274e35a68b5e00bfc1de0ed5f5484d4b3d"
"ondevice_hash": "c1be2f4aac5b3af49b904b027faec418d05efd7bd5144eb4fdfcba602bcf2180"
},
{
"name": "devcfg",
"url": "https://commadist.azureedge.net/agnosupdate/devcfg-225b24ea7b1d2fee7f7d2da21386920ddacac2e33e9e938168436292f4eae180.img.xz",
"hash": "225b24ea7b1d2fee7f7d2da21386920ddacac2e33e9e938168436292f4eae180",
"hash_raw": "225b24ea7b1d2fee7f7d2da21386920ddacac2e33e9e938168436292f4eae180",
"url": "https://commadist.azureedge.net/agnosupdate/devcfg-d7d7e52963bbedbbf8a7e66847579ca106a0a729ce2cf60f4b8d8ea4b535d620.img.xz",
"hash": "d7d7e52963bbedbbf8a7e66847579ca106a0a729ce2cf60f4b8d8ea4b535d620",
"hash_raw": "d7d7e52963bbedbbf8a7e66847579ca106a0a729ce2cf60f4b8d8ea4b535d620",
"size": 40336,
"sparse": false,
"full_check": true,
"has_ab": true,
"ondevice_hash": "70f682b59ca0fe2f197d1486bd8be7b9b7e560798ad40ddef83b9f0a2f497938"
"ondevice_hash": "17b229668b20305ff8fa3cd5f94716a3aaa1e5bf9d1c24117eff7f2f81ae719f"
},
{
"name": "boot",
"url": "https://commadist.azureedge.net/agnosupdate/boot-3d8e848796924081f5a6b3d745808b1117ae2ec41c03f2d41ee2e75633bd6425.img.xz",
"hash": "3d8e848796924081f5a6b3d745808b1117ae2ec41c03f2d41ee2e75633bd6425",
"hash_raw": "3d8e848796924081f5a6b3d745808b1117ae2ec41c03f2d41ee2e75633bd6425",
"url": "https://commadist.azureedge.net/agnosupdate/boot-eb0dd0a8ffdc1c03aa28d055788d70be885a00f523b552d7acf79b954838e968.img.xz",
"hash": "eb0dd0a8ffdc1c03aa28d055788d70be885a00f523b552d7acf79b954838e968",
"hash_raw": "eb0dd0a8ffdc1c03aa28d055788d70be885a00f523b552d7acf79b954838e968",
"size": 18479104,
"sparse": false,
"full_check": true,
"has_ab": true,
"ondevice_hash": "2075104847d1c96a06f07e85efb9f48d0e792d75a059047eae7ba4b463ffeadf"
"ondevice_hash": "800868bd9d340f1fdf8340924caca374409624658324607621663fc5c7d10d4f"
},
{
"name": "system",
"url": "https://commadist.azureedge.net/agnosupdate/system-b22bd239c6f9527596fd49b98b7d521a563f99b90953ce021ee36567498e99c7.img.xz",
"hash": "cb9bfde1e995b97f728f5d5ad8d7a0f7a01544db5d138ead9b2350f222640939",
"hash_raw": "b22bd239c6f9527596fd49b98b7d521a563f99b90953ce021ee36567498e99c7",
"url": "https://commadist.azureedge.net/agnosupdate/system-4c3fa88833a8a3876653fa53a658769ecc64be8fc6f6a28ce520ef79bf3061c1.img.xz",
"hash": "226794914e9d157b34cc86f1fbe1f2827a9a9919dbe560021b61573a7e5d3101",
"hash_raw": "4c3fa88833a8a3876653fa53a658769ecc64be8fc6f6a28ce520ef79bf3061c1",
"size": 5368709120,
"sparse": true,
"full_check": false,
"has_ab": true,
"ondevice_hash": "e92a1f34158c60364c8d47b8ebbb6e59edf8d4865cd5edfeb2355d6f54f617fc",
"ondevice_hash": "236890f04ee21b7eb92a59bc47971bd96cad5a4381ed8935eb4be0cb5a4cf48b",
"alt": {
"hash": "b22bd239c6f9527596fd49b98b7d521a563f99b90953ce021ee36567498e99c7",
"url": "https://commadist.azureedge.net/agnosupdate/system-b22bd239c6f9527596fd49b98b7d521a563f99b90953ce021ee36567498e99c7.img",
"hash": "4c3fa88833a8a3876653fa53a658769ecc64be8fc6f6a28ce520ef79bf3061c1",
"url": "https://commadist.azureedge.net/agnosupdate/system-4c3fa88833a8a3876653fa53a658769ecc64be8fc6f6a28ce520ef79bf3061c1.img",
"size": 5368709120
}
}
+38 -38
View File
@@ -130,25 +130,25 @@
},
{
"name": "xbl",
"url": "https://commadist.azureedge.net/agnosupdate/xbl-468f1ad6ab55e198647ff9191f91bd2918db9c0a3e27bae5673b4c5575c1254c.img.xz",
"hash": "468f1ad6ab55e198647ff9191f91bd2918db9c0a3e27bae5673b4c5575c1254c",
"hash_raw": "468f1ad6ab55e198647ff9191f91bd2918db9c0a3e27bae5673b4c5575c1254c",
"url": "https://commadist.azureedge.net/agnosupdate/xbl-6710967ca9701f205d7ab19c3a9b0dd2f547e65b3d96048b7c2b03755aafa0f1.img.xz",
"hash": "6710967ca9701f205d7ab19c3a9b0dd2f547e65b3d96048b7c2b03755aafa0f1",
"hash_raw": "6710967ca9701f205d7ab19c3a9b0dd2f547e65b3d96048b7c2b03755aafa0f1",
"size": 3282256,
"sparse": false,
"full_check": true,
"has_ab": true,
"ondevice_hash": "d35a86e7b8ddd9279b513a6f27da1521aa0f89fb93987ea74d57d0f0bbbbd247"
"ondevice_hash": "003a17ab1be68a696f7efe4c9938e8be511d4aacfc2f3211fc896bdc1681d089"
},
{
"name": "xbl_config",
"url": "https://commadist.azureedge.net/agnosupdate/xbl_config-92b675dc2862ed15c732d91d9eb307d7e852e349217db8bee8f8829db543686b.img.xz",
"hash": "92b675dc2862ed15c732d91d9eb307d7e852e349217db8bee8f8829db543686b",
"hash_raw": "92b675dc2862ed15c732d91d9eb307d7e852e349217db8bee8f8829db543686b",
"url": "https://commadist.azureedge.net/agnosupdate/xbl_config-63922cfbfdf4ab87986c4ba8f3a4df5bf28414b3f71a29ec5947336722215535.img.xz",
"hash": "63922cfbfdf4ab87986c4ba8f3a4df5bf28414b3f71a29ec5947336722215535",
"hash_raw": "63922cfbfdf4ab87986c4ba8f3a4df5bf28414b3f71a29ec5947336722215535",
"size": 98124,
"sparse": false,
"full_check": true,
"has_ab": true,
"ondevice_hash": "623f1568072ee2d687ba8449a3d894c1c83dc4131b2e79eff35696885f70a419"
"ondevice_hash": "2a855dd636cc94718b64bea83a44d0a31741ecaa8f72a63613ff348ec7404091"
},
{
"name": "abl",
@@ -163,14 +163,14 @@
},
{
"name": "aop",
"url": "https://commadist.azureedge.net/agnosupdate/aop-f0fcf7611d0890a72984f15a516dd37fa532dfcb70d428a8406838cf74ce23d5.img.xz",
"hash": "f0fcf7611d0890a72984f15a516dd37fa532dfcb70d428a8406838cf74ce23d5",
"hash_raw": "f0fcf7611d0890a72984f15a516dd37fa532dfcb70d428a8406838cf74ce23d5",
"url": "https://commadist.azureedge.net/agnosupdate/aop-21370172e590bd4ea907a558bcd6df20dc7a6c7d38b8e62fdde18f4a512ba9e9.img.xz",
"hash": "21370172e590bd4ea907a558bcd6df20dc7a6c7d38b8e62fdde18f4a512ba9e9",
"hash_raw": "21370172e590bd4ea907a558bcd6df20dc7a6c7d38b8e62fdde18f4a512ba9e9",
"size": 184364,
"sparse": false,
"full_check": true,
"has_ab": true,
"ondevice_hash": "bf74feca486f650589f6b7c90eab73274e35a68b5e00bfc1de0ed5f5484d4b3d"
"ondevice_hash": "c1be2f4aac5b3af49b904b027faec418d05efd7bd5144eb4fdfcba602bcf2180"
},
{
"name": "bluetooth",
@@ -207,14 +207,14 @@
},
{
"name": "devcfg",
"url": "https://commadist.azureedge.net/agnosupdate/devcfg-225b24ea7b1d2fee7f7d2da21386920ddacac2e33e9e938168436292f4eae180.img.xz",
"hash": "225b24ea7b1d2fee7f7d2da21386920ddacac2e33e9e938168436292f4eae180",
"hash_raw": "225b24ea7b1d2fee7f7d2da21386920ddacac2e33e9e938168436292f4eae180",
"url": "https://commadist.azureedge.net/agnosupdate/devcfg-d7d7e52963bbedbbf8a7e66847579ca106a0a729ce2cf60f4b8d8ea4b535d620.img.xz",
"hash": "d7d7e52963bbedbbf8a7e66847579ca106a0a729ce2cf60f4b8d8ea4b535d620",
"hash_raw": "d7d7e52963bbedbbf8a7e66847579ca106a0a729ce2cf60f4b8d8ea4b535d620",
"size": 40336,
"sparse": false,
"full_check": true,
"has_ab": true,
"ondevice_hash": "70f682b59ca0fe2f197d1486bd8be7b9b7e560798ad40ddef83b9f0a2f497938"
"ondevice_hash": "17b229668b20305ff8fa3cd5f94716a3aaa1e5bf9d1c24117eff7f2f81ae719f"
},
{
"name": "devinfo",
@@ -339,62 +339,62 @@
},
{
"name": "boot",
"url": "https://commadist.azureedge.net/agnosupdate/boot-3d8e848796924081f5a6b3d745808b1117ae2ec41c03f2d41ee2e75633bd6425.img.xz",
"hash": "3d8e848796924081f5a6b3d745808b1117ae2ec41c03f2d41ee2e75633bd6425",
"hash_raw": "3d8e848796924081f5a6b3d745808b1117ae2ec41c03f2d41ee2e75633bd6425",
"url": "https://commadist.azureedge.net/agnosupdate/boot-eb0dd0a8ffdc1c03aa28d055788d70be885a00f523b552d7acf79b954838e968.img.xz",
"hash": "eb0dd0a8ffdc1c03aa28d055788d70be885a00f523b552d7acf79b954838e968",
"hash_raw": "eb0dd0a8ffdc1c03aa28d055788d70be885a00f523b552d7acf79b954838e968",
"size": 18479104,
"sparse": false,
"full_check": true,
"has_ab": true,
"ondevice_hash": "2075104847d1c96a06f07e85efb9f48d0e792d75a059047eae7ba4b463ffeadf"
"ondevice_hash": "800868bd9d340f1fdf8340924caca374409624658324607621663fc5c7d10d4f"
},
{
"name": "system",
"url": "https://commadist.azureedge.net/agnosupdate/system-b22bd239c6f9527596fd49b98b7d521a563f99b90953ce021ee36567498e99c7.img.xz",
"hash": "cb9bfde1e995b97f728f5d5ad8d7a0f7a01544db5d138ead9b2350f222640939",
"hash_raw": "b22bd239c6f9527596fd49b98b7d521a563f99b90953ce021ee36567498e99c7",
"url": "https://commadist.azureedge.net/agnosupdate/system-4c3fa88833a8a3876653fa53a658769ecc64be8fc6f6a28ce520ef79bf3061c1.img.xz",
"hash": "226794914e9d157b34cc86f1fbe1f2827a9a9919dbe560021b61573a7e5d3101",
"hash_raw": "4c3fa88833a8a3876653fa53a658769ecc64be8fc6f6a28ce520ef79bf3061c1",
"size": 5368709120,
"sparse": true,
"full_check": false,
"has_ab": true,
"ondevice_hash": "e92a1f34158c60364c8d47b8ebbb6e59edf8d4865cd5edfeb2355d6f54f617fc",
"ondevice_hash": "236890f04ee21b7eb92a59bc47971bd96cad5a4381ed8935eb4be0cb5a4cf48b",
"alt": {
"hash": "b22bd239c6f9527596fd49b98b7d521a563f99b90953ce021ee36567498e99c7",
"url": "https://commadist.azureedge.net/agnosupdate/system-b22bd239c6f9527596fd49b98b7d521a563f99b90953ce021ee36567498e99c7.img",
"hash": "4c3fa88833a8a3876653fa53a658769ecc64be8fc6f6a28ce520ef79bf3061c1",
"url": "https://commadist.azureedge.net/agnosupdate/system-4c3fa88833a8a3876653fa53a658769ecc64be8fc6f6a28ce520ef79bf3061c1.img",
"size": 5368709120
}
},
{
"name": "userdata_90",
"url": "https://commadist.azureedge.net/agnosupdate/userdata_90-4bb7239f7e82c846e4d2584c0c433f03c582a80950de4094e6c190563d6d84ac.img.xz",
"hash": "b18001a2a87caa070fabf6321f8215ac353d6444564e3f86329b4dccc039ce54",
"hash_raw": "4bb7239f7e82c846e4d2584c0c433f03c582a80950de4094e6c190563d6d84ac",
"url": "https://commadist.azureedge.net/agnosupdate/userdata_90-f6d24876234f6bea9cc753892eea99ac4b0c8646e93b93d76fc5dddce67347f1.img.xz",
"hash": "2485459e9fbfc0d88a59a017bfa193d97b80afed30d6f59db711e53f62a21c72",
"hash_raw": "f6d24876234f6bea9cc753892eea99ac4b0c8646e93b93d76fc5dddce67347f1",
"size": 96636764160,
"sparse": true,
"full_check": true,
"has_ab": false,
"ondevice_hash": "15ce16f2349d5b4d5fec6ad1e36222b1ae744ed10b8930bc9af75bd244dccb3c"
"ondevice_hash": "893cfb3e95d6025b9fa6a5c8c3b97bf2abc4ff036c3da846a07536653dbb3a1d"
},
{
"name": "userdata_89",
"url": "https://commadist.azureedge.net/agnosupdate/userdata_89-e36b59bf9ff755b6ca488df2ba1e20da8f7dab6b8843129f3fdcccd7ff2ff7d8.img.xz",
"hash": "12682cf54596ab1bd1c2464c4ca85888e4e06b47af5ff7d0432399e9907e2f64",
"hash_raw": "e36b59bf9ff755b6ca488df2ba1e20da8f7dab6b8843129f3fdcccd7ff2ff7d8",
"url": "https://commadist.azureedge.net/agnosupdate/userdata_89-14f0f02e53ce62a79b337d7b3174dcaffebaa4130264bb5ed9105d942338230b.img.xz",
"hash": "9222aebb280531b7015ee9bab3848dd195567d075b6f37f74d4cd0186685a11f",
"hash_raw": "14f0f02e53ce62a79b337d7b3174dcaffebaa4130264bb5ed9105d942338230b",
"size": 95563022336,
"sparse": true,
"full_check": true,
"has_ab": false,
"ondevice_hash": "e4df9dea47ff04967d971263d50c17460ef240457e8d814e7c4f409f7493eb8a"
"ondevice_hash": "a5459ec858d39e3d0709a09af430cb644fe640f92e0cd216b138f01401e4e17e"
},
{
"name": "userdata_30",
"url": "https://commadist.azureedge.net/agnosupdate/userdata_30-fe1d86f5322c675c58b3ae9753a4670abf44a25746bf6ac822aed108bb577282.img.xz",
"hash": "fa471703be0f0647617d183312d5209d23407f1628e4ab0934e6ec54b1a6b263",
"hash_raw": "fe1d86f5322c675c58b3ae9753a4670abf44a25746bf6ac822aed108bb577282",
"url": "https://commadist.azureedge.net/agnosupdate/userdata_30-180ad331538f20c00218d41591afed3a25bb0f76fa30b1a665cad102cf6c9f7d.img.xz",
"hash": "7fc09317f005676150bfcced43ebb1aa919d854c2511d547a588a35c1122bfe3",
"hash_raw": "180ad331538f20c00218d41591afed3a25bb0f76fa30b1a665cad102cf6c9f7d",
"size": 32212254720,
"sparse": true,
"full_check": true,
"has_ab": false,
"ondevice_hash": "0b5b2402c9caa1ed7b832818e66580c974251e735bda91f2f226c41499d5616e"
"ondevice_hash": "1a2f2d14c922bd2895073446460a0bd680711a7b14318a4402631635e6d133b3"
}
]
Executable → Regular
+88 -92
View File
@@ -1,115 +1,111 @@
#!/usr/bin/env python3
import json
import os
import math
import time
import binascii
import requests
import serial
import shutil
import subprocess
from typing import Literal
from openpilot.system.hardware.base import LPABase, LPAError, LPAProfileNotFoundError, Profile
def post(url, payload):
print()
print("POST to", url)
r = requests.post(
url,
data=payload,
verify=False,
headers={
"Content-Type": "application/json",
"X-Admin-Protocol": "gsma/rsp/v2.2.0",
"charset": "utf-8",
"User-Agent": "gsma-rsp-lpad",
},
)
print("resp", r)
print("resp text", repr(r.text))
print()
r.raise_for_status()
class TiciLPA(LPABase):
def __init__(self, interface: Literal['qmi', 'at'] = 'qmi'):
self.env = os.environ.copy()
self.env['LPAC_APDU'] = interface
self.env['QMI_DEVICE'] = '/dev/cdc-wdm0'
self.env['AT_DEVICE'] = '/dev/ttyUSB2'
ret = f"HTTP/1.1 {r.status_code}"
ret += ''.join(f"{k}: {v}" for k, v in r.headers.items() if k != 'Connection')
return ret.encode() + r.content
self.timeout_sec = 45
if shutil.which('lpac') is None:
raise LPAError('lpac not found, must be installed!')
class LPA:
def __init__(self):
self.dev = serial.Serial('/dev/ttyUSB2', baudrate=57600, timeout=1, bytesize=8)
self.dev.reset_input_buffer()
self.dev.reset_output_buffer()
assert "OK" in self.at("AT")
def list_profiles(self) -> list[Profile]:
msgs = self._invoke('profile', 'list')
self._validate_successful(msgs)
return [Profile(
iccid=p['iccid'],
nickname=p['profileNickname'],
enabled=p['profileState'] == 'enabled',
provider=p['serviceProviderName']
) for p in msgs[-1]['payload']['data']]
def at(self, cmd):
print(f"==> {cmd}")
self.dev.write(cmd.encode() + b'\r\n')
def get_active_profile(self) -> Profile | None:
return next((p for p in self.list_profiles() if p.enabled), None)
r = b""
cnt = 0
while b"OK" not in r and b"ERROR" not in r and cnt < 20:
r += self.dev.read(8192).strip()
cnt += 1
r = r.decode()
print(f"<== {repr(r)}")
return r
def delete_profile(self, iccid: str) -> None:
self._validate_profile_exists(iccid)
latest = self.get_active_profile()
if latest is not None and latest.iccid == iccid:
raise LPAError('cannot delete active profile, switch to another profile first')
self._validate_successful(self._invoke('profile', 'delete', iccid))
self._process_notifications()
def download_ota(self, qr):
return self.at(f'AT+QESIM="ota","{qr}"')
def download_profile(self, qr: str, nickname: str | None = None) -> None:
msgs = self._invoke('profile', 'download', '-a', qr)
self._validate_successful(msgs)
new_profile = next((m for m in msgs if m['payload']['message'] == 'es8p_meatadata_parse'), None)
if new_profile is None:
raise LPAError('no new profile found')
if nickname:
self.nickname_profile(new_profile['payload']['data']['iccid'], nickname)
self._process_notifications()
def download(self, qr):
smdp = qr.split('$')[1]
out = self.at(f'AT+QESIM="download","{qr}"')
for _ in range(5):
line = out.split("+QESIM: ")[1].split("\r\n\r\nOK")[0]
def nickname_profile(self, iccid: str, nickname: str) -> None:
self._validate_profile_exists(iccid)
self._validate_successful(self._invoke('profile', 'nickname', iccid, nickname))
parts = [x.strip().strip('"') for x in line.split(',', maxsplit=4)]
print(repr(parts))
trans, ret, url, payloadlen, payload = parts
assert trans == "trans" and ret == "0"
assert len(payload) == int(payloadlen)
def switch_profile(self, iccid: str) -> None:
self._enable_profile(iccid)
r = post(f"https://{smdp}/{url}", payload)
to_send = binascii.hexlify(r).decode()
def _enable_profile(self, iccid: str) -> None:
self._validate_profile_exists(iccid)
latest = self.get_active_profile()
if latest:
if latest.iccid == iccid:
return
self._validate_successful(self._invoke('profile', 'disable', latest.iccid))
self._validate_successful(self._invoke('profile', 'enable', iccid))
self._process_notifications()
chunk_len = 1400
for i in range(math.ceil(len(to_send) / chunk_len)):
state = 1 if (i+1)*chunk_len < len(to_send) else 0
data = to_send[i * chunk_len : (i+1)*chunk_len]
out = self.at(f'AT+QESIM="trans",{len(to_send)},{state},{i},{len(data)},"{data}"')
assert "OK" in out
def _invoke(self, *cmd: str):
proc = subprocess.Popen(['sudo', '-E', 'lpac'] + list(cmd), stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=self.env)
try:
out, err = proc.communicate(timeout=self.timeout_sec)
except subprocess.TimeoutExpired as e:
proc.kill()
raise LPAError(f"lpac {cmd} timed out after {self.timeout_sec} seconds") from e
if '+QESIM:"download",1' in out:
raise Exception("profile install failed")
elif '+QESIM:"download",0' in out:
print("done, successfully loaded")
break
messages = []
for line in out.decode().strip().splitlines():
if line.startswith('{'):
message = json.loads(line)
def enable(self, iccid):
self.at(f'AT+QESIM="enable","{iccid}"')
# lpac response format validations
assert 'type' in message, 'expected type in message'
assert message['type'] == 'lpa' or message['type'] == 'progress', 'expected lpa or progress message type'
assert 'payload' in message, 'expected payload in message'
assert 'code' in message['payload'], 'expected code in message payload'
assert 'data' in message['payload'], 'expected data in message payload'
def disable(self, iccid):
self.at(f'AT+QESIM="disable","{iccid}"')
msg_ret_code = message['payload']['code']
if msg_ret_code != 0:
raise LPAError(f"lpac {' '.join(cmd)} failed with code {msg_ret_code}: <{message['payload']['message']}> {message['payload']['data']}")
def delete(self, iccid):
self.at(f'AT+QESIM="delete","{iccid}"')
messages.append(message)
def list_profiles(self):
out = self.at('AT+QESIM="list"')
return out.strip().splitlines()[1:]
if len(messages) == 0:
raise LPAError(f"lpac {cmd} returned no messages")
return messages
if __name__ == "__main__":
import sys
def _process_notifications(self) -> None:
"""
Process notifications stored on the eUICC, typically to activate/deactivate the profile with the carrier.
"""
self._validate_successful(self._invoke('notification', 'process', '-a', '-r'))
if "RESTART" in os.environ:
subprocess.check_call("sudo systemctl stop ModemManager", shell=True)
subprocess.check_call("/usr/comma/lte/lte.sh stop_blocking", shell=True)
subprocess.check_call("/usr/comma/lte/lte.sh start", shell=True)
while not os.path.exists('/dev/ttyUSB2'):
time.sleep(1)
time.sleep(3)
def _validate_profile_exists(self, iccid: str) -> None:
if not any(p.iccid == iccid for p in self.list_profiles()):
raise LPAProfileNotFoundError(f'profile {iccid} does not exist')
lpa = LPA()
print(lpa.list_profiles())
if len(sys.argv) > 1:
lpa.download(sys.argv[1])
print(lpa.list_profiles())
def _validate_successful(self, msgs: list[dict]) -> None:
assert msgs[-1]['payload']['message'] == 'success', 'expected success notification'
+7 -4
View File
@@ -10,8 +10,9 @@ from pathlib import Path
from cereal import log
from openpilot.common.gpio import gpio_set, gpio_init, get_irqs_for_action
from openpilot.system.hardware.base import HardwareBase, ThermalConfig, ThermalZone
from openpilot.system.hardware.base import HardwareBase, LPABase, ThermalConfig, ThermalZone
from openpilot.system.hardware.tici import iwlist
from openpilot.system.hardware.tici.esim import TiciLPA
from openpilot.system.hardware.tici.pins import GPIO
from openpilot.system.hardware.tici.amplifier import Amplifier
@@ -198,6 +199,9 @@ class Tici(HardwareBase):
'data_connected': modem.Get(MM_MODEM, 'State', dbus_interface=DBUS_PROPS, timeout=TIMEOUT) == MM_MODEM_STATE.CONNECTED,
}
def get_sim_lpa(self) -> LPABase:
return TiciLPA()
def get_imei(self, slot):
if slot != 0:
return ""
@@ -297,13 +301,11 @@ class Tici(HardwareBase):
return None
def get_modem_temperatures(self):
if self.get_device_type() == "mici":
return []
timeout = 0.2 # Default timeout is too short
try:
modem = self.get_modem()
temps = modem.Command("AT+QTEMP", math.ceil(timeout), dbus_interface=MM_MODEM, timeout=timeout)
return list(map(int, temps.split(' ')[1].split(',')))
return list(filter(lambda t: t != 255, map(int, temps.split(' ')[1].split(','))))
except Exception:
return []
@@ -335,6 +337,7 @@ class Tici(HardwareBase):
return ThermalConfig(cpu=[ThermalZone(f"cpu{i}-silver-usr") for i in range(4)] +
[ThermalZone(f"cpu{i}-gold-usr") for i in range(4)],
gpu=[ThermalZone("gpu0-usr"), ThermalZone("gpu1-usr")],
dsp=ThermalZone("compute-hvx-usr"),
memory=ThermalZone("ddr-usr"),
pmic=[ThermalZone("pm8998_tz"), ThermalZone("pm8005_tz")],
intake=intake,
+51
View File
@@ -0,0 +1,51 @@
import pytest
from openpilot.system.hardware import HARDWARE, TICI
from openpilot.system.hardware.base import LPAProfileNotFoundError
# https://euicc-manual.osmocom.org/docs/rsp/known-test-profile
# iccid is always the same for the given activation code
TEST_ACTIVATION_CODE = 'LPA:1$rsp.truphone.com$QRF-BETTERROAMING-PMRDGIR2EARDEIT5'
TEST_ICCID = '8944476500001944011'
TEST_NICKNAME = 'test_profile'
def cleanup():
lpa = HARDWARE.get_sim_lpa()
try:
lpa.delete_profile(TEST_ICCID)
except LPAProfileNotFoundError:
pass
lpa.process_notifications()
class TestEsim:
@classmethod
def setup_class(cls):
if not TICI:
pytest.skip()
cleanup()
@classmethod
def teardown_class(cls):
cleanup()
def test_provision_enable_disable(self):
lpa = HARDWARE.get_sim_lpa()
current_active = lpa.get_active_profile()
lpa.download_profile(TEST_ACTIVATION_CODE, TEST_NICKNAME)
assert any(p.iccid == TEST_ICCID and p.nickname == TEST_NICKNAME for p in lpa.list_profiles())
lpa.enable_profile(TEST_ICCID)
new_active = lpa.get_active_profile()
assert new_active is not None
assert new_active.iccid == TEST_ICCID
assert new_active.nickname == TEST_NICKNAME
lpa.disable_profile(TEST_ICCID)
new_active = lpa.get_active_profile()
assert new_active is None
if current_active:
lpa.enable_profile(current_active.iccid)
@@ -33,7 +33,7 @@ class Proc:
PROCS = [
Proc(['camerad'], 1.75, msgs=['roadCameraState', 'wideRoadCameraState', 'driverCameraState']),
Proc(['modeld'], 1.12, atol=0.2, msgs=['modelV2']),
Proc(['dmonitoringmodeld'], 1.4, msgs=['driverStateV2']),
Proc(['dmonitoringmodeld'], 0.6, msgs=['driverStateV2']),
Proc(['encoderd'], 0.23, msgs=[]),
]
+1
View File
@@ -5,5 +5,6 @@ The user interfaces here are built with [raylib](https://www.raylib.com/).
Quick start:
* set `DEBUG_FPS=1` to show the FPS
* set `STRICT_MODE=1` to kill the app if it drops too much below 60fps
* set `SCALE=1.5` to scale the entire UI by 1.5x
* https://www.raylib.com/cheatsheet/cheatsheet.html
* https://electronstudio.github.io/raylib-python-cffi/README.html#quickstart
+141 -39
View File
@@ -3,7 +3,7 @@ import os
import time
import pyray as rl
from enum import IntEnum
from openpilot.common.basedir import BASEDIR
from importlib.resources import as_file, files
from openpilot.common.swaglog import cloudlog
from openpilot.system.hardware import HARDWARE
@@ -12,23 +12,28 @@ FPS_LOG_INTERVAL = 5 # Seconds between logging FPS drops
FPS_DROP_THRESHOLD = 0.9 # FPS drop threshold for triggering a warning
FPS_CRITICAL_THRESHOLD = 0.5 # Critical threshold for triggering strict actions
ENABLE_VSYNC = os.getenv("ENABLE_VSYNC") == "1"
DEBUG_FPS = os.getenv("DEBUG_FPS") == '1'
STRICT_MODE = os.getenv("STRICT_MODE") == '1'
SCALE = float(os.getenv("SCALE", "1.0"))
DEFAULT_TEXT_SIZE = 60
DEFAULT_TEXT_COLOR = rl.Color(200, 200, 200, 255)
FONT_DIR = os.path.join(BASEDIR, "selfdrive/assets/fonts")
DEFAULT_TEXT_COLOR = rl.WHITE
ASSETS_DIR = files("openpilot.selfdrive").joinpath("assets")
FONT_DIR = ASSETS_DIR.joinpath("fonts")
class FontWeight(IntEnum):
BLACK = 0
BOLD = 1
EXTRA_BOLD = 2
EXTRA_LIGHT = 3
THIN = 0
EXTRA_LIGHT = 1
LIGHT = 2
NORMAL = 3
MEDIUM = 4
NORMAL = 5
SEMI_BOLD = 6
THIN = 7
SEMI_BOLD = 5
BOLD = 6
EXTRA_BOLD = 7
BLACK = 8
class GuiApplication:
@@ -36,10 +41,15 @@ class GuiApplication:
self._fonts: dict[FontWeight, rl.Font] = {}
self._width = width
self._height = height
self._textures: list[rl.Texture] = []
self._scale = SCALE
self._scaled_width = int(self._width * self._scale)
self._scaled_height = int(self._height * self._scale)
self._render_texture: rl.RenderTexture | None = None
self._textures: dict[str, rl.Texture] = {}
self._target_fps: int = DEFAULT_FPS
self._last_fps_log_time: float = time.monotonic()
self._window_close_requested = False
self._trace_log_callback = None
def request_close(self):
self._window_close_requested = True
@@ -50,57 +60,113 @@ class GuiApplication:
HARDWARE.set_display_power(True)
HARDWARE.set_screen_brightness(65)
rl.set_config_flags(rl.ConfigFlags.FLAG_MSAA_4X_HINT | rl.ConfigFlags.FLAG_VSYNC_HINT)
rl.init_window(self._width, self._height, title)
self._set_log_callback()
rl.set_trace_log_level(rl.TraceLogLevel.LOG_ALL)
flags = rl.ConfigFlags.FLAG_MSAA_4X_HINT
if ENABLE_VSYNC:
flags |= rl.ConfigFlags.FLAG_VSYNC_HINT
rl.set_config_flags(flags)
rl.init_window(self._scaled_width, self._scaled_height, title)
if self._scale != 1.0:
rl.set_mouse_scale(1 / self._scale, 1 / self._scale)
self._render_texture = rl.load_render_texture(self._width, self._height)
rl.set_texture_filter(self._render_texture.texture, rl.TextureFilter.TEXTURE_FILTER_BILINEAR)
rl.set_target_fps(fps)
self._target_fps = fps
self._set_styles()
self._load_fonts()
def load_texture_from_image(self, file_name: str, width: int, height: int, alpha_premultiply = False):
def texture(self, asset_path: str, width: int, height: int, alpha_premultiply=False, keep_aspect_ratio=True):
cache_key = f"{asset_path}_{width}_{height}_{alpha_premultiply}{keep_aspect_ratio}"
if cache_key in self._textures:
return self._textures[cache_key]
with as_file(ASSETS_DIR.joinpath(asset_path)) as fspath:
texture_obj = self._load_texture_from_image(fspath.as_posix(), width, height, alpha_premultiply, keep_aspect_ratio)
self._textures[cache_key] = texture_obj
return texture_obj
def _load_texture_from_image(self, image_path: str, width: int, height: int, alpha_premultiply=False, keep_aspect_ratio=True):
"""Load and resize a texture, storing it for later automatic unloading."""
image = rl.load_image(file_name)
image = rl.load_image(image_path)
if alpha_premultiply:
rl.image_alpha_premultiply(image)
rl.image_resize(image, width, height)
# Resize with aspect ratio preservation if requested
if keep_aspect_ratio:
orig_width = image.width
orig_height = image.height
scale_width = width / orig_width
scale_height = height / orig_height
# Calculate new dimensions
scale = min(scale_width, scale_height)
new_width = int(orig_width * scale)
new_height = int(orig_height * scale)
rl.image_resize(image, new_width, new_height)
else:
rl.image_resize(image, width, height)
texture = rl.load_texture_from_image(image)
# Set texture filtering to smooth the result
rl.set_texture_filter(texture, rl.TextureFilter.TEXTURE_FILTER_BILINEAR)
rl.unload_image(image)
self._textures.append(texture)
return texture
def close(self):
if not rl.is_window_ready():
return
for texture in self._textures:
for texture in self._textures.values():
rl.unload_texture(texture)
self._textures = []
self._textures = {}
for font in self._fonts.values():
rl.unload_font(font)
self._fonts = {}
if self._render_texture is not None:
rl.unload_render_texture(self._render_texture)
self._render_texture = None
rl.close_window()
def render(self):
while not (self._window_close_requested or rl.window_should_close()):
rl.begin_drawing()
rl.clear_background(rl.BLACK)
try:
while not (self._window_close_requested or rl.window_should_close()):
if self._render_texture:
rl.begin_texture_mode(self._render_texture)
rl.clear_background(rl.BLACK)
else:
rl.begin_drawing()
rl.clear_background(rl.BLACK)
yield
yield
if DEBUG_FPS:
rl.draw_fps(10, 10)
if self._render_texture:
rl.end_texture_mode()
rl.begin_drawing()
rl.clear_background(rl.BLACK)
src_rect = rl.Rectangle(0, 0, float(self._width), -float(self._height))
dst_rect = rl.Rectangle(0, 0, float(self._scaled_width), float(self._scaled_height))
rl.draw_texture_pro(self._render_texture.texture, src_rect, dst_rect, rl.Vector2(0, 0), 0.0, rl.WHITE)
rl.end_drawing()
self._monitor_fps()
if DEBUG_FPS:
rl.draw_fps(10, 10)
def font(self, font_weight: FontWeight=FontWeight.NORMAL):
rl.end_drawing()
self._monitor_fps()
except KeyboardInterrupt:
pass
def font(self, font_weight: FontWeight = FontWeight.NORMAL):
return self._fonts[font_weight]
@property
@@ -113,21 +179,34 @@ class GuiApplication:
def _load_fonts(self):
font_files = (
"Inter-Black.ttf",
"Inter-Thin.ttf",
"Inter-ExtraLight.ttf",
"Inter-Light.ttf",
"Inter-Regular.ttf",
"Inter-Medium.ttf",
"Inter-SemiBold.ttf",
"Inter-Bold.ttf",
"Inter-ExtraBold.ttf",
"Inter-ExtraLight.ttf",
"Inter-Medium.ttf",
"Inter-Regular.ttf",
"Inter-SemiBold.ttf",
"Inter-Thin.ttf"
)
"Inter-Black.ttf",
)
# Create a character set from our keyboard layouts
from openpilot.system.ui.widgets.keyboard import KEYBOARD_LAYOUTS
all_chars = set()
for layout in KEYBOARD_LAYOUTS.values():
all_chars.update(key for row in layout for key in row)
all_chars = "".join(all_chars)
codepoint_count = rl.ffi.new("int *", 1)
codepoints = rl.load_codepoints(all_chars, codepoint_count)
for index, font_file in enumerate(font_files):
font = rl.load_font_ex(os.path.join(FONT_DIR, font_file), 120, None, 0)
rl.set_texture_filter(font.texture, rl.TextureFilter.TEXTURE_FILTER_BILINEAR)
self._fonts[index] = font
with as_file(FONT_DIR.joinpath(font_file)) as fspath:
font = rl.load_font_ex(fspath.as_posix(), 120, codepoints, codepoint_count[0])
rl.set_texture_filter(font.texture, rl.TextureFilter.TEXTURE_FILTER_BILINEAR)
self._fonts[index] = font
rl.unload_codepoints(codepoints)
rl.gui_set_font(self._fonts[FontWeight.NORMAL])
def _set_styles(self):
@@ -137,6 +216,29 @@ class GuiApplication:
rl.gui_set_style(rl.GuiControl.DEFAULT, rl.GuiControlProperty.TEXT_COLOR_NORMAL, rl.color_to_int(DEFAULT_TEXT_COLOR))
rl.gui_set_style(rl.GuiControl.DEFAULT, rl.GuiControlProperty.BASE_COLOR_NORMAL, rl.color_to_int(rl.Color(50, 50, 50, 255)))
def _set_log_callback(self):
@rl.ffi.callback("void(int, char *, void *)")
def trace_log_callback(log_level, text, args):
try:
text_str = rl.ffi.string(text).decode('utf-8')
except (TypeError, UnicodeDecodeError):
text_str = str(text)
if log_level == rl.TraceLogLevel.LOG_ERROR:
cloudlog.error(f"raylib: {text_str}")
elif log_level == rl.TraceLogLevel.LOG_WARNING:
cloudlog.warning(f"raylib: {text_str}")
elif log_level == rl.TraceLogLevel.LOG_INFO:
cloudlog.info(f"raylib: {text_str}")
elif log_level == rl.TraceLogLevel.LOG_DEBUG:
cloudlog.debug(f"raylib: {text_str}")
else:
cloudlog.error(f"raylib: Unknown level {log_level}: {text_str}")
# Store callback reference
self._trace_log_callback = trace_log_callback
rl.set_trace_log_callback(self._trace_log_callback)
def _monitor_fps(self):
fps = rl.get_fps()
+77 -10
View File
@@ -8,11 +8,21 @@ class ButtonStyle(IntEnum):
PRIMARY = 1 # For main actions
DANGER = 2 # For critical actions, like reboot or delete
TRANSPARENT = 3 # For buttons with transparent background and border
ACTION = 4
class TextAlignment(IntEnum):
LEFT = 0
CENTER = 1
RIGHT = 2
ICON_PADDING = 15
DEFAULT_BUTTON_FONT_SIZE = 60
BUTTON_ENABLED_TEXT_COLOR = rl.Color(228, 228, 228, 255)
BUTTON_DISABLED_TEXT_COLOR = rl.Color(228, 228, 228, 51)
ACTION_BUTTON_FONT_SIZE = 48
ACTION_BUTTON_TEXT_COLOR = rl.Color(0, 0, 0, 255)
BUTTON_BACKGROUND_COLORS = {
@@ -20,6 +30,7 @@ BUTTON_BACKGROUND_COLORS = {
ButtonStyle.PRIMARY: rl.Color(70, 91, 234, 255),
ButtonStyle.DANGER: rl.Color(255, 36, 36, 255),
ButtonStyle.TRANSPARENT: rl.BLACK,
ButtonStyle.ACTION: rl.Color(189, 189, 189, 255),
}
BUTTON_PRESSED_BACKGROUND_COLORS = {
@@ -27,8 +38,11 @@ BUTTON_PRESSED_BACKGROUND_COLORS = {
ButtonStyle.PRIMARY: rl.Color(48, 73, 244, 255),
ButtonStyle.DANGER: rl.Color(255, 36, 36, 255),
ButtonStyle.TRANSPARENT: rl.BLACK,
ButtonStyle.ACTION: rl.Color(130, 130, 130, 255),
}
_pressed_buttons: set[str] = set() # Track mouse press state globally
def gui_button(
rect: rl.Rectangle,
@@ -38,16 +52,42 @@ def gui_button(
button_style: ButtonStyle = ButtonStyle.NORMAL,
is_enabled: bool = True,
border_radius: int = 10, # Corner rounding in pixels
text_alignment: TextAlignment = TextAlignment.CENTER,
text_padding: int = 20, # Padding for left/right alignment
icon=None,
) -> int:
button_id = f"{rect.x}_{rect.y}_{rect.width}_{rect.height}"
result = 0
if button_style in (ButtonStyle.PRIMARY, ButtonStyle.DANGER) and not is_enabled:
button_style = ButtonStyle.NORMAL
if button_style == ButtonStyle.ACTION and font_size == DEFAULT_BUTTON_FONT_SIZE:
font_size = ACTION_BUTTON_FONT_SIZE
# Set background color based on button type
bg_color = BUTTON_BACKGROUND_COLORS[button_style]
if is_enabled and rl.check_collision_point_rec(rl.get_mouse_position(), rect):
if rl.is_mouse_button_down(rl.MouseButton.MOUSE_BUTTON_LEFT):
mouse_over = is_enabled and rl.check_collision_point_rec(rl.get_mouse_position(), rect)
is_pressed = button_id in _pressed_buttons
if mouse_over:
if rl.is_mouse_button_pressed(rl.MouseButton.MOUSE_BUTTON_LEFT):
# Only this button enters pressed state
_pressed_buttons.add(button_id)
is_pressed = True
# Use pressed color when mouse is down over this button
if is_pressed and rl.is_mouse_button_down(rl.MouseButton.MOUSE_BUTTON_LEFT):
bg_color = BUTTON_PRESSED_BACKGROUND_COLORS[button_style]
elif rl.is_mouse_button_released(rl.MouseButton.MOUSE_BUTTON_LEFT):
# Handle button click
if rl.is_mouse_button_released(rl.MouseButton.MOUSE_BUTTON_LEFT) and is_pressed:
result = 1
_pressed_buttons.remove(button_id)
# Clean up pressed state if mouse is released anywhere
if rl.is_mouse_button_released(rl.MouseButton.MOUSE_BUTTON_LEFT) and button_id in _pressed_buttons:
_pressed_buttons.remove(button_id)
# Draw the button with rounded corners
roundness = border_radius / (min(rect.width, rect.height) / 2)
@@ -57,15 +97,42 @@ def gui_button(
rl.draw_rectangle_rounded(rect, roundness, 20, rl.BLACK)
rl.draw_rectangle_rounded_lines_ex(rect, roundness, 20, 2, rl.WHITE)
# Handle icon and text positioning
font = gui_app.font(font_weight)
# Center text in the button
text_size = rl.measure_text_ex(font, text, font_size, 0)
text_pos = rl.Vector2(
rect.x + (rect.width - text_size.x) // 2, rect.y + (rect.height - text_size.y) // 2
)
text_pos = rl.Vector2(0, rect.y + (rect.height - text_size.y) // 2) # Vertical centering
# Draw the button text
text_color = BUTTON_ENABLED_TEXT_COLOR if is_enabled else BUTTON_DISABLED_TEXT_COLOR
rl.draw_text_ex(font, text, text_pos, font_size, 0, text_color)
# Draw icon if provided
if icon:
icon_y = rect.y + (rect.height - icon.height) / 2
if text:
if text_alignment == TextAlignment.LEFT:
icon_x = rect.x + text_padding
text_pos.x = icon_x + icon.width + ICON_PADDING
elif text_alignment == TextAlignment.CENTER:
total_width = icon.width + ICON_PADDING + text_size.x
icon_x = rect.x + (rect.width - total_width) / 2
text_pos.x = icon_x + icon.width + ICON_PADDING
else: # RIGHT
text_pos.x = rect.x + rect.width - text_size.x - text_padding
icon_x = text_pos.x - ICON_PADDING - icon.width
else:
# Center icon when no text
icon_x = rect.x + (rect.width - icon.width) / 2
rl.draw_texture_v(icon, rl.Vector2(icon_x, icon_y), rl.WHITE if is_enabled else rl.Color(255, 255, 255, 100))
else:
# No icon, position text normally
if text_alignment == TextAlignment.LEFT:
text_pos.x = rect.x + text_padding
elif text_alignment == TextAlignment.CENTER:
text_pos.x = rect.x + (rect.width - text_size.x) // 2
elif text_alignment == TextAlignment.RIGHT:
text_pos.x = rect.x + rect.width - text_size.x - text_padding
# Draw the button text if any
if text:
text_color = ACTION_BUTTON_TEXT_COLOR if button_style == ButtonStyle.ACTION else BUTTON_ENABLED_TEXT_COLOR if is_enabled else BUTTON_DISABLED_TEXT_COLOR
rl.draw_text_ex(font, text, text_pos, font_size, 0, text_color)
return result
+177
View File
@@ -0,0 +1,177 @@
import os
import cffi
from dataclasses import dataclass
from typing import Any
from openpilot.common.swaglog import cloudlog
# EGL constants
EGL_LINUX_DMA_BUF_EXT = 0x3270
EGL_WIDTH = 0x3057
EGL_HEIGHT = 0x3056
EGL_LINUX_DRM_FOURCC_EXT = 0x3271
EGL_DMA_BUF_PLANE0_FD_EXT = 0x3272
EGL_DMA_BUF_PLANE0_OFFSET_EXT = 0x3273
EGL_DMA_BUF_PLANE0_PITCH_EXT = 0x3274
EGL_DMA_BUF_PLANE1_FD_EXT = 0x3275
EGL_DMA_BUF_PLANE1_OFFSET_EXT = 0x3276
EGL_DMA_BUF_PLANE1_PITCH_EXT = 0x3277
EGL_NONE = 0x3038
GL_TEXTURE0 = 0x84C0
GL_TEXTURE_EXTERNAL_OES = 0x8D65
# DRM Format for NV12
DRM_FORMAT_NV12 = 842094158
@dataclass
class EGLImage:
"""Container for EGL image and associated resources"""
egl_image: Any
fd: int
@dataclass
class EGLState:
"""Container for all EGL-related state"""
initialized: bool = False
ffi: Any = None
egl_lib: Any = None
gles_lib: Any = None
# EGL display connection - shared across all users
display: Any = None
# Constants
NO_CONTEXT: Any = None
NO_DISPLAY: Any = None
NO_IMAGE_KHR: Any = None
# Function pointers
get_current_display: Any = None
create_image_khr: Any = None
destroy_image_khr: Any = None
image_target_texture: Any = None
get_error: Any = None
bind_texture: Any = None
active_texture: Any = None
# Create a single instance of the state
_egl = EGLState()
def init_egl() -> bool:
"""Initialize EGL and load necessary functions"""
global _egl
# Don't re-initialize if already done
if _egl.initialized:
return True
try:
_egl.ffi = cffi.FFI()
_egl.ffi.cdef("""
typedef int EGLint;
typedef unsigned int EGLBoolean;
typedef unsigned int EGLenum;
typedef unsigned int GLenum;
typedef void *EGLContext;
typedef void *EGLDisplay;
typedef void *EGLClientBuffer;
typedef void *EGLImageKHR;
typedef void *GLeglImageOES;
EGLDisplay eglGetCurrentDisplay(void);
EGLint eglGetError(void);
EGLImageKHR eglCreateImageKHR(EGLDisplay dpy, EGLContext ctx,
EGLenum target, EGLClientBuffer buffer,
const EGLint *attrib_list);
EGLBoolean eglDestroyImageKHR(EGLDisplay dpy, EGLImageKHR image);
void glEGLImageTargetTexture2DOES(GLenum target, GLeglImageOES image);
void glBindTexture(GLenum target, unsigned int texture);
void glActiveTexture(GLenum texture);
""")
# Load libraries
_egl.egl_lib = _egl.ffi.dlopen("libEGL.so")
_egl.gles_lib = _egl.ffi.dlopen("libGLESv2.so")
# Cast NULL pointers
_egl.NO_CONTEXT = _egl.ffi.cast("void *", 0)
_egl.NO_DISPLAY = _egl.ffi.cast("void *", 0)
_egl.NO_IMAGE_KHR = _egl.ffi.cast("void *", 0)
# Bind functions
_egl.get_current_display = _egl.egl_lib.eglGetCurrentDisplay
_egl.create_image_khr = _egl.egl_lib.eglCreateImageKHR
_egl.destroy_image_khr = _egl.egl_lib.eglDestroyImageKHR
_egl.image_target_texture = _egl.gles_lib.glEGLImageTargetTexture2DOES
_egl.get_error = _egl.egl_lib.eglGetError
_egl.bind_texture = _egl.gles_lib.glBindTexture
_egl.active_texture = _egl.gles_lib.glActiveTexture
# Initialize EGL display once here
_egl.display = _egl.get_current_display()
if _egl.display == _egl.NO_DISPLAY:
raise RuntimeError("Failed to get EGL display")
_egl.initialized = True
return True
except Exception as e:
cloudlog.exception(f"EGL initialization failed: {e}")
_egl.initialized = False
return False
def create_egl_image(width: int, height: int, stride: int, fd: int, uv_offset: int) -> EGLImage | None:
assert _egl.initialized, "EGL not initialized"
# Duplicate fd since EGL needs it
dup_fd = os.dup(fd)
# Create image attributes for EGL
img_attrs = [
EGL_WIDTH, width,
EGL_HEIGHT, height,
EGL_LINUX_DRM_FOURCC_EXT, DRM_FORMAT_NV12,
EGL_DMA_BUF_PLANE0_FD_EXT, dup_fd,
EGL_DMA_BUF_PLANE0_OFFSET_EXT, 0,
EGL_DMA_BUF_PLANE0_PITCH_EXT, stride,
EGL_DMA_BUF_PLANE1_FD_EXT, dup_fd,
EGL_DMA_BUF_PLANE1_OFFSET_EXT, uv_offset,
EGL_DMA_BUF_PLANE1_PITCH_EXT, stride,
EGL_NONE
]
attr_array = _egl.ffi.new("int[]", img_attrs)
egl_image = _egl.create_image_khr(_egl.display, _egl.NO_CONTEXT, EGL_LINUX_DMA_BUF_EXT, _egl.ffi.NULL, attr_array)
if egl_image == _egl.NO_IMAGE_KHR:
cloudlog.error(f"Failed to create EGL image: {_egl.get_error()}")
os.close(dup_fd)
return None
return EGLImage(egl_image=egl_image, fd=dup_fd)
def destroy_egl_image(egl_image: EGLImage) -> None:
assert _egl.initialized, "EGL not initialized"
_egl.destroy_image_khr(_egl.display, egl_image.egl_image)
# Close the duplicated fd we created in create_egl_image()
# We need to handle OSError since the fd might already be closed
try:
os.close(egl_image.fd)
except OSError:
pass
def bind_egl_image_to_texture(texture_id: int, egl_image: EGLImage) -> None:
assert _egl.initialized, "EGL not initialized"
_egl.active_texture(GL_TEXTURE0)
_egl.bind_texture(GL_TEXTURE_EXTERNAL_OES, texture_id)
_egl.image_target_texture(GL_TEXTURE_EXTERNAL_OES, egl_image.egl_image)
+232
View File
@@ -0,0 +1,232 @@
import pyray as rl
import time
from openpilot.system.ui.lib.application import gui_app
PASSWORD_MASK_CHAR = ""
PASSWORD_MASK_DELAY = 1.5 # Seconds to show character before masking
class InputBox:
def __init__(self, max_text_size=255, password_mode=False):
self._max_text_size = max_text_size
self._input_text = ""
self._cursor_position = 0
self._password_mode = password_mode
self._blink_counter = 0
self._show_cursor = False
self._last_key_pressed = 0
self._key_press_time = 0
self._repeat_delay = 30
self._repeat_rate = 4
self._text_offset = 0
self._visible_width = 0
self._last_char_time = 0 # Track when last character was added
self._masked_length = 0 # How many characters are currently masked
@property
def text(self):
return self._input_text
@text.setter
def text(self, value):
self._input_text = value[: self._max_text_size]
self._cursor_position = len(self._input_text)
self._update_text_offset()
def set_password_mode(self, password_mode):
self._password_mode = password_mode
def clear(self):
self._input_text = ''
self._cursor_position = 0
self._text_offset = 0
def set_cursor_position(self, position):
"""Set the cursor position and reset the blink counter."""
if 0 <= position <= len(self._input_text):
self._cursor_position = position
self._blink_counter = 0
self._show_cursor = True
self._update_text_offset()
def _update_text_offset(self):
"""Ensure the cursor is visible by adjusting text offset."""
if self._visible_width == 0:
return
font = gui_app.font()
display_text = self._get_display_text()
padding = 10
if self._cursor_position > 0:
cursor_x = rl.measure_text_ex(font, display_text[: self._cursor_position], self._font_size, 0).x
else:
cursor_x = 0
visible_width = self._visible_width - (padding * 2)
# Adjust offset if cursor would be outside visible area
if cursor_x < self._text_offset:
self._text_offset = max(0, cursor_x - padding)
elif cursor_x > self._text_offset + visible_width:
self._text_offset = cursor_x - visible_width + padding
def add_char_at_cursor(self, char):
"""Add a character at the current cursor position."""
if len(self._input_text) < self._max_text_size:
self._input_text = self._input_text[: self._cursor_position] + char + self._input_text[self._cursor_position :]
self.set_cursor_position(self._cursor_position + 1)
if self._password_mode:
self._last_char_time = time.time()
return True
return False
def delete_char_before_cursor(self):
"""Delete the character before the cursor position (backspace)."""
if self._cursor_position > 0:
self._input_text = self._input_text[: self._cursor_position - 1] + self._input_text[self._cursor_position :]
self.set_cursor_position(self._cursor_position - 1)
return True
return False
def delete_char_at_cursor(self):
"""Delete the character at the cursor position (delete)."""
if self._cursor_position < len(self._input_text):
self._input_text = self._input_text[: self._cursor_position] + self._input_text[self._cursor_position + 1 :]
self.set_cursor_position(self._cursor_position)
return True
return False
def render(self, rect, color=rl.BLACK, border_color=rl.DARKGRAY, text_color=rl.WHITE, font_size=80):
# Store dimensions for text offset calculations
self._visible_width = rect.width
self._font_size = font_size
# Handle mouse input
self._handle_mouse_input(rect, font_size)
# Draw input box
rl.draw_rectangle_rec(rect, color)
# Process keyboard input
self._handle_keyboard_input()
# Update cursor blink
self._blink_counter += 1
if self._blink_counter >= 30:
self._show_cursor = not self._show_cursor
self._blink_counter = 0
# Display text
font = gui_app.font()
display_text = self._get_display_text()
padding = 10
# Clip text within input box bounds
buffer = 2
rl.begin_scissor_mode(int(rect.x + padding - buffer), int(rect.y), int(rect.width - padding * 2 + buffer * 2), int(rect.height))
rl.draw_text_ex(
font,
display_text,
rl.Vector2(int(rect.x + padding - self._text_offset), int(rect.y + rect.height / 2 - font_size / 2)),
font_size,
0,
text_color,
)
# Draw cursor
if self._show_cursor:
cursor_x = rect.x + padding
if len(display_text) > 0 and self._cursor_position > 0:
cursor_x += rl.measure_text_ex(font, display_text[: self._cursor_position], font_size, 0).x
# Apply text offset to cursor position
cursor_x -= self._text_offset
cursor_height = font_size + 4
cursor_y = rect.y + rect.height / 2 - cursor_height / 2
rl.draw_line(int(cursor_x), int(cursor_y), int(cursor_x), int(cursor_y + cursor_height), rl.WHITE)
rl.end_scissor_mode()
def _get_display_text(self):
"""Get text to display, applying password masking with delay if needed."""
if not self._password_mode:
return self._input_text
# Show character at last edited position if within delay window
masked_text = PASSWORD_MASK_CHAR * len(self._input_text)
recent_edit = time.time() - self._last_char_time < PASSWORD_MASK_DELAY
if recent_edit and self._input_text:
last_pos = max(0, self._cursor_position - 1)
if last_pos < len(self._input_text):
return masked_text[:last_pos] + self._input_text[last_pos] + masked_text[last_pos + 1 :]
return masked_text
def _handle_mouse_input(self, rect, font_size):
"""Handle mouse clicks to position cursor."""
mouse_pos = rl.get_mouse_position()
if rl.is_mouse_button_pressed(rl.MOUSE_LEFT_BUTTON) and rl.check_collision_point_rec(mouse_pos, rect):
# Calculate cursor position from click
if len(self._input_text) > 0:
font = gui_app.font()
display_text = self._get_display_text()
# Find the closest character position to the click
relative_x = mouse_pos.x - (rect.x + 10) + self._text_offset
best_pos = 0
min_distance = float('inf')
for i in range(len(self._input_text) + 1):
char_width = rl.measure_text_ex(font, display_text[:i], font_size, 0).x
distance = abs(relative_x - char_width)
if distance < min_distance:
min_distance = distance
best_pos = i
self.set_cursor_position(best_pos)
else:
self.set_cursor_position(0)
def _handle_keyboard_input(self):
# Handle navigation keys
key = rl.get_key_pressed()
if key != 0:
self._process_key(key)
if key in (rl.KEY_LEFT, rl.KEY_RIGHT, rl.KEY_BACKSPACE, rl.KEY_DELETE):
self._last_key_pressed = key
self._key_press_time = 0
# Handle repeats for held keys
elif self._last_key_pressed != 0:
if rl.is_key_down(self._last_key_pressed):
self._key_press_time += 1
if self._key_press_time > self._repeat_delay and self._key_press_time % self._repeat_rate == 0:
self._process_key(self._last_key_pressed)
else:
self._last_key_pressed = 0
# Handle text input
char = rl.get_char_pressed()
if char != 0 and char >= 32: # Filter out control characters
self.add_char_at_cursor(chr(char))
def _process_key(self, key):
if key == rl.KEY_LEFT:
if self._cursor_position > 0:
self.set_cursor_position(self._cursor_position - 1)
elif key == rl.KEY_RIGHT:
if self._cursor_position < len(self._input_text):
self.set_cursor_position(self._cursor_position + 1)
elif key == rl.KEY_BACKSPACE:
self.delete_char_before_cursor()
elif key == rl.KEY_DELETE:
self.delete_char_at_cursor()
elif key == rl.KEY_HOME:
self.set_cursor_position(0)
elif key == rl.KEY_END:
self.set_cursor_position(len(self._input_text))
+19 -5
View File
@@ -10,13 +10,27 @@ def gui_label(
color: rl.Color = DEFAULT_TEXT_COLOR,
font_weight: FontWeight = FontWeight.NORMAL,
alignment: int = rl.GuiTextAlignment.TEXT_ALIGN_LEFT,
alignment_vertical: int = rl.GuiTextAlignmentVertical.TEXT_ALIGN_MIDDLE
alignment_vertical: int = rl.GuiTextAlignmentVertical.TEXT_ALIGN_MIDDLE,
elide_right: bool = True
):
# Set font based on the provided weight
font = gui_app.font(font_weight)
# Measure text size
text_size = rl.measure_text_ex(font, text, font_size, 0)
display_text = text
# Elide text to fit within the rectangle
if elide_right and text_size.x > rect.width:
ellipsis = "..."
left, right = 0, len(text)
while left < right:
mid = (left + right) // 2
candidate = text[:mid] + ellipsis
candidate_size = rl.measure_text_ex(font, candidate, font_size, 0)
if candidate_size.x <= rect.width:
left = mid + 1
else:
right = mid
display_text = text[: left - 1] + ellipsis if left > 0 else ellipsis
text_size = rl.measure_text_ex(font, display_text, font_size, 0)
# Calculate horizontal position based on alignment
text_x = rect.x + {
@@ -33,7 +47,7 @@ def gui_label(
}.get(alignment_vertical, 0)
# Draw the text in the specified rectangle
rl.draw_text_ex(font, text, rl.Vector2(text_x, text_y), font_size, 0, color)
rl.draw_text_ex(font, display_text, rl.Vector2(text_x, text_y), font_size, 0, color)
def gui_text_box(
+138 -30
View File
@@ -1,16 +1,23 @@
import pyray as rl
from enum import IntEnum
# Scroll constants for smooth scrolling behavior
MOUSE_WHEEL_SCROLL_SPEED = 30
INERTIA_FRICTION = 0.95 # The rate at which the inertia slows down
MIN_VELOCITY = 0.1 # Minimum velocity before stopping the inertia
DRAG_THRESHOLD = 5 # Pixels of movement to consider it a drag, not a click
INERTIA_FRICTION = 0.92 # The rate at which the inertia slows down
MIN_VELOCITY = 0.5 # Minimum velocity before stopping the inertia
DRAG_THRESHOLD = 5 # Pixels of movement to consider it a drag, not a click
BOUNCE_FACTOR = 0.2 # Elastic bounce when scrolling past boundaries
BOUNCE_RETURN_SPEED = 0.15 # How quickly it returns from the bounce
MAX_BOUNCE_DISTANCE = 150 # Maximum distance for bounce effect
FLICK_MULTIPLIER = 1.8 # Multiplier for flick gestures
VELOCITY_HISTORY_SIZE = 5 # Track velocity over multiple frames for smoother motion
class ScrollState(IntEnum):
IDLE = 0
DRAGGING_CONTENT = 1
DRAGGING_SCROLLBAR = 2
BOUNCING = 3
class GuiScrollPanel:
@@ -22,14 +29,33 @@ class GuiScrollPanel:
self._view = rl.Rectangle(0, 0, 0, 0)
self._show_vertical_scroll_bar: bool = show_vertical_scroll_bar
self._velocity_y = 0.0 # Velocity for inertia
self._is_dragging = False
self._is_dragging: bool = False
self._bounce_offset: float = 0.0
self._last_frame_time = rl.get_time()
self._velocity_history: list[float] = []
self._last_drag_time: float = 0.0
self._content_rect: rl.Rectangle | None = None
self._bounds_rect: rl.Rectangle | None = None
def handle_scroll(self, bounds: rl.Rectangle, content: rl.Rectangle) -> rl.Vector2:
mouse_pos = rl.get_mouse_position()
# Store rectangles for reference
self._content_rect = content
self._bounds_rect = bounds
# Handle dragging logic
# Calculate time delta
current_time = rl.get_time()
delta_time = current_time - self._last_frame_time
self._last_frame_time = current_time
# Prevent large jumps
delta_time = min(delta_time, 0.05)
mouse_pos = rl.get_mouse_position()
max_scroll_y = max(content.height - bounds.height, 0)
# Start dragging on mouse press
if rl.check_collision_point_rec(mouse_pos, bounds) and rl.is_mouse_button_pressed(rl.MouseButton.MOUSE_BUTTON_LEFT):
if self._scroll_state == ScrollState.IDLE:
if self._scroll_state == ScrollState.IDLE or self._scroll_state == ScrollState.BOUNCING:
self._scroll_state = ScrollState.DRAGGING_CONTENT
if self._show_vertical_scroll_bar:
scrollbar_width = rl.gui_get_style(rl.GuiControl.LISTVIEW, rl.GuiListViewProperty.SCROLLBAR_WIDTH)
@@ -38,51 +64,133 @@ class GuiScrollPanel:
self._scroll_state = ScrollState.DRAGGING_SCROLLBAR
self._last_mouse_y = mouse_pos.y
self._start_mouse_y = mouse_pos.y # Record starting position
self._velocity_y = 0.0 # Reset velocity when drag starts
self._is_dragging = False # Reset dragging flag
self._start_mouse_y = mouse_pos.y
self._last_drag_time = current_time
self._velocity_history = []
self._velocity_y = 0.0
self._bounce_offset = 0.0
self._is_dragging = False
if self._scroll_state != ScrollState.IDLE:
# Handle active dragging
if self._scroll_state == ScrollState.DRAGGING_CONTENT or self._scroll_state == ScrollState.DRAGGING_SCROLLBAR:
if rl.is_mouse_button_down(rl.MouseButton.MOUSE_BUTTON_LEFT):
delta_y = mouse_pos.y - self._last_mouse_y
# Check if movement exceeds the drag threshold
# Track velocity for inertia
time_since_last_drag = current_time - self._last_drag_time
if time_since_last_drag > 0:
drag_velocity = delta_y / time_since_last_drag / 60.0
self._velocity_history.append(drag_velocity)
if len(self._velocity_history) > VELOCITY_HISTORY_SIZE:
self._velocity_history.pop(0)
self._last_drag_time = current_time
# Detect actual dragging
total_drag = abs(mouse_pos.y - self._start_mouse_y)
if total_drag > DRAG_THRESHOLD:
self._is_dragging = True
if self._scroll_state == ScrollState.DRAGGING_CONTENT:
# Add resistance at boundaries
if (self._offset.y > 0 and delta_y > 0) or (self._offset.y < -max_scroll_y and delta_y < 0):
delta_y *= BOUNCE_FACTOR
self._offset.y += delta_y
elif self._scroll_state == ScrollState.DRAGGING_SCROLLBAR:
delta_y = -delta_y
scroll_ratio = content.height / bounds.height
self._offset.y -= delta_y * scroll_ratio
self._last_mouse_y = mouse_pos.y
self._velocity_y = delta_y # Update velocity during drag
elif rl.is_mouse_button_released(rl.MouseButton.MOUSE_BUTTON_LEFT):
self._scroll_state = ScrollState.IDLE
# Handle mouse wheel scrolling
elif rl.is_mouse_button_released(rl.MouseButton.MOUSE_BUTTON_LEFT):
# Calculate flick velocity
if self._velocity_history:
total_weight = 0
weighted_velocity = 0.0
for i, v in enumerate(self._velocity_history):
weight = i + 1
weighted_velocity += v * weight
total_weight += weight
if total_weight > 0:
avg_velocity = weighted_velocity / total_weight
self._velocity_y = avg_velocity * FLICK_MULTIPLIER
# Check bounds
if self._offset.y > 0 or self._offset.y < -max_scroll_y:
self._scroll_state = ScrollState.BOUNCING
else:
self._scroll_state = ScrollState.IDLE
# Handle mouse wheel
wheel_move = rl.get_mouse_wheel_move()
if self._show_vertical_scroll_bar:
self._offset.y += wheel_move * (MOUSE_WHEEL_SCROLL_SPEED - 20)
rl.gui_scroll_panel(bounds, rl.ffi.NULL, content, self._offset, self._view)
else:
self._offset.y += wheel_move * MOUSE_WHEEL_SCROLL_SPEED
if wheel_move != 0:
self._velocity_y = 0.0
if self._show_vertical_scroll_bar:
self._offset.y += wheel_move * (MOUSE_WHEEL_SCROLL_SPEED - 20)
rl.gui_scroll_panel(bounds, rl.ffi.NULL, content, self._offset, self._view)
else:
self._offset.y += wheel_move * MOUSE_WHEEL_SCROLL_SPEED
if self._offset.y > 0 or self._offset.y < -max_scroll_y:
self._scroll_state = ScrollState.BOUNCING
# Apply inertia (continue scrolling after mouse release)
if self._scroll_state == ScrollState.IDLE:
self._offset.y += self._velocity_y
self._velocity_y *= INERTIA_FRICTION # Slow down velocity over time
if abs(self._velocity_y) > MIN_VELOCITY:
self._offset.y += self._velocity_y
self._velocity_y *= INERTIA_FRICTION
# Stop scrolling when velocity is low
if abs(self._velocity_y) < MIN_VELOCITY:
if self._offset.y > 0 or self._offset.y < -max_scroll_y:
self._scroll_state = ScrollState.BOUNCING
else:
self._velocity_y = 0.0
# Ensure scrolling doesn't go beyond bounds
max_scroll_y = max(content.height - bounds.height, 0)
self._offset.y = max(min(self._offset.y, 0), -max_scroll_y)
# Handle bouncing effect
elif self._scroll_state == ScrollState.BOUNCING:
target_y = 0.0
if self._offset.y < -max_scroll_y:
target_y = -max_scroll_y
distance = target_y - self._offset.y
bounce_step = distance * BOUNCE_RETURN_SPEED
self._offset.y += bounce_step
self._velocity_y *= INERTIA_FRICTION * 0.8
if abs(distance) < 0.5 and abs(self._velocity_y) < MIN_VELOCITY:
self._offset.y = target_y
self._velocity_y = 0.0
self._scroll_state = ScrollState.IDLE
# Limit bounce distance
if self._scroll_state != ScrollState.DRAGGING_CONTENT:
if self._offset.y > MAX_BOUNCE_DISTANCE:
self._offset.y = MAX_BOUNCE_DISTANCE
elif self._offset.y < -(max_scroll_y + MAX_BOUNCE_DISTANCE):
self._offset.y = -(max_scroll_y + MAX_BOUNCE_DISTANCE)
return self._offset
def is_click_valid(self) -> bool:
return self._scroll_state == ScrollState.IDLE and not self._is_dragging and rl.is_mouse_button_released(rl.MouseButton.MOUSE_BUTTON_LEFT)
# Check if this is a click rather than a drag
return (
self._scroll_state == ScrollState.IDLE
and not self._is_dragging
and rl.is_mouse_button_released(rl.MouseButton.MOUSE_BUTTON_LEFT)
)
def get_normalized_scroll_position(self) -> float:
"""Returns the current scroll position as a value from 0.0 to 1.0"""
if not self._content_rect or not self._bounds_rect:
return 0.0
max_scroll_y = max(self._content_rect.height - self._bounds_rect.height, 0)
if max_scroll_y == 0:
return 0.0
normalized = -self._offset.y / max_scroll_y
return max(0.0, min(1.0, normalized))
+243 -13
View File
@@ -1,17 +1,27 @@
import asyncio
import concurrent.futures
import copy
import threading
import time
import uuid
from collections.abc import Callable
from dataclasses import dataclass
from enum import IntEnum
from typing import TypeVar
from dbus_next.aio import MessageBus
from dbus_next import BusType, Variant, Message
from dbus_next.errors import DBusError
from dbus_next.constants import MessageType
try:
from openpilot.common.params import Params
except ImportError:
# Params/Cythonized modules are not available in zipapp
Params = None
from openpilot.common.swaglog import cloudlog
T = TypeVar("T")
# NetworkManager constants
NM = "org.freedesktop.NetworkManager"
NM_PATH = '/org/freedesktop/NetworkManager'
@@ -25,6 +35,9 @@ NM_DEVICE_IFACE = "org.freedesktop.NetworkManager.Device"
NM_DEVICE_STATE_REASON_SUPPLICANT_DISCONNECT = 8
TETHERING_IP_ADDRESS = "192.168.43.1"
DEFAULT_TETHERING_PASSWORD = "12345678"
# NetworkManager device states
class NMDeviceState(IntEnum):
DISCONNECTED = 30
@@ -48,14 +61,16 @@ class NetworkInfo:
security_type: SecurityType
path: str
bssid: str
is_saved: bool = False
# saved_path: str
@dataclass
class WifiManagerCallbacks:
need_auth: Callable[[], None] | None = None
need_auth: Callable[[str], None] | None = None
activated: Callable[[], None] | None = None
forgotten: Callable[[], None] | None = None
networks_updated: Callable[[list[NetworkInfo]], None] | None = None
class WifiManager:
@@ -68,7 +83,14 @@ class WifiManager:
self.saved_connections: dict[str, str] = {}
self.active_ap_path: str = ""
self.scan_task: asyncio.Task | None = None
# Set tethering ssid as "weedle" + first 4 characters of a dongle id
self._tethering_ssid = "weedle"
if Params is not None:
dongle_id = Params().get("DongleId", encoding="utf-8")
if dongle_id:
self._tethering_ssid += "-" + dongle_id[:4]
self.running: bool = True
self._current_connection_ssid: str | None = None
async def connect(self) -> None:
"""Connect to the DBus system bus."""
@@ -79,6 +101,7 @@ class WifiManager:
await self._setup_signals(self.device_path)
self.active_ap_path = await self.get_active_access_point()
await self.add_tethering_connection(self._tethering_ssid, DEFAULT_TETHERING_PASSWORD)
self.saved_connections = await self._get_saved_connections()
self.scan_task = asyncio.create_task(self._periodic_scan())
except DBusError as e:
@@ -97,7 +120,7 @@ class WifiManager:
except asyncio.CancelledError:
pass
if self.bus:
await self.bus.disconnect()
self.bus.disconnect()
async def request_scan(self) -> None:
try:
@@ -123,6 +146,12 @@ class WifiManager:
try:
nm_iface = await self._get_interface(NM, path, NM_CONNECTION_IFACE)
await nm_iface.call_delete()
if self._current_connection_ssid == ssid:
self._current_connection_ssid = None
if ssid in self.saved_connections:
del self.saved_connections[ssid]
return True
except DBusError as e:
cloudlog.error(f"Failed to delete connection for SSID: {ssid}. Error: {e}")
@@ -143,6 +172,18 @@ class WifiManager:
async def connect_to_network(self, ssid: str, password: str = None, bssid: str = None, is_hidden: bool = False) -> None:
"""Connect to a selected Wi-Fi network."""
try:
self._current_connection_ssid = ssid
if ssid in self.saved_connections:
# Forget old connection if new password provided
if password:
await self.forget_connection(ssid)
await asyncio.sleep(0.2) # NetworkManager delay
else:
# Just activate existing connection
await self.activate_connection(ssid)
return
connection = {
'connection': {
'type': Variant('s', '802-11-wireless'),
@@ -172,8 +213,8 @@ class WifiManager:
nm_iface = await self._get_interface(NM, NM_PATH, NM_IFACE)
await nm_iface.call_add_and_activate_connection(connection, self.device_path, "/")
await self._update_connection_status()
except DBusError as e:
self._current_connection_ssid = None
cloudlog.error(f"Error connecting to network: {e}")
def is_saved(self, ssid: str) -> bool:
@@ -195,6 +236,160 @@ class WifiManager:
return False
async def add_tethering_connection(self, ssid: str, password: str = "12345678") -> bool:
"""Create a WiFi tethering connection."""
if len(password) < 8:
print("Tethering password must be at least 8 characters")
return False
try:
# First, check if a hotspot connection already exists
settings_iface = await self._get_interface(NM, NM_SETTINGS_PATH, NM_SETTINGS_IFACE)
connection_paths = await settings_iface.call_list_connections()
# Look for an existing hotspot connection
for path in connection_paths:
try:
settings = await self._get_connection_settings(path)
conn_type = settings.get('connection', {}).get('type', Variant('s', '')).value
wifi_mode = settings.get('802-11-wireless', {}).get('mode', Variant('s', '')).value
if conn_type == '802-11-wireless' and wifi_mode == 'ap':
# Extract the SSID to check
connection_ssid = self._extract_ssid(settings)
if connection_ssid == ssid:
return True
except DBusError:
continue
connection = {
'connection': {
'id': Variant('s', 'Hotspot'),
'uuid': Variant('s', str(uuid.uuid4())),
'type': Variant('s', '802-11-wireless'),
'interface-name': Variant('s', 'wlan0'),
'autoconnect': Variant('b', False),
},
'802-11-wireless': {
'band': Variant('s', 'bg'),
'mode': Variant('s', 'ap'),
'ssid': Variant('ay', ssid.encode('utf-8')),
},
'802-11-wireless-security': {
'group': Variant('as', ['ccmp']),
'key-mgmt': Variant('s', 'wpa-psk'),
'pairwise': Variant('as', ['ccmp']),
'proto': Variant('as', ['rsn']),
'psk': Variant('s', password),
},
'ipv4': {
'method': Variant('s', 'shared'),
'address-data': Variant('aa{sv}', [{'address': Variant('s', TETHERING_IP_ADDRESS), 'prefix': Variant('u', 24)}]),
'gateway': Variant('s', TETHERING_IP_ADDRESS),
'never-default': Variant('b', True),
},
'ipv6': {
'method': Variant('s', 'ignore'),
},
}
settings_iface = await self._get_interface(NM, NM_SETTINGS_PATH, NM_SETTINGS_IFACE)
new_connection = await settings_iface.call_add_connection(connection)
print(f"Added tethering connection with path: {new_connection}")
return True
except DBusError as e:
print(f"Failed to add tethering connection: {e}")
return False
except Exception as e:
print(f"Unexpected error adding tethering connection: {e}")
return False
async def get_tethering_password(self) -> str:
"""Get the current tethering password."""
try:
hotspot_path = self.saved_connections.get(self._tethering_ssid)
if hotspot_path:
conn_iface = await self._get_interface(NM, hotspot_path, NM_CONNECTION_IFACE)
secrets = await conn_iface.call_get_secrets('802-11-wireless-security')
if secrets and '802-11-wireless-security' in secrets:
psk = secrets.get('802-11-wireless-security', {}).get('psk', Variant('s', '')).value
return str(psk) if psk is not None else ""
return ""
except DBusError as e:
print(f"Failed to get tethering password: {e}")
return ""
except Exception as e:
print(f"Unexpected error getting tethering password: {e}")
return ""
async def set_tethering_password(self, password: str) -> bool:
"""Set the tethering password."""
if len(password) < 8:
cloudlog.error("Tethering password must be at least 8 characters")
return False
try:
hotspot_path = self.saved_connections.get(self._tethering_ssid)
if not hotspot_path:
print("No hotspot connection found")
return False
# Update the connection settings with new password
settings = await self._get_connection_settings(hotspot_path)
if '802-11-wireless-security' not in settings:
settings['802-11-wireless-security'] = {}
settings['802-11-wireless-security']['psk'] = Variant('s', password)
# Apply changes
conn_iface = await self._get_interface(NM, hotspot_path, NM_CONNECTION_IFACE)
await conn_iface.call_update(settings)
# Check if connection is active and restart if needed
is_active = False
nm_iface = await self._get_interface(NM, NM_PATH, NM_IFACE)
active_connections = await nm_iface.get_active_connections()
for conn_path in active_connections:
props_iface = await self._get_interface(NM, conn_path, NM_PROPERTIES_IFACE)
conn_id_path = await props_iface.call_get('org.freedesktop.NetworkManager.Connection.Active', 'Connection')
if conn_id_path.value == hotspot_path:
is_active = True
await nm_iface.call_deactivate_connection(conn_path)
break
if is_active:
await nm_iface.call_activate_connection(hotspot_path, self.device_path, "/")
print("Tethering password updated successfully")
return True
except DBusError as e:
print(f"Failed to set tethering password: {e}")
return False
except Exception as e:
print(f"Unexpected error setting tethering password: {e}")
return False
async def is_tethering_active(self) -> bool:
"""Check if tethering is active for the specified SSID."""
try:
hotspot_path = self.saved_connections.get(self._tethering_ssid)
if not hotspot_path:
return False
nm_iface = await self._get_interface(NM, NM_PATH, NM_IFACE)
active_connections = await nm_iface.get_active_connections()
for conn_path in active_connections:
props_iface = await self._get_interface(NM, conn_path, NM_PROPERTIES_IFACE)
conn_id_path = await props_iface.call_get('org.freedesktop.NetworkManager.Connection.Active', 'Connection')
if conn_id_path.value == hotspot_path:
return True
return False
except Exception:
return False
async def _periodic_scan(self):
while self.running:
try:
@@ -239,11 +434,22 @@ class WifiManager:
if self.callbacks.activated:
self.callbacks.activated()
asyncio.create_task(self._update_connection_status())
self._current_connection_ssid = None
elif new_state in (NMDeviceState.DISCONNECTED, NMDeviceState.NEED_AUTH):
for network in self.networks:
network.is_connected = False
if new_state == NMDeviceState.NEED_AUTH and reason == NM_DEVICE_STATE_REASON_SUPPLICANT_DISCONNECT and self.callbacks.need_auth:
self.callbacks.need_auth()
if self._current_connection_ssid:
self.callbacks.need_auth(self._current_connection_ssid)
else:
# Try to find the network from active_ap_path
for network in self.networks:
if network.path == self.active_ap_path:
self.callbacks.need_auth(network.ssid)
break
else:
# Couldn't identify the network that needs auth
cloudlog.error("Network needs authentication but couldn't identify which one")
def _on_new_connection(self, path: str) -> None:
"""Callback for NewConnection signal."""
@@ -258,6 +464,8 @@ class WifiManager:
del self.saved_connections[ssid]
if self.callbacks.forgotten:
self.callbacks.forgotten()
# Update network list to reflect the removed saved connection
asyncio.create_task(self._update_connection_status())
break
async def _add_saved_connection(self, path: str) -> None:
@@ -266,6 +474,7 @@ class WifiManager:
settings = await self._get_connection_settings(path)
if ssid := self._extract_ssid(settings):
self.saved_connections[ssid] = path
await self._update_connection_status()
except DBusError as e:
cloudlog.error(f"Failed to add connection {path}: {e}")
@@ -323,6 +532,7 @@ class WifiManager:
path=ap_path,
bssid=bssid,
is_connected=self.active_ap_path == ap_path,
is_saved=ssid in self.saved_connections
)
except DBusError as e:
@@ -339,6 +549,9 @@ class WifiManager:
),
)
if self.callbacks.networks_updated:
self.callbacks.networks_updated(copy.deepcopy(self.networks))
async def _get_connection_settings(self, path):
"""Fetch connection settings for a specific connection path."""
try:
@@ -383,7 +596,8 @@ class WifiManager:
if flags == 0 and not (wpa_flags or rsn_flags):
return SecurityType.OPEN
if rsn_flags & 0x200: # SAE (WPA3 Personal)
return SecurityType.WPA3
# TODO: support WPA3
return SecurityType.UNSUPPORTED
if rsn_flags: # RSN indicates WPA2 or higher
return SecurityType.WPA2
if wpa_flags: # WPA flags indicate WPA
@@ -426,22 +640,19 @@ class WifiManagerWrapper:
def shutdown(self) -> None:
if self._running:
if self._manager is not None:
self._run_coroutine(self._manager.shutdown())
if self._manager is not None and self._loop:
shutdown_future = asyncio.run_coroutine_threadsafe(self._manager.shutdown(), self._loop)
shutdown_future.result(timeout=3.0)
if self._loop and self._loop.is_running():
self._loop.call_soon_threadsafe(self._loop.stop)
if self._thread and self._thread.is_alive():
self._thread.join(timeout=2.0)
self._running = False
@property
def networks(self) -> list[NetworkInfo]:
"""Get the current list of networks."""
return self._manager.networks if self._manager else []
def is_saved(self, ssid: str) -> bool:
"""Check if a network is saved."""
return self._manager.is_saved(ssid) if self._manager else False
return self._run_coroutine_sync(lambda manager: manager.is_saved(ssid), default=False)
def connect(self):
"""Connect to DBus and start Wi-Fi scanning."""
@@ -479,3 +690,22 @@ class WifiManagerWrapper:
cloudlog.error("WifiManager thread is not running")
return
asyncio.run_coroutine_threadsafe(coro, self._loop)
def _run_coroutine_sync(self, func: Callable[[WifiManager], T], default: T) -> T:
"""Run a function synchronously in the async thread."""
if not self._running or not self._loop or not self._manager:
return default
future = concurrent.futures.Future[T]()
def wrapper(manager: WifiManager) -> None:
try:
future.set_result(func(manager))
except Exception as e:
future.set_exception(e)
try:
self._loop.call_soon_threadsafe(wrapper, self._manager)
return future.result(timeout=1.0)
except Exception as e:
cloudlog.error(f"WifiManagerWrapper property access failed: {e}")
return default
+6 -2
View File
@@ -5,6 +5,7 @@ import sys
import threading
from enum import IntEnum
from openpilot.system.hardware import PC
from openpilot.system.ui.lib.application import gui_app, FontWeight
from openpilot.system.ui.lib.button import gui_button, ButtonStyle
from openpilot.system.ui.lib.label import gui_label, gui_text_box
@@ -31,7 +32,10 @@ class Reset:
self.mode = mode
self.reset_state = ResetState.NONE
def do_reset(self):
def _do_erase(self):
if PC:
return
# Best effort to wipe NVME
os.system(f"sudo umount {NVME}")
os.system(f"yes | sudo mkfs.ext4 {NVME}")
@@ -48,7 +52,7 @@ class Reset:
def start_reset(self):
self.reset_state = ResetState.RESETTING
threading.Timer(0.1, self.do_reset).start()
threading.Timer(0.1, self._do_erase).start()
def render(self, rect: rl.Rectangle):
label_rect = rl.Rectangle(rect.x + 140, rect.y, rect.width - 280, 100)
+349
View File
@@ -0,0 +1,349 @@
#!/usr/bin/env python3
import os
import re
import threading
import time
import urllib.request
from enum import IntEnum
import pyray as rl
from cereal import log
from openpilot.system.hardware import HARDWARE
from openpilot.system.ui.lib.application import gui_app, FontWeight
from openpilot.system.ui.lib.button import gui_button, ButtonStyle
from openpilot.system.ui.lib.label import gui_label, gui_text_box
from openpilot.system.ui.widgets.network import WifiManagerUI, WifiManagerWrapper
from openpilot.system.ui.widgets.keyboard import Keyboard
NetworkType = log.DeviceState.NetworkType
MARGIN = 50
TITLE_FONT_SIZE = 116
TITLE_FONT_WEIGHT = FontWeight.MEDIUM
NEXT_BUTTON_WIDTH = 310
BODY_FONT_SIZE = 96
BUTTON_HEIGHT = 160
BUTTON_SPACING = 50
OPENPILOT_URL = "https://openpilot.comma.ai"
USER_AGENT = f"AGNOSSetup-{HARDWARE.get_os_version()}"
class SetupState(IntEnum):
LOW_VOLTAGE = 0
GETTING_STARTED = 1
NETWORK_SETUP = 2
SOFTWARE_SELECTION = 3
CUSTOM_URL = 4
DOWNLOADING = 5
DOWNLOAD_FAILED = 6
class Setup:
def __init__(self):
self.state = SetupState.GETTING_STARTED
self.network_check_thread = None
self.network_connected = threading.Event()
self.wifi_connected = threading.Event()
self.stop_network_check_thread = threading.Event()
self.failed_url = ""
self.failed_reason = ""
self.download_url = ""
self.download_progress = 0
self.download_thread = None
self.wifi_manager = WifiManagerWrapper()
self.wifi_ui = WifiManagerUI(self.wifi_manager)
self.keyboard = Keyboard()
self.selected_radio = None
self.warning = gui_app.texture("icons/warning.png", 150, 150)
self.checkmark = gui_app.texture("icons/circled_check.png", 100, 100)
try:
with open("/sys/class/hwmon/hwmon1/in1_input") as f:
voltage = float(f.read().strip()) / 1000.0
if voltage < 7:
self.state = SetupState.LOW_VOLTAGE
except (FileNotFoundError, ValueError):
self.state = SetupState.LOW_VOLTAGE
def render(self, rect: rl.Rectangle):
if self.state == SetupState.LOW_VOLTAGE:
self.render_low_voltage(rect)
elif self.state == SetupState.GETTING_STARTED:
self.render_getting_started(rect)
elif self.state == SetupState.NETWORK_SETUP:
self.render_network_setup(rect)
elif self.state == SetupState.SOFTWARE_SELECTION:
self.render_software_selection(rect)
elif self.state == SetupState.CUSTOM_URL:
self.render_custom_url()
elif self.state == SetupState.DOWNLOADING:
self.render_downloading(rect)
elif self.state == SetupState.DOWNLOAD_FAILED:
self.render_download_failed(rect)
def render_low_voltage(self, rect: rl.Rectangle):
rl.draw_texture(self.warning, int(rect.x + 150), int(rect.y + 110), rl.WHITE)
title_rect = rl.Rectangle(rect.x + 150, rect.y + 110 + 150 + 100, rect.width - 500 - 150, TITLE_FONT_SIZE)
gui_label(title_rect, "WARNING: Low Voltage", TITLE_FONT_SIZE, rl.Color(255, 89, 79, 255), FontWeight.MEDIUM)
body_rect = rl.Rectangle(rect.x + 150, rect.y + 110 + 150 + 100 + TITLE_FONT_SIZE + 25, rect.width - 500 - 150, BODY_FONT_SIZE * 3)
gui_text_box(body_rect, "Power your device in a car with a harness or proceed at your own risk.", BODY_FONT_SIZE)
button_width = (rect.width - MARGIN * 3) / 2
button_y = rect.height - MARGIN - BUTTON_HEIGHT
if gui_button(rl.Rectangle(rect.x + MARGIN, button_y, button_width, BUTTON_HEIGHT), "Power off"):
HARDWARE.shutdown()
if gui_button(rl.Rectangle(rect.x + MARGIN * 2 + button_width, button_y, button_width, BUTTON_HEIGHT), "Continue"):
self.state = SetupState.GETTING_STARTED
def render_getting_started(self, rect: rl.Rectangle):
title_rect = rl.Rectangle(rect.x + 165, rect.y + 280, rect.width - 265, TITLE_FONT_SIZE)
gui_label(title_rect, "Getting Started", TITLE_FONT_SIZE, font_weight=FontWeight.MEDIUM)
desc_rect = rl.Rectangle(rect.x + 165, rect.y + 280 + TITLE_FONT_SIZE + 90, rect.width - 500, BODY_FONT_SIZE * 3)
gui_text_box(desc_rect, "Before we get on the road, let's finish installation and cover some details.", BODY_FONT_SIZE)
btn_rect = rl.Rectangle(rect.width - NEXT_BUTTON_WIDTH, 0, NEXT_BUTTON_WIDTH, rect.height)
ret = gui_button(btn_rect, "", button_style=ButtonStyle.PRIMARY, border_radius=0)
triangle = gui_app.texture("images/button_continue_triangle.png", 54, int(btn_rect.height))
rl.draw_texture_v(triangle, rl.Vector2(btn_rect.x + btn_rect.width / 2 - triangle.width / 2, btn_rect.height / 2 - triangle.height / 2), rl.WHITE)
if ret:
self.state = SetupState.NETWORK_SETUP
self.wifi_manager.request_scan()
self.start_network_check()
def check_network_connectivity(self):
while not self.stop_network_check_thread.is_set():
if self.state == SetupState.NETWORK_SETUP:
try:
urllib.request.urlopen(OPENPILOT_URL, timeout=2)
self.network_connected.set()
if HARDWARE.get_network_type() == NetworkType.wifi:
self.wifi_connected.set()
else:
self.wifi_connected.clear()
except Exception:
self.network_connected.clear()
time.sleep(1)
def start_network_check(self):
if self.network_check_thread is None or not self.network_check_thread.is_alive():
self.network_check_thread = threading.Thread(target=self.check_network_connectivity, daemon=True)
self.network_check_thread.start()
def close(self):
if self.network_check_thread is not None:
self.stop_network_check_thread.set()
self.network_check_thread.join()
def render_network_setup(self, rect: rl.Rectangle):
if self.wifi_ui.require_full_screen:
self.wifi_ui.render(rect)
return
title_rect = rl.Rectangle(rect.x + MARGIN, rect.y + MARGIN, rect.width - MARGIN * 2, TITLE_FONT_SIZE)
gui_label(title_rect, "Connect to Wi-Fi", TITLE_FONT_SIZE, font_weight=FontWeight.MEDIUM)
wifi_rect = rl.Rectangle(rect.x + MARGIN, rect.y + TITLE_FONT_SIZE + MARGIN + 25, rect.width - MARGIN * 2,
rect.height - TITLE_FONT_SIZE - 25 - BUTTON_HEIGHT - MARGIN * 3)
rl.draw_rectangle_rounded(wifi_rect, 0.05, 10, rl.Color(51, 51, 51, 255))
wifi_content_rect = rl.Rectangle(wifi_rect.x + MARGIN, wifi_rect.y, wifi_rect.width - MARGIN * 2, wifi_rect.height)
self.wifi_ui.render(wifi_content_rect)
button_width = (rect.width - BUTTON_SPACING - MARGIN * 2) / 2
button_y = rect.height - BUTTON_HEIGHT - MARGIN
if gui_button(rl.Rectangle(rect.x + MARGIN, button_y, button_width, BUTTON_HEIGHT), "Back"):
self.state = SetupState.GETTING_STARTED
# Check network connectivity status
continue_enabled = self.network_connected.is_set()
continue_text = ("Continue" if self.wifi_connected.is_set() else "Continue without Wi-Fi") if continue_enabled else "Waiting for internet"
if gui_button(
rl.Rectangle(rect.x + MARGIN + button_width + BUTTON_SPACING, button_y, button_width, BUTTON_HEIGHT),
continue_text,
button_style=ButtonStyle.PRIMARY if continue_enabled else ButtonStyle.NORMAL,
is_enabled=continue_enabled,
):
self.state = SetupState.SOFTWARE_SELECTION
self.stop_network_check_thread.set()
def render_software_selection(self, rect: rl.Rectangle):
title_rect = rl.Rectangle(rect.x + MARGIN, rect.y + MARGIN, rect.width - MARGIN * 2, TITLE_FONT_SIZE)
gui_label(title_rect, "Choose Software to Install", TITLE_FONT_SIZE, font_weight=FontWeight.MEDIUM)
radio_height = 230
radio_spacing = 30
openpilot_rect = rl.Rectangle(rect.x + MARGIN, rect.y + TITLE_FONT_SIZE + MARGIN * 2, rect.width - MARGIN * 2, radio_height)
openpilot_selected = self.selected_radio == "openpilot"
rl.draw_rectangle_rounded(openpilot_rect, 0.1, 10, rl.Color(70, 91, 234, 255) if openpilot_selected else rl.Color(79, 79, 79, 255))
gui_label(rl.Rectangle(openpilot_rect.x + 100, openpilot_rect.y, openpilot_rect.width - 200, radio_height), "openpilot", BODY_FONT_SIZE)
if openpilot_selected:
checkmark_pos = rl.Vector2(openpilot_rect.x + openpilot_rect.width - 100 - self.checkmark.width,
openpilot_rect.y + radio_height / 2 - self.checkmark.height / 2)
rl.draw_texture_v(self.checkmark, checkmark_pos, rl.WHITE)
custom_rect = rl.Rectangle(rect.x + MARGIN, rect.y + TITLE_FONT_SIZE + MARGIN * 2 + radio_height + radio_spacing, rect.width - MARGIN * 2, radio_height)
custom_selected = self.selected_radio == "custom"
rl.draw_rectangle_rounded(custom_rect, 0.1, 10, rl.Color(70, 91, 234, 255) if custom_selected else rl.Color(79, 79, 79, 255))
gui_label(rl.Rectangle(custom_rect.x + 100, custom_rect.y, custom_rect.width - 200, radio_height), "Custom Software", BODY_FONT_SIZE)
if custom_selected:
checkmark_pos = rl.Vector2(custom_rect.x + custom_rect.width - 100 - self.checkmark.width, custom_rect.y + radio_height / 2 - self.checkmark.height / 2)
rl.draw_texture_v(self.checkmark, checkmark_pos, rl.WHITE)
mouse_pos = rl.get_mouse_position()
if rl.is_mouse_button_released(rl.MouseButton.MOUSE_BUTTON_LEFT):
if rl.check_collision_point_rec(mouse_pos, openpilot_rect):
self.selected_radio = "openpilot"
elif rl.check_collision_point_rec(mouse_pos, custom_rect):
self.selected_radio = "custom"
button_width = (rect.width - BUTTON_SPACING - MARGIN * 2) / 2
button_y = rect.height - BUTTON_HEIGHT - MARGIN
if gui_button(rl.Rectangle(rect.x + MARGIN, button_y, button_width, BUTTON_HEIGHT), "Back"):
self.state = SetupState.NETWORK_SETUP
continue_enabled = self.selected_radio is not None
if gui_button(
rl.Rectangle(rect.x + MARGIN + button_width + BUTTON_SPACING, button_y, button_width, BUTTON_HEIGHT),
"Continue",
button_style=ButtonStyle.PRIMARY,
is_enabled=continue_enabled,
):
if continue_enabled:
if self.selected_radio == "openpilot":
self.download(OPENPILOT_URL)
else:
self.state = SetupState.CUSTOM_URL
def render_downloading(self, rect: rl.Rectangle):
title_rect = rl.Rectangle(rect.x, rect.y + rect.height / 2 - TITLE_FONT_SIZE / 2, rect.width, TITLE_FONT_SIZE)
gui_label(title_rect, "Downloading...", TITLE_FONT_SIZE, font_weight=FontWeight.MEDIUM, alignment=rl.GuiTextAlignment.TEXT_ALIGN_CENTER)
def render_download_failed(self, rect: rl.Rectangle):
title_rect = rl.Rectangle(rect.x + 117, rect.y + 185, rect.width - 117, TITLE_FONT_SIZE)
gui_label(title_rect, "Download Failed", TITLE_FONT_SIZE, font_weight=FontWeight.MEDIUM)
url_rect = rl.Rectangle(rect.x + 117, rect.y + 185 + TITLE_FONT_SIZE + 67, rect.width - 117 - 100, 64)
gui_label(url_rect, self.failed_url, 64, font_weight=FontWeight.NORMAL)
error_rect = rl.Rectangle(rect.x + 117, rect.y + 185 + TITLE_FONT_SIZE + 67 + 64 + 48,
rect.width - 117 - 100, rect.height - 185 + TITLE_FONT_SIZE + 67 + 64 + 48 - BUTTON_HEIGHT - MARGIN * 2)
gui_text_box(error_rect, self.failed_reason, BODY_FONT_SIZE)
button_width = (rect.width - BUTTON_SPACING - MARGIN * 2) / 2
button_y = rect.height - BUTTON_HEIGHT - MARGIN
if gui_button(rl.Rectangle(rect.x + MARGIN, button_y, button_width, BUTTON_HEIGHT), "Reboot device"):
HARDWARE.reboot()
if gui_button(rl.Rectangle(rect.x + MARGIN + button_width + BUTTON_SPACING, button_y, button_width, BUTTON_HEIGHT), "Start over",
button_style=ButtonStyle.PRIMARY):
self.state = SetupState.GETTING_STARTED
def render_custom_url(self):
result = self.keyboard.render("Enter URL", "for Custom Software")
# Enter pressed
if result == 1:
url = self.keyboard.text
self.keyboard.clear()
if url:
self.download(url)
# Cancel pressed
elif result == 0:
self.state = SetupState.SOFTWARE_SELECTION
def download(self, url: str):
# autocomplete incomplete URLs
if re.match("^([^/.]+)/([^/]+)$", url):
url = f"https://installer.comma.ai/{url}"
self.download_url = url
self.state = SetupState.DOWNLOADING
self.download_thread = threading.Thread(target=self._download_thread, daemon=True)
self.download_thread.start()
def _download_thread(self):
try:
import tempfile
_, tmpfile = tempfile.mkstemp(prefix="installer_")
headers = {"User-Agent": USER_AGENT, "X-openpilot-serial": HARDWARE.get_serial()}
req = urllib.request.Request(self.download_url, headers=headers)
with open(tmpfile, 'wb') as f, urllib.request.urlopen(req, timeout=30) as response:
total_size = int(response.headers.get('content-length', 0))
downloaded = 0
block_size = 8192
while True:
buffer = response.read(block_size)
if not buffer:
break
downloaded += len(buffer)
f.write(buffer)
if total_size:
self.download_progress = int(downloaded * 100 / total_size)
is_elf = False
with open(tmpfile, 'rb') as f:
header = f.read(4)
is_elf = header == b'\x7fELF'
if not is_elf:
self.download_failed(self.download_url, "No custom software found at this URL.")
return
os.rename(tmpfile, "/tmp/installer")
os.chmod("/tmp/installer", 0o755)
with open("/tmp/installer_url", "w") as f:
f.write(self.download_url)
gui_app.request_close()
except Exception:
error_msg = "Ensure the entered URL is valid, and the device's internet connection is good."
self.download_failed(self.download_url, error_msg)
def download_failed(self, url: str, reason: str):
self.failed_url = url
self.failed_reason = reason
self.state = SetupState.DOWNLOAD_FAILED
def main():
try:
gui_app.init_window("Setup")
setup = Setup()
for _ in gui_app.render():
setup.render(rl.Rectangle(0, 0, gui_app.width, gui_app.height))
setup.close()
except Exception as e:
print(f"Setup error: {e}")
finally:
gui_app.close()
if __name__ == "__main__":
main()
+7 -6
View File
@@ -1,10 +1,8 @@
#!/usr/bin/env python3
import pyray as rl
import os
import threading
import time
from openpilot.common.basedir import BASEDIR
from openpilot.system.ui.lib.application import gui_app
from openpilot.system.ui.lib.window import BaseWindow
from openpilot.system.ui.text import wrap_text
@@ -26,9 +24,8 @@ def clamp(value, min_value, max_value):
class SpinnerRenderer:
def __init__(self):
self._comma_texture = gui_app.load_texture_from_image(os.path.join(BASEDIR, "selfdrive/assets/img_spinner_comma.png"), TEXTURE_SIZE, TEXTURE_SIZE)
self._spinner_texture = gui_app.load_texture_from_image(os.path.join(BASEDIR, "selfdrive/assets/img_spinner_track.png"), TEXTURE_SIZE, TEXTURE_SIZE,
alpha_premultiply=True)
self._comma_texture = gui_app.texture("images/spinner_comma.png", TEXTURE_SIZE, TEXTURE_SIZE)
self._spinner_texture = gui_app.texture("images/spinner_track.png", TEXTURE_SIZE, TEXTURE_SIZE, alpha_premultiply=True)
self._rotation = 0.0
self._progress: int | None = None
self._wrapped_lines: list[str] = []
@@ -101,7 +98,11 @@ class Spinner(BaseWindow[SpinnerRenderer]):
self.update(str(round(100 * cur / total)))
if __name__ == "__main__":
def main():
with Spinner() as s:
s.update("Spinner text")
time.sleep(5)
if __name__ == "__main__":
main()
+1 -1
View File
@@ -88,4 +88,4 @@ class TextWindow(BaseWindow[TextWindowRenderer]):
if __name__ == "__main__":
with TextWindow(DEMO_TEXT):
time.sleep(5)
time.sleep(30)
+13 -39
View File
@@ -9,6 +9,9 @@ from openpilot.system.hardware import HARDWARE
from openpilot.system.ui.lib.application import gui_app, FontWeight
from openpilot.system.ui.lib.button import gui_button, ButtonStyle
from openpilot.system.ui.lib.label import gui_text_box, gui_label
from openpilot.system.ui.lib.wifi_manager import WifiManagerWrapper
from openpilot.system.ui.widgets.network import WifiManagerUI
# Constants
MARGIN = 50
@@ -39,6 +42,8 @@ class Updater:
self.show_reboot_button = False
self.process = None
self.update_thread = None
self.wifi_manager = WifiManagerWrapper()
self.wifi_manager_ui = WifiManagerUI(self.wifi_manager)
def install_update(self):
self.current_screen = Screen.PROGRESS
@@ -79,8 +84,9 @@ class Updater:
gui_label(title_rect, "Update Required", TITLE_FONT_SIZE, font_weight=FontWeight.BOLD)
# Description
desc_text = "An operating system update is required. Connect your device to Wi-Fi for the fastest update experience. \
The download size is approximately 1GB."
desc_text = ("An operating system update is required. Connect your device to Wi-Fi for the fastest update experience. " +
"The download size is approximately 1GB.")
desc_rect = rl.Rectangle(MARGIN + 50, 250 + TITLE_FONT_SIZE + 75, gui_app.width - MARGIN * 2 - 100, BODY_FONT_SIZE * 3)
gui_text_box(desc_rect, desc_text, BODY_FONT_SIZE)
@@ -101,49 +107,17 @@ class Updater:
return # Return to avoid further processing after action
def render_wifi_screen(self):
# Title and back button
title_rect = rl.Rectangle(MARGIN + 50, MARGIN, gui_app.width - MARGIN * 2 - 100, 60)
gui_label(title_rect, "Wi-Fi Networks", 60, font_weight=FontWeight.BOLD)
# Draw the Wi-Fi manager UI
wifi_rect = rl.Rectangle(MARGIN + 50, MARGIN, gui_app.width - MARGIN * 2 - 100, gui_app.height - MARGIN * 2 - BUTTON_HEIGHT - 20)
self.wifi_manager_ui.render(wifi_rect)
if self.wifi_manager_ui.require_full_screen:
return
back_button_rect = rl.Rectangle(MARGIN, gui_app.height - MARGIN - BUTTON_HEIGHT, BUTTON_WIDTH, BUTTON_HEIGHT)
if gui_button(back_button_rect, "Back"):
self.current_screen = Screen.PROMPT
return # Return to avoid processing other interactions after screen change
# Draw placeholder for WiFi implementation
placeholder_rect = rl.Rectangle(
MARGIN,
title_rect.y + title_rect.height + MARGIN,
gui_app.width - MARGIN * 2,
gui_app.height - title_rect.height - MARGIN * 3 - BUTTON_HEIGHT
)
# Draw rounded rectangle background
rl.draw_rectangle_rounded(
placeholder_rect,
0.1,
10,
rl.Color(41, 41, 41, 255)
)
# Draw placeholder text
placeholder_text = "WiFi Implementation Placeholder"
text_size = rl.measure_text_ex(gui_app.font(), placeholder_text, 80, 1)
text_pos = rl.Vector2(
placeholder_rect.x + (placeholder_rect.width - text_size.x) / 2,
placeholder_rect.y + (placeholder_rect.height - text_size.y) / 2
)
rl.draw_text_ex(gui_app.font(), placeholder_text, text_pos, 80, 1, rl.WHITE)
# Draw instructions
instructions_text = "Real WiFi functionality would be implemented here"
instructions_size = rl.measure_text_ex(gui_app.font(), instructions_text, 40, 1)
instructions_pos = rl.Vector2(
placeholder_rect.x + (placeholder_rect.width - instructions_size.x) / 2,
text_pos.y + text_size.y + 20
)
rl.draw_text_ex(gui_app.font(), instructions_text, instructions_pos, 40, 1, rl.GRAY)
def render_progress_screen(self):
title_rect = rl.Rectangle(MARGIN + 100, 330, gui_app.width - MARGIN * 2 - 200, 100)
gui_label(title_rect, self.progress_text, 90, font_weight=FontWeight.SEMI_BOLD)
+204
View File
@@ -0,0 +1,204 @@
import pyray as rl
from openpilot.system.hardware import TICI
from msgq.visionipc import VisionIpcClient, VisionStreamType, VisionBuf
from openpilot.system.ui.lib.application import gui_app
from openpilot.system.ui.lib.egl import init_egl, create_egl_image, destroy_egl_image, bind_egl_image_to_texture, EGLImage
VERTEX_SHADER = """
#version 300 es
in vec3 vertexPosition;
in vec2 vertexTexCoord;
in vec3 vertexNormal;
in vec4 vertexColor;
uniform mat4 mvp;
out vec2 fragTexCoord;
out vec4 fragColor;
void main() {
fragTexCoord = vertexTexCoord;
fragColor = vertexColor;
gl_Position = mvp * vec4(vertexPosition, 1.0);
}
"""
# Choose fragment shader based on platform capabilities
if TICI:
FRAME_FRAGMENT_SHADER = """
#version 300 es
#extension GL_OES_EGL_image_external_essl3 : enable
precision mediump float;
in vec2 fragTexCoord;
uniform samplerExternalOES texture0;
out vec4 fragColor;
void main() {
vec4 color = texture(texture0, fragTexCoord);
fragColor = vec4(pow(color.rgb, vec3(1.0/1.28)), color.a);
}
"""
else:
FRAME_FRAGMENT_SHADER = """
#version 300 es
precision mediump float;
in vec2 fragTexCoord;
uniform sampler2D texture0;
uniform sampler2D texture1;
out vec4 fragColor;
void main() {
float y = texture(texture0, fragTexCoord).r;
vec2 uv = texture(texture1, fragTexCoord).ra - 0.5;
fragColor = vec4(y + 1.402*uv.y, y - 0.344*uv.x - 0.714*uv.y, y + 1.772*uv.x, 1.0);
}
"""
class CameraView:
def __init__(self, name: str, stream_type: VisionStreamType):
self.client = VisionIpcClient(name, stream_type, False)
self.shader = rl.load_shader_from_memory(VERTEX_SHADER, FRAME_FRAGMENT_SHADER)
self.frame: VisionBuf | None = None
self.texture_y: rl.Texture | None = None
self.texture_uv: rl.Texture | None = None
# EGL resources
self.egl_images: dict[int, EGLImage] = {}
self.egl_texture: rl.Texture | None = None
# Initialize EGL for zero-copy rendering on TICI
if TICI:
if not init_egl():
raise RuntimeError("Failed to initialize EGL")
# Create a 1x1 pixel placeholder texture for EGL image binding
temp_image = rl.gen_image_color(1, 1, rl.BLACK)
self.egl_texture = rl.load_texture_from_image(temp_image)
rl.unload_image(temp_image)
def close(self) -> None:
self._clear_textures()
# Clean up EGL texture
if TICI and self.egl_texture:
rl.unload_texture(self.egl_texture)
self.egl_texture = None
# Clean up shader
if self.shader and self.shader.id:
rl.unload_shader(self.shader)
def render(self, rect: rl.Rectangle):
if not self._ensure_connection():
return
# Try to get a new buffer without blocking
buffer = self.client.recv(timeout_ms=0)
if buffer:
self.frame = buffer
if not self.frame:
return
# Calculate scaling to maintain aspect ratio
scale = min(rect.width / self.frame.width, rect.height / self.frame.height)
x_offset = rect.x + (rect.width - (self.frame.width * scale)) / 2
y_offset = rect.y + (rect.height - (self.frame.height * scale)) / 2
src_rect = rl.Rectangle(0, 0, float(self.frame.width), float(self.frame.height))
dst_rect = rl.Rectangle(x_offset, y_offset, self.frame.width * scale, self.frame.height * scale)
# Render with appropriate method
if TICI:
self._render_egl(src_rect, dst_rect)
else:
self._render_textures(src_rect, dst_rect)
def _render_egl(self, src_rect: rl.Rectangle, dst_rect: rl.Rectangle) -> None:
"""Render using EGL for direct buffer access"""
if self.frame is None or self.egl_texture is None:
return
idx = self.frame.idx
egl_image = self.egl_images.get(idx)
# Create EGL image if needed
if egl_image is None:
egl_image = create_egl_image(self.frame.width, self.frame.height, self.frame.stride, self.frame.fd, self.frame.uv_offset)
if egl_image:
self.egl_images[idx] = egl_image
else:
return
# Update texture dimensions to match current frame
self.egl_texture.width = self.frame.width
self.egl_texture.height = self.frame.height
# Bind the EGL image to our texture
bind_egl_image_to_texture(self.egl_texture.id, egl_image)
# Render with shader
rl.begin_shader_mode(self.shader)
rl.draw_texture_pro(self.egl_texture, src_rect, dst_rect, rl.Vector2(0, 0), 0.0, rl.WHITE)
rl.end_shader_mode()
def _render_textures(self, src_rect: rl.Rectangle, dst_rect: rl.Rectangle) -> None:
"""Render using texture copies"""
if not self.texture_y or not self.texture_uv or self.frame is None:
return
# Update textures with new frame data
y_data = self.frame.data[: self.frame.uv_offset]
uv_data = self.frame.data[self.frame.uv_offset :]
rl.update_texture(self.texture_y, rl.ffi.cast("void *", y_data.ctypes.data))
rl.update_texture(self.texture_uv, rl.ffi.cast("void *", uv_data.ctypes.data))
# Render with shader
rl.begin_shader_mode(self.shader)
rl.set_shader_value_texture(self.shader, rl.get_shader_location(self.shader, "texture1"), self.texture_uv)
rl.draw_texture_pro(self.texture_y, src_rect, dst_rect, rl.Vector2(0, 0), 0.0, rl.WHITE)
rl.end_shader_mode()
def _ensure_connection(self) -> bool:
if not self.client.is_connected():
self.frame = None
if not self.client.connect(False) or not self.client.num_buffers:
return False
self._clear_textures()
if not TICI:
self.texture_y = rl.load_texture_from_image(rl.Image(None, int(self.client.stride),
int(self.client.height), 1, rl.PixelFormat.PIXELFORMAT_UNCOMPRESSED_GRAYSCALE))
self.texture_uv = rl.load_texture_from_image(rl.Image(None, int(self.client.stride // 2),
int(self.client.height // 2), 1, rl.PixelFormat.PIXELFORMAT_UNCOMPRESSED_GRAY_ALPHA))
return True
def _clear_textures(self):
if self.texture_y and self.texture_y.id:
rl.unload_texture(self.texture_y)
self.texture_y = None
if self.texture_uv and self.texture_uv.id:
rl.unload_texture(self.texture_uv)
self.texture_uv = None
# Clean up EGL resources
if TICI:
for data in self.egl_images.values():
destroy_egl_image(data)
self.egl_images = {}
if __name__ == "__main__":
gui_app.init_window("watch3")
road_camera_view = CameraView("camerad", VisionStreamType.VISION_STREAM_ROAD)
driver_camera_view = CameraView("camerad", VisionStreamType.VISION_STREAM_DRIVER)
wide_road_camera_view = CameraView("camerad", VisionStreamType.VISION_STREAM_WIDE_ROAD)
try:
for _ in gui_app.render():
road_camera_view.render(rl.Rectangle(gui_app.width // 4, 0, gui_app.width // 2, gui_app.height // 2))
driver_camera_view.render(rl.Rectangle(0, gui_app.height // 2, gui_app.width // 2, gui_app.height // 2))
wide_road_camera_view.render(rl.Rectangle(gui_app.width // 2, gui_app.height // 2, gui_app.width // 2, gui_app.height // 2))
finally:
road_camera_view.close()
driver_camera_view.close()
wide_road_camera_view.close()
+6 -11
View File
@@ -1,4 +1,5 @@
import pyray as rl
from openpilot.system.ui.lib.application import gui_app
from openpilot.system.ui.lib.button import gui_button, ButtonStyle
from openpilot.system.ui.lib.label import gui_text_box
@@ -11,10 +12,9 @@ TEXT_AREA_HEIGHT_REDUCTION = 200
BACKGROUND_COLOR = rl.Color(27, 27, 27, 255)
def confirm_dialog(rect: rl.Rectangle, message: str, confirm_text: str, cancel_text: str = "Cancel") -> int:
# Calculate dialog position and size, centered within the parent rectangle
dialog_x = rect.x + (rect.width - DIALOG_WIDTH) / 2
dialog_y = rect.y + (rect.height - DIALOG_HEIGHT) / 2
def confirm_dialog(message: str, confirm_text: str, cancel_text: str = "Cancel") -> int:
dialog_x = (gui_app.width - DIALOG_WIDTH) / 2
dialog_y = (gui_app.height - DIALOG_HEIGHT) / 2
dialog_rect = rl.Rectangle(dialog_x, dialog_y, DIALOG_WIDTH, DIALOG_HEIGHT)
# Calculate button positions at the bottom of the dialog
@@ -27,19 +27,14 @@ def confirm_dialog(rect: rl.Rectangle, message: str, confirm_text: str, cancel_t
yes_button = rl.Rectangle(yes_button_x, button_y, button_width, BUTTON_HEIGHT)
# Draw the dialog background
rl.draw_rectangle(
int(dialog_rect.x),
int(dialog_rect.y),
int(dialog_rect.width),
int(dialog_rect.height),
BACKGROUND_COLOR,
)
rl.draw_rectangle_rec(dialog_rect, BACKGROUND_COLOR)
# Draw the message in the dialog, centered
text_rect = rl.Rectangle(dialog_rect.x, dialog_rect.y, dialog_rect.width, dialog_rect.height - TEXT_AREA_HEIGHT_REDUCTION)
gui_text_box(
text_rect,
message,
font_size=88,
alignment=rl.GuiTextAlignment.TEXT_ALIGN_CENTER,
alignment_vertical=rl.GuiTextAlignmentVertical.TEXT_ALIGN_MIDDLE,
)
+161 -46
View File
@@ -1,29 +1,40 @@
import time
from typing import Literal
import pyray as rl
from openpilot.system.ui.lib.button import gui_button
from openpilot.system.ui.lib.application import gui_app, FontWeight
from openpilot.system.ui.lib.button import ButtonStyle, gui_button
from openpilot.system.ui.lib.inputbox import InputBox
from openpilot.system.ui.lib.label import gui_label
KEY_FONT_SIZE = 96
DOUBLE_CLICK_THRESHOLD = 0.5 # seconds
DELETE_REPEAT_DELAY = 0.5
DELETE_REPEAT_INTERVAL = 0.07
# Constants for special keys
CONTENT_MARGIN = 50
BACKSPACE_KEY = "<-"
ENTER_KEY = "Enter"
ENTER_KEY = "->"
SPACE_KEY = " "
SHIFT_KEY = ""
SHIFT_DOWN_KEY = ""
SHIFT_INACTIVE_KEY = "SHIFT_OFF"
SHIFT_ACTIVE_KEY = "SHIFT_ON"
CAPS_LOCK_KEY = "CAPS"
NUMERIC_KEY = "123"
SYMBOL_KEY = "#+="
ABC_KEY = "ABC"
# Define keyboard layouts as a dictionary for easier access
keyboard_layouts = {
KEYBOARD_LAYOUTS = {
"lowercase": [
["q", "w", "e", "r", "t", "y", "u", "i", "o", "p"],
["a", "s", "d", "f", "g", "h", "j", "k", "l"],
[SHIFT_KEY, "z", "x", "c", "v", "b", "n", "m", BACKSPACE_KEY],
[SHIFT_INACTIVE_KEY, "z", "x", "c", "v", "b", "n", "m", BACKSPACE_KEY],
[NUMERIC_KEY, "/", "-", SPACE_KEY, ".", ENTER_KEY],
],
"uppercase": [
["Q", "W", "E", "R", "T", "Y", "U", "I", "O", "P"],
["A", "S", "D", "F", "G", "H", "J", "K", "L"],
[SHIFT_DOWN_KEY, "Z", "X", "C", "V", "B", "N", "M", BACKSPACE_KEY],
[SHIFT_ACTIVE_KEY, "Z", "X", "C", "V", "B", "N", "M", BACKSPACE_KEY],
[NUMERIC_KEY, "/", "-", SPACE_KEY, ".", ENTER_KEY],
],
"numbers": [
@@ -42,37 +53,79 @@ keyboard_layouts = {
class Keyboard:
def __init__(self, max_text_size: int = 255):
self._layout = keyboard_layouts["lowercase"]
def __init__(self, max_text_size: int = 255, min_text_size: int = 0, password_mode: bool = False, show_password_toggle: bool = False):
self._layout_name: Literal["lowercase", "uppercase", "numbers", "specials"] = "lowercase"
self._caps_lock = False
self._last_shift_press_time = 0
self._max_text_size = max_text_size
self._string_pointer = rl.ffi.new("char[]", max_text_size)
self._input_text = ""
self._clear()
self._min_text_size = min_text_size
self._input_box = InputBox(max_text_size)
self._password_mode = password_mode
self._show_password_toggle = show_password_toggle
# Backspace key repeat tracking
self._backspace_pressed: bool = False
self._backspace_press_time: float = 0.0
self._backspace_last_repeat:float = 0.0
self._eye_open_texture = gui_app.texture("icons/eye_open.png", 81, 54)
self._eye_closed_texture = gui_app.texture("icons/eye_closed.png", 81, 54)
self._key_icons = {
BACKSPACE_KEY: gui_app.texture("icons/backspace.png", 80, 80),
SHIFT_INACTIVE_KEY: gui_app.texture("icons/shift.png", 80, 80),
SHIFT_ACTIVE_KEY: gui_app.texture("icons/shift-fill.png", 80, 80),
CAPS_LOCK_KEY: gui_app.texture("icons/capslock-fill.png", 80, 80),
ENTER_KEY: gui_app.texture("icons/arrow-right.png", 80, 80),
}
@property
def text(self):
result = rl.ffi.string(self._string_pointer).decode("utf-8")
self._clear()
return result
return self._input_box.text
def render(self, rect, title, sub_title):
gui_label(rl.Rectangle(rect.x, rect.y, rect.width, 95), title, 90)
gui_label(rl.Rectangle(rect.x, rect.y + 95, rect.width, 60), sub_title, 55, rl.GRAY)
if gui_button(rl.Rectangle(rect.x + rect.width - 300, rect.y, 300, 100), "Cancel"):
self._clear()
def clear(self):
self._layout_name = "lowercase"
self._caps_lock = False
self._input_box.clear()
self._backspace_pressed = False
def render(self, title: str, sub_title: str):
rect = rl.Rectangle(CONTENT_MARGIN, CONTENT_MARGIN, gui_app.width - 2 * CONTENT_MARGIN, gui_app.height - 2 * CONTENT_MARGIN)
gui_label(rl.Rectangle(rect.x, rect.y, rect.width, 95), title, 90, font_weight=FontWeight.BOLD)
gui_label(rl.Rectangle(rect.x, rect.y + 95, rect.width, 60), sub_title, 55, font_weight=FontWeight.NORMAL)
if gui_button(rl.Rectangle(rect.x + rect.width - 386, rect.y, 386, 125), "Cancel"):
self.clear()
return 0
# Text box for input
self._sync_string_pointer()
rl.gui_text_box(rl.Rectangle(rect.x, rect.y + 160, rect.width, 100), self._string_pointer, self._max_text_size, True)
self._input_text = rl.ffi.string(self._string_pointer).decode("utf-8")
# Draw input box and password toggle
input_margin = 25
input_box_rect = rl.Rectangle(rect.x + input_margin, rect.y + 160, rect.width - input_margin, 100)
self._render_input_area(input_box_rect)
# Process backspace key repeat if it's held down
if not rl.is_mouse_button_down(rl.MouseButton.MOUSE_BUTTON_LEFT):
self._backspace_pressed = False
if self._backspace_pressed:
current_time = time.monotonic()
time_since_press = current_time - self._backspace_press_time
# After initial delay, start repeating with shorter intervals
if time_since_press > DELETE_REPEAT_DELAY:
time_since_last_repeat = current_time - self._backspace_last_repeat
if time_since_last_repeat > DELETE_REPEAT_INTERVAL:
self._input_box.delete_char_before_cursor()
self._backspace_last_repeat = current_time
layout = KEYBOARD_LAYOUTS[self._layout_name]
h_space, v_space = 15, 15
row_y_start = rect.y + 300 # Starting Y position for the first row
key_height = (rect.height - 300 - 3 * v_space) / 4
key_max_width = (rect.width - (len(self._layout[2]) - 1) * h_space) / len(self._layout[2])
key_max_width = (rect.width - (len(layout[2]) - 1) * h_space) / len(layout[2])
# Iterate over the rows of keys in the current layout
for row, keys in enumerate(self._layout):
for row, keys in enumerate(layout):
key_width = min((rect.width - (180 if row == 1 else 0) - h_space * (len(keys) - 1)) / len(keys), key_max_width)
start_x = rect.x + (90 if row == 1 else 0)
@@ -84,7 +137,28 @@ class Keyboard:
key_rect = rl.Rectangle(start_x, row_y_start + row * (key_height + v_space), new_width, key_height)
start_x += new_width
if gui_button(key_rect, key):
is_enabled = key != ENTER_KEY or len(self._input_box.text) >= self._min_text_size
result = -1
# Check for backspace key press-and-hold
mouse_pos = rl.get_mouse_position()
mouse_over_key = rl.check_collision_point_rec(mouse_pos, key_rect)
if key == BACKSPACE_KEY and mouse_over_key:
if rl.is_mouse_button_pressed(rl.MouseButton.MOUSE_BUTTON_LEFT):
self._backspace_pressed = True
self._backspace_press_time = time.monotonic()
self._backspace_last_repeat = time.monotonic()
if key in self._key_icons:
if key == SHIFT_ACTIVE_KEY and self._caps_lock:
key = CAPS_LOCK_KEY
texture = self._key_icons[key]
result = gui_button(key_rect, "", icon=texture, button_style=ButtonStyle.PRIMARY if key == ENTER_KEY else ButtonStyle.NORMAL, is_enabled=is_enabled)
else:
result = gui_button(key_rect, key, KEY_FONT_SIZE, is_enabled=is_enabled)
if result:
if key == ENTER_KEY:
return 1
else:
@@ -92,27 +166,68 @@ class Keyboard:
return -1
def _render_input_area(self, input_rect: rl.Rectangle):
if self._show_password_toggle:
self._input_box.set_password_mode(self._password_mode)
self._input_box.render(rl.Rectangle(input_rect.x, input_rect.y, input_rect.width - 100, input_rect.height))
# render eye icon
eye_texture = self._eye_closed_texture if self._password_mode else self._eye_open_texture
eye_rect = rl.Rectangle(input_rect.x + input_rect.width - 90, input_rect.y, 80, input_rect.height)
eye_x = eye_rect.x + (eye_rect.width - eye_texture.width) / 2
eye_y = eye_rect.y + (eye_rect.height - eye_texture.height) / 2
rl.draw_texture_v(eye_texture, rl.Vector2(eye_x, eye_y), rl.WHITE)
# Handle click on eye icon
if rl.is_mouse_button_pressed(rl.MouseButton.MOUSE_BUTTON_LEFT) and rl.check_collision_point_rec(
rl.get_mouse_position(), eye_rect
):
self._password_mode = not self._password_mode
else:
self._input_box.render(input_rect)
rl.draw_line_ex(
rl.Vector2(input_rect.x, input_rect.y + input_rect.height - 2),
rl.Vector2(input_rect.x + input_rect.width, input_rect.y + input_rect.height - 2),
3.0, # 3 pixel thickness
rl.Color(189, 189, 189, 255),
)
def handle_key_press(self, key):
if key in (SHIFT_DOWN_KEY, ABC_KEY):
self._layout = keyboard_layouts["lowercase"]
elif key == SHIFT_KEY:
self._layout = keyboard_layouts["uppercase"]
if key in (CAPS_LOCK_KEY, ABC_KEY):
self._caps_lock = False
self._layout_name = "lowercase"
elif key == SHIFT_INACTIVE_KEY:
self._last_shift_press_time = time.monotonic()
self._layout_name = "uppercase"
elif key == SHIFT_ACTIVE_KEY:
if time.monotonic() - self._last_shift_press_time < DOUBLE_CLICK_THRESHOLD:
self._caps_lock = True
else:
self._layout_name = "lowercase"
elif key == NUMERIC_KEY:
self._layout = keyboard_layouts["numbers"]
self._layout_name = "numbers"
elif key == SYMBOL_KEY:
self._layout = keyboard_layouts["specials"]
elif key == BACKSPACE_KEY and len(self._input_text) > 0:
self._input_text = self._input_text[:-1]
elif key != BACKSPACE_KEY and len(self._input_text) < self._max_text_size:
self._input_text += key
self._layout_name = "specials"
elif key == BACKSPACE_KEY:
self._input_box.delete_char_before_cursor()
else:
self._input_box.add_char_at_cursor(key)
if not self._caps_lock and self._layout_name == "uppercase":
self._layout_name = "lowercase"
def _clear(self):
self._input_text = ''
self._string_pointer[0] = b'\0'
def _sync_string_pointer(self):
"""Sync the C-string pointer with the internal Python string."""
encoded = self._input_text.encode("utf-8")[:self._max_text_size - 1] # Leave room for the null terminator
buffer = rl.ffi.buffer(self._string_pointer)
buffer[:len(encoded)] = encoded
self._string_pointer[len(encoded)] = b'\0' # Null terminator
if __name__ == "__main__":
gui_app.init_window("Keyboard")
keyboard = Keyboard(min_text_size=8, show_password_toggle=True)
for _ in gui_app.render():
result = keyboard.render("Keyboard", "Type here")
if result == 1:
print(f"You typed: {keyboard.text}")
gui_app.request_close()
elif result == 0:
print("Canceled")
gui_app.request_close()
gui_app.close()
+89 -34
View File
@@ -2,17 +2,26 @@ from dataclasses import dataclass
from typing import Literal
import pyray as rl
from openpilot.system.ui.lib.wifi_manager import NetworkInfo, WifiManagerCallbacks, WifiManagerWrapper
from openpilot.system.ui.lib.application import gui_app
from openpilot.system.ui.lib.button import gui_button
from openpilot.system.ui.lib.button import ButtonStyle, gui_button
from openpilot.system.ui.lib.label import gui_label
from openpilot.system.ui.lib.scroll_panel import GuiScrollPanel
from openpilot.system.ui.lib.wifi_manager import NetworkInfo, WifiManagerCallbacks, WifiManagerWrapper, SecurityType
from openpilot.system.ui.widgets.keyboard import Keyboard
from openpilot.system.ui.widgets.confirm_dialog import confirm_dialog
NM_DEVICE_STATE_NEED_AUTH = 60
MIN_PASSWORD_LENGTH = 8
MAX_PASSWORD_LENGTH = 64
ITEM_HEIGHT = 160
ICON_SIZE = 49
STRENGTH_ICONS = [
"icons/wifi_strength_low.png",
"icons/wifi_strength_medium.png",
"icons/wifi_strength_high.png",
"icons/wifi_strength_full.png",
]
@dataclass
class StateIdle:
@@ -46,28 +55,34 @@ class WifiManagerUI:
self.state: UIState = StateIdle()
self.btn_width = 200
self.scroll_panel = GuiScrollPanel()
self.keyboard = Keyboard()
self.keyboard = Keyboard(max_text_size=MAX_PASSWORD_LENGTH, min_text_size=MIN_PASSWORD_LENGTH, show_password_toggle=True)
self._networks: list[NetworkInfo] = []
self.wifi_manager = wifi_manager
self.wifi_manager.set_callbacks(WifiManagerCallbacks(self._on_need_auth, self._on_activated, self._on_forgotten))
self.wifi_manager.set_callbacks(WifiManagerCallbacks(self._on_need_auth, self._on_activated, self._on_forgotten, self._on_network_updated))
self.wifi_manager.start()
self.wifi_manager.connect()
def render(self, rect: rl.Rectangle):
if not self.wifi_manager.networks:
if not self._networks:
gui_label(rect, "Scanning Wi-Fi networks...", 72, alignment=rl.GuiTextAlignment.TEXT_ALIGN_CENTER)
return
match self.state:
case StateNeedsAuth(network):
result = self.keyboard.render(rect, "Enter password", f"for {network.ssid}")
result = self.keyboard.render("Enter password", f"for {network.ssid}")
if result == 1:
self.connect_to_network(network, self.keyboard.text)
password = self.keyboard.text
self.keyboard.clear()
if len(password) >= MIN_PASSWORD_LENGTH:
self.connect_to_network(network, password)
elif result == 0:
self.state = StateIdle()
case StateShowForgetConfirm(network):
result = confirm_dialog(rect, f'Forget Wi-Fi Network "{network.ssid}"?', "Forget")
result = confirm_dialog(f'Forget Wi-Fi Network "{network.ssid}"?', "Forget")
if result == 1:
self.forget_network(network)
elif result == 0:
@@ -76,34 +91,39 @@ class WifiManagerUI:
case _:
self._draw_network_list(rect)
@property
def require_full_screen(self) -> bool:
"""Check if the WiFi UI requires exclusive full-screen rendering."""
return isinstance(self.state, (StateNeedsAuth, StateShowForgetConfirm))
def _draw_network_list(self, rect: rl.Rectangle):
content_rect = rl.Rectangle(rect.x, rect.y, rect.width, len(self.wifi_manager.networks) * ITEM_HEIGHT)
content_rect = rl.Rectangle(rect.x, rect.y, rect.width, len(self._networks) * ITEM_HEIGHT)
offset = self.scroll_panel.handle_scroll(rect, content_rect)
clicked = self.scroll_panel.is_click_valid()
rl.begin_scissor_mode(int(rect.x), int(rect.y), int(rect.width), int(rect.height))
for i, network in enumerate(self.wifi_manager.networks):
for i, network in enumerate(self._networks):
y_offset = rect.y + i * ITEM_HEIGHT + offset.y
item_rect = rl.Rectangle(rect.x, y_offset, rect.width, ITEM_HEIGHT)
if not rl.check_collision_recs(item_rect, rect):
continue
self._draw_network_item(item_rect, network, clicked)
if i < len(self.wifi_manager.networks) - 1:
if i < len(self._networks) - 1:
line_y = int(item_rect.y + item_rect.height - 1)
rl.draw_line(int(item_rect.x), int(line_y), int(item_rect.x + item_rect.width), line_y, rl.LIGHTGRAY)
rl.end_scissor_mode()
def _draw_network_item(self, rect, network: NetworkInfo, clicked: bool):
label_rect = rl.Rectangle(rect.x, rect.y, rect.width - self.btn_width * 2, ITEM_HEIGHT)
state_rect = rl.Rectangle(rect.x + rect.width - self.btn_width * 2 - 150, rect.y, 300, ITEM_HEIGHT)
spacing = 50
ssid_rect = rl.Rectangle(rect.x, rect.y, rect.width - self.btn_width * 2, ITEM_HEIGHT)
signal_icon_rect = rl.Rectangle(rect.x + rect.width - ICON_SIZE, rect.y + (ITEM_HEIGHT - ICON_SIZE) / 2, ICON_SIZE, ICON_SIZE)
security_icon_rect = rl.Rectangle(signal_icon_rect.x - spacing - ICON_SIZE, rect.y + (ITEM_HEIGHT - ICON_SIZE) / 2, ICON_SIZE, ICON_SIZE)
gui_label(label_rect, network.ssid, 55)
gui_label(ssid_rect, network.ssid, 55)
status_text = ""
if network.is_connected:
status_text = "Connected"
match self.state:
case StateConnecting(network=connecting):
if connecting.ssid == network.ssid:
@@ -111,41 +131,76 @@ class WifiManagerUI:
case StateForgetting(network=forgetting):
if forgetting.ssid == network.ssid:
status_text = "FORGETTING..."
if status_text:
rl.gui_label(state_rect, status_text)
status_text_rect = rl.Rectangle(security_icon_rect.x - 410, rect.y, 410, ITEM_HEIGHT)
gui_label(status_text_rect, status_text, font_size=48, alignment=rl.GuiTextAlignment.TEXT_ALIGN_CENTER)
else:
# If the network is saved, show the "Forget" button
if network.is_saved:
forget_btn_rect = rl.Rectangle(security_icon_rect.x - self.btn_width - spacing,
rect.y + (ITEM_HEIGHT - 80) / 2,
self.btn_width,
80,
)
if isinstance(self.state, StateIdle) and gui_button(forget_btn_rect, "Forget", button_style=ButtonStyle.ACTION) and clicked:
self.state = StateShowForgetConfirm(network)
# If the network is saved, show the "Forget" button
if self.wifi_manager.is_saved(network.ssid):
forget_btn_rect = rl.Rectangle(
rect.x + rect.width - self.btn_width,
rect.y + (ITEM_HEIGHT - 80) / 2,
self.btn_width,
80,
)
if isinstance(self.state, StateIdle) and gui_button(forget_btn_rect, "Forget") and clicked:
self.state = StateShowForgetConfirm(network)
self._draw_status_icon(security_icon_rect, network)
self._draw_signal_strength_icon(signal_icon_rect, network)
if isinstance(self.state, StateIdle) and rl.check_collision_point_rec(rl.get_mouse_position(), label_rect) and clicked:
if not self.wifi_manager.is_saved(network.ssid):
if isinstance(self.state, StateIdle) and rl.check_collision_point_rec(rl.get_mouse_position(), ssid_rect) and clicked:
if not network.is_saved and network.security_type != SecurityType.OPEN:
self.state = StateNeedsAuth(network)
else:
elif not network.is_connected:
self.connect_to_network(network)
def _draw_status_icon(self, rect, network: NetworkInfo):
"""Draw the status icon based on network's connection state"""
icon_file = None
if network.is_connected:
icon_file = "icons/checkmark.png"
elif network.security_type == SecurityType.UNSUPPORTED:
icon_file = "icons/circled_slash.png"
elif network.security_type != SecurityType.OPEN:
icon_file = "icons/lock_closed.png"
if not icon_file:
return
texture = gui_app.texture(icon_file, ICON_SIZE, ICON_SIZE)
icon_rect = rl.Vector2(rect.x, rect.y + (ICON_SIZE - texture.height) / 2)
rl.draw_texture_v(texture, icon_rect, rl.WHITE)
def _draw_signal_strength_icon(self, rect: rl.Rectangle, network: NetworkInfo):
"""Draw the Wi-Fi signal strength icon based on network's signal strength"""
strength_level = max(0, min(3, round(network.strength / 33.0)))
rl.draw_texture_v(gui_app.texture(STRENGTH_ICONS[strength_level], ICON_SIZE, ICON_SIZE), rl.Vector2(rect.x, rect.y), rl.WHITE)
def connect_to_network(self, network: NetworkInfo, password=''):
self.state = StateConnecting(network)
if self.wifi_manager.is_saved(network.ssid) and not password:
if network.is_saved and not password:
self.wifi_manager.activate_connection(network.ssid)
else:
self.wifi_manager.connect_to_network(network.ssid, password)
def forget_network(self, network: NetworkInfo):
self.state = StateForgetting(network)
network.is_saved = False
self.wifi_manager.forget_connection(network.ssid)
def _on_need_auth(self):
def _on_network_updated(self, networks: list[NetworkInfo]):
self._networks = networks
def _on_need_auth(self, ssid):
match self.state:
case StateConnecting(network):
self.state = StateNeedsAuth(network)
case StateConnecting(ssid):
self.state = StateNeedsAuth(ssid)
case _:
# Find network by SSID
network = next((n for n in self.wifi_manager.networks if n.ssid == ssid), None)
if network:
self.state = StateNeedsAuth(network)
def _on_activated(self):
if isinstance(self.state, StateConnecting):
+81
View File
@@ -0,0 +1,81 @@
import pyray as rl
from openpilot.system.ui.lib.button import gui_button, ButtonStyle, TextAlignment
from openpilot.system.ui.lib.label import gui_label
from openpilot.system.ui.lib.scroll_panel import GuiScrollPanel
class MultiOptionDialog:
def __init__(self, title, options, current=""):
self._title = title
self._options = options
self._current = current if current in options else ""
self._selection = self._current
self._option_height = 80
self._padding = 20
self.scroll_panel = GuiScrollPanel()
@property
def selection(self):
return self._selection
def render(self, rect):
title_rect = rl.Rectangle(rect.x + self._padding, rect.y + self._padding, rect.width - 2 * self._padding, 70)
gui_label(title_rect, self._title, 70)
options_y_start = rect.y + 120
options_height = len(self._options) * (self._option_height + 10)
options_rect = rl.Rectangle(rect.x + self._padding, options_y_start, rect.width - 2 * self._padding, options_height)
view_rect = rl.Rectangle(
rect.x + self._padding, options_y_start, rect.width - 2 * self._padding, rect.height - 200 - 2 * self._padding
)
offset = self.scroll_panel.handle_scroll(view_rect, options_rect)
is_click_valid = self.scroll_panel.is_click_valid()
rl.begin_scissor_mode(int(view_rect.x), int(view_rect.y), int(view_rect.width), int(view_rect.height))
for i, option in enumerate(self._options):
y_pos = view_rect.y + i * (self._option_height + 10) + offset.y
item_rect = rl.Rectangle(view_rect.x, y_pos, view_rect.width, self._option_height)
if not rl.check_collision_recs(item_rect, view_rect):
continue
is_selected = option == self._selection
button_style = ButtonStyle.PRIMARY if is_selected else ButtonStyle.NORMAL
if gui_button(item_rect, option, button_style=button_style, text_alignment=TextAlignment.LEFT) and is_click_valid:
self._selection = option
rl.end_scissor_mode()
button_y = rect.y + rect.height - 80 - self._padding
button_width = (rect.width - 3 * self._padding) / 2
cancel_rect = rl.Rectangle(rect.x + self._padding, button_y, button_width, 80)
if gui_button(cancel_rect, "Cancel"):
return 0 # Canceled
select_rect = rl.Rectangle(rect.x + 2 * self._padding + button_width, button_y, button_width, 80)
has_new_selection = self._selection != "" and self._selection != self._current
if gui_button(select_rect, "Select", is_enabled=has_new_selection, button_style=ButtonStyle.PRIMARY):
return 1 # Selected
return -1 # Still active
if __name__ == "__main__":
from openpilot.system.ui.lib.application import gui_app
gui_app.init_window("Multi Option Dialog Example")
options = [f"Option {i}" for i in range(1, 11)]
dialog = MultiOptionDialog("Choose an option", options, options[0])
for _ in gui_app.render():
result = dialog.render(rl.Rectangle(100, 100, 1024, 800))
if result >= 0:
print(f"Selected: {dialog.selection}" if result > 0 else "Canceled")
break