Merge commaai/openpilot fdd1df79 into sync-20260721

This commit is contained in:
Jason Wen
2026-07-21 01:48:30 -04:00
247 changed files with 4628 additions and 3056 deletions
+16 -15
View File
@@ -679,18 +679,17 @@ def add_log_to_queue(log_path, log_id, is_sunnylink=False):
f"after compression: {compressed_size} bytes, " +
f"after encoding: {encoded_size} bytes")
jsonrpc = {
params: dict[str, str | bool] = {"logs": payload}
if is_sunnylink and is_compressed:
params["compressed"] = is_compressed
jsonrpc: dict = {
"method": "forwardLogs",
"params": {
"logs": payload
},
"params": params,
"jsonrpc": "2.0",
"id": log_id
}
if is_sunnylink and is_compressed:
jsonrpc["params"]["compressed"] = is_compressed
jsonrpc_str = json.dumps(jsonrpc)
size_in_bytes = len(jsonrpc_str.encode('utf-8'))
@@ -783,18 +782,17 @@ def stat_handler(end_event: threading.Event, stats_dir=None, is_sunnylink=False)
payload = base64.b64encode(compressed_data).decode()
is_compressed = True
jsonrpc = {
params: dict[str, str | bool] = {"stats": payload}
if is_sunnylink and is_compressed:
params["compressed"] = is_compressed
jsonrpc: dict = {
"method": "storeStats",
"params": {
"stats": payload
},
"params": params,
"jsonrpc": "2.0",
"id": stat_filenames[0]
}
if is_sunnylink and is_compressed:
jsonrpc["params"]["compressed"] = is_compressed
send_queue_push(json.dumps(jsonrpc), SEND_PRIORITY_LOW)
os.remove(stat_path)
last_scan = curr_scan
@@ -806,7 +804,10 @@ def stat_handler(end_event: threading.Event, stats_dir=None, is_sunnylink=False)
def ws_proxy_recv(ws: WebSocket, local_sock: socket.socket, ssock: socket.socket, end_event: threading.Event, global_end_event: threading.Event) -> None:
while not (end_event.is_set() or global_end_event.is_set()):
try:
r = select.select((ws.sock,), (), (), 30)
sock = ws.sock
if sock is None:
return
r = select.select((sock,), (), (), 30)
if r[0]:
data = ws.recv()
if isinstance(data, str):
+3
View File
@@ -83,6 +83,9 @@ def register(show_spinner=False) -> str | None:
dongleauth = json.loads(resp.text)
dongle_id = dongleauth["dongle_id"]
break
except NotImplementedError:
# dependency issues with PyJWT will hang the registration test in backoff loop otherwise
raise
except Exception:
cloudlog.exception("failed to authenticate")
backoff = min(backoff + 1, 15)
@@ -60,7 +60,7 @@ class TestAthenadMethods:
@classmethod
def setup_class(cls):
cls.SOCKET_PORT = 45454
athenad.Api = MockApi
athenad.Api = MockApi # ty: ignore[invalid-assignment] # test double
athenad.LOCAL_PORT_WHITELIST = {cls.SOCKET_PORT}
def setup_method(self):
@@ -351,6 +351,7 @@ class TestAthenadMethods:
assert items[0] == asdict(item)
assert not items[0]['current']
assert item.id is not None
athenad.cancelled_uploads.add(item.id)
items = dispatcher["listUploadQueue"]()
assert len(items) == 0
@@ -363,6 +364,7 @@ class TestAthenadMethods:
athenad.upload_queue.put_nowait(item2)
# Ensure canceled items are not persisted
assert item2.id is not None
athenad.cancelled_uploads.add(item2.id)
# serialize item
@@ -437,7 +439,7 @@ class TestAthenadMethods:
thread.join()
def test_get_logs_to_send_sorted(self):
fl = list()
fl = []
for i in range(10):
file = f'swaglog.{i:010}'
self._create_file(file, Paths.swaglog_root())
@@ -35,7 +35,7 @@ class PowerMonitoring:
self.car_battery_capacity_uWh = max((CAR_BATTERY_CAPACITY_uWh / 10), car_battery_capacity_uWh)
# Calculation tick
def calculate(self, voltage: int | None, ignition: bool):
def calculate(self, voltage: float | None, ignition: bool):
try:
now = time.monotonic()
@@ -35,7 +35,7 @@ class TestPowerMonitoring:
def test_panda_state_present(self):
pm = PowerMonitoring()
for _ in range(10):
pm.calculate(None, None)
pm.calculate(None, False)
assert pm.get_power_used() == 0
assert pm.get_car_battery_capacity() == (CAR_BATTERY_CAPACITY_uWh / 10)
@@ -93,7 +93,7 @@ void JpegEncoder::compressToJpeg(uint8_t *y_plane, uint8_t *u_plane, uint8_t *v_
frame->data[2] = v_plane;
// Required for MJPEG qscale to take effect (global_quality alone is not enough).
frame->quality = FF_QP2LAMBDA * MJPEG_QSCALE;
frame->pts = 0;
frame->pts = AV_NOPTS_VALUE;
int err = avcodec_send_frame(codec_ctx, frame);
if (err < 0) {
@@ -63,10 +63,10 @@ class UploaderTestCase:
seg_dir: str
def set_ignore(self):
uploader.Api = MockApiIgnore
uploader.Api = MockApiIgnore # ty: ignore[invalid-assignment] # test double
def setup_method(self):
uploader.Api = MockApi
uploader.Api = MockApi # ty: ignore[invalid-assignment] # test double
uploader.fake_upload = True
uploader.force_wifi = True
uploader.allow_sleep = False
@@ -19,7 +19,7 @@ class TestDeleter(UploaderTestCase):
self.f_type = "fcamera.hevc"
super().setup_method()
self.fake_stats = Stats(f_bavail=0, f_blocks=10, f_frsize=4096)
deleter.os.statvfs = self.fake_statvfs
deleter.os.statvfs = self.fake_statvfs # ty: ignore[invalid-assignment] # test double
def start_thread(self):
self.end_event = threading.Event()
@@ -5,6 +5,7 @@ import random
import string
import subprocess
import time
from collections.abc import Collection
from collections import defaultdict
from pathlib import Path
import pytest
@@ -74,8 +75,8 @@ class TestLoggerd:
end_type = SentinelType.endOfRoute if route else SentinelType.endOfSegment
assert msgs[-1].sentinel.type == end_type
def _publish_random_messages(self, services: list[str]) -> dict[str, list]:
pm = messaging.PubMaster(services)
def _publish_random_messages(self, services: Collection[str]) -> dict[str, list]:
pm = messaging.PubMaster(list(services))
managed_processes["loggerd"].start()
for s in services:
@@ -277,7 +278,9 @@ class TestLoggerd:
assert recv_cnt == 0, f"got {recv_cnt} {s} msgs in qlog"
else:
# check logged message count matches decimation
expected_cnt = (len(msgs) - 1) // SERVICE_LIST[s].decimation + 1
decimation = SERVICE_LIST[s].decimation
assert decimation is not None
expected_cnt = (len(msgs) - 1) // decimation + 1
assert recv_cnt == expected_cnt, f"expected {expected_cnt} msgs for {s}, got {recv_cnt}"
def test_rlog(self):
@@ -18,8 +18,8 @@ class FakeLogHandler(logging.Handler):
self.reset()
def reset(self):
self.upload_order = list()
self.upload_ignored = list()
self.upload_order = []
self.upload_ignored = []
def emit(self, record):
try:
+1 -3
View File
@@ -46,9 +46,7 @@ class FakeResponse:
def get_directory_sort(d: str) -> list[str]:
# ensure old format is sorted sooner
o = ["0", ] if d.startswith("2024-") else ["1", ]
return o + [s.rjust(10, '0') for s in d.rsplit('--', 1)]
return [s.rjust(10, '0') for s in d.rsplit('--', 1)]
def listdir_by_creation(d: str) -> list[str]:
if not os.path.isdir(d):
+50 -3
View File
@@ -1,6 +1,53 @@
import ctypes
import errno
import os
import sys
import xattr
if sys.platform == "darwin":
_libc = ctypes.CDLL(None, use_errno=True)
_libc.getxattr.argtypes = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_void_p, ctypes.c_size_t, ctypes.c_uint32, ctypes.c_int]
_libc.getxattr.restype = ctypes.c_ssize_t
_libc.setxattr.argtypes = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_void_p, ctypes.c_size_t, ctypes.c_uint32, ctypes.c_int]
_libc.setxattr.restype = ctypes.c_int
def _raise_os_error(path: str) -> None:
error = ctypes.get_errno()
raise OSError(error, os.strerror(error), path)
def _getxattr(path: str, attr_name: str) -> bytes:
if sys.platform != "darwin":
return os.getxattr(path, attr_name)
encoded_path = os.fsencode(path)
encoded_attr_name = os.fsencode(attr_name)
while True:
size = _libc.getxattr(encoded_path, encoded_attr_name, None, 0, 0, 0)
if size == -1:
_raise_os_error(path)
if size == 0:
return b""
value = ctypes.create_string_buffer(size)
result = _libc.getxattr(encoded_path, encoded_attr_name, value, size, 0, 0)
if result != -1:
return value.raw[:result]
if ctypes.get_errno() != errno.ERANGE:
_raise_os_error(path)
def _setxattr(path: str, attr_name: str, attr_value: bytes) -> None:
if sys.platform != "darwin":
os.setxattr(path, attr_name, attr_value)
return
encoded_path = os.fsencode(path)
encoded_attr_name = os.fsencode(attr_name)
value = ctypes.create_string_buffer(attr_value)
if _libc.setxattr(encoded_path, encoded_attr_name, value, len(attr_value), 0, 0) == -1:
_raise_os_error(path)
_cached_attributes: dict[tuple, bytes | None] = {}
@@ -8,7 +55,7 @@ def getxattr(path: str, attr_name: str) -> bytes | None:
key = (path, attr_name)
if key not in _cached_attributes:
try:
response = xattr.getxattr(path, attr_name)
response = _getxattr(path, attr_name)
except OSError as e:
# ENODATA (Linux) or ENOATTR (macOS) means attribute hasn't been set
if e.errno == errno.ENODATA or (hasattr(errno, 'ENOATTR') and e.errno == errno.ENOATTR):
@@ -20,4 +67,4 @@ def getxattr(path: str, attr_name: str) -> bytes | None:
def setxattr(path: str, attr_name: str, attr_value: bytes) -> None:
_cached_attributes.pop((path, attr_name), None)
xattr.setxattr(path, attr_name, attr_value)
_setxattr(path, attr_name, attr_value)
-5
View File
@@ -104,11 +104,6 @@ def manager_init() -> None:
dirty=build_metadata.openpilot.is_dirty,
device=HARDWARE.get_device_type())
# preimport all processes
for p in managed_processes.values():
p.prepare()
def manager_cleanup() -> None:
# send signals to kill all procs
for p in managed_processes.values():
-15
View File
@@ -70,10 +70,6 @@ class ManagerProcess(ABC):
shutting_down = False
restart_if_crash = False
@abstractmethod
def prepare(self) -> None:
pass
@abstractmethod
def start(self) -> None:
pass
@@ -150,9 +146,6 @@ class NativeProcess(ManagerProcess):
self.sigkill = sigkill
self.launcher = nativelauncher
def prepare(self) -> None:
pass
def start(self) -> None:
# In case we only tried a non blocking stop we need to stop it before restarting
if self.shutting_down:
@@ -178,11 +171,6 @@ class PythonProcess(ManagerProcess):
self.launcher = launcher
self.restart_if_crash = restart_if_crash
def prepare(self) -> None:
if self.enabled:
cloudlog.info(f"preimporting {self.module}")
importlib.import_module(self.module)
def start(self) -> None:
# In case we only tried a non blocking stop we need to stop it before restarting
if self.shutting_down:
@@ -211,9 +199,6 @@ class DaemonProcess(ManagerProcess):
def should_run(started, params, CP):
return True
def prepare(self) -> None:
pass
def start(self) -> None:
if self.params is None:
self.params = Params()
+35 -33
View File
@@ -1,6 +1,6 @@
import os
import sys
from dataclasses import dataclass, fields
from dataclasses import dataclass
from subprocess import check_output, CalledProcessError
from time import sleep
from typing import NoReturn
@@ -17,26 +17,27 @@ class GnssClockNmeaPort:
# 0x10 = bias_uncertainty_ns valid
# 0x20 = drift_nsps valid
# 0x40 = drift_uncertainty_nsps valid
flags: int
leap_seconds: int
time_ns: int
time_uncertainty_ns: int # 1-sigma
full_bias_ns: int
bias_ns: float
bias_uncertainty_ns: float # 1-sigma
drift_nsps: float
drift_uncertainty_nsps: float # 1-sigma
flags: int | None
leap_seconds: int | None
time_ns: int | None
time_uncertainty_ns: int | None # 1-sigma
full_bias_ns: int | None
bias_ns: float | None
bias_uncertainty_ns: float | None # 1-sigma
drift_nsps: float | None
drift_uncertainty_nsps: float | None # 1-sigma
def __post_init__(self):
for field in fields(self):
val = getattr(self, field.name)
setattr(self, field.name, field.type(val) if val else None)
@classmethod
def from_fields(cls, values: list[str]) -> 'GnssClockNmeaPort':
ints = [int(value) if value else None for value in values[:5]]
floats = [float(value) if value else None for value in values[5:9]]
return cls(*ints, *floats)
@dataclass
class GnssMeasNmeaPort:
messageCount: int
messageNum: int
svCount: int
messageCount: int | None
messageNum: int | None
svCount: int | None
# constellation enum:
# 1 = GPS
# 2 = SBAS
@@ -44,10 +45,10 @@ class GnssMeasNmeaPort:
# 4 = QZSS
# 5 = BEIDOU
# 6 = GALILEO
constellation: int
svId: int
flags: int # always zero
time_offset_ns: int
constellation: int | None
svId: int | None
flags: int | None # always zero
time_offset_ns: int | None
# state bit mask:
# 0x0001 = CODE LOCK
# 0x0002 = BIT SYNC
@@ -63,17 +64,18 @@ class GnssMeasNmeaPort:
# 0x0800 = GALILEO E1C 2ND CODE LOCK
# 0x1000 = GALILEO E1B PAGE SYNC
# 0x2000 = GALILEO E1B PAGE SYNC
state: int
time_of_week_ns: int
time_of_week_uncertainty_ns: int # 1-sigma
carrier_to_noise_ratio: float
pseudorange_rate: float
pseudorange_rate_uncertainty: float # 1-sigma
state: int | None
time_of_week_ns: int | None
time_of_week_uncertainty_ns: int | None # 1-sigma
carrier_to_noise_ratio: float | None
pseudorange_rate: float | None
pseudorange_rate_uncertainty: float | None # 1-sigma
def __post_init__(self):
for field in fields(self):
val = getattr(self, field.name)
setattr(self, field.name, field.type(val) if val else None)
@classmethod
def from_fields(cls, values: list[str]) -> 'GnssMeasNmeaPort':
ints = [int(value) if value else None for value in values[:10]]
floats = [float(value) if value else None for value in values[10:13]]
return cls(*ints, *floats)
def nmea_checksum_ok(s):
checksum = 0
@@ -107,11 +109,11 @@ def process_nmea_port_messages(device:str="/dev/ttyUSB1") -> NoReturn:
match fields[0]:
case "$GNCLK":
# fields at end are reserved (not used)
gnss_clock = GnssClockNmeaPort(*fields[1:10])
gnss_clock = GnssClockNmeaPort.from_fields(fields[1:10])
print(gnss_clock)
case "$GNMEAS":
# fields at end are reserved (not used)
gnss_meas = GnssMeasNmeaPort(*fields[1:14])
gnss_meas = GnssMeasNmeaPort.from_fields(fields[1:14])
print(gnss_meas)
except Exception as e:
print(e)
@@ -20,7 +20,7 @@ SENSOR_CONFIGS = (
)
SENSOR_CONFIGS_BY_MEASUREMENT = {config.measurement: config for config in SENSOR_CONFIGS}
def get_irq_count(irq: int):
def get_irq_count(irq: str):
with open(f"/sys/kernel/irq/{irq}/per_cpu_count") as f:
per_cpu = map(int, f.read().split(","))
return sum(per_cpu)
+1 -1
View File
@@ -184,7 +184,7 @@ class BinaryStruct:
setattr(obj, name, value)
return obj
cls._read = _read
cls._read = _read # ty: ignore[invalid-assignment] # installed dynamically for each subclass
@classmethod
def from_bytes(cls: type[T], data: bytes) -> T:
+1
View File
@@ -831,6 +831,7 @@ class GuiApplication(GuiApplicationExt):
import pstats
self._render_profiler.disable()
assert self._render_profile_start_time is not None
elapsed_ms = (time.monotonic() - self._render_profile_start_time) * 1e3
avg_frame_time = elapsed_ms / self._frame if self._frame > 0 else 0
+2 -2
View File
@@ -152,7 +152,7 @@ class ShaderState:
self.initialized = False
def _configure_shader_color(state: ShaderState, color: Optional[rl.Color],
def _configure_shader_color(state: ShaderState, color: Optional[rl.Color], # noqa: UP045 # rl.Color is a function, so `rl.Color | None` fails
gradient: Gradient | None, origin_rect: rl.Rectangle):
assert (color is not None) != (gradient is not None), "Either color or gradient must be provided"
@@ -204,7 +204,7 @@ def triangulate(pts: np.ndarray) -> list[tuple[float, float]]:
def draw_polygon(origin_rect: rl.Rectangle, points: np.ndarray,
color: Optional[rl.Color] = None, gradient: Gradient | None = None):
color: Optional[rl.Color] = None, gradient: Gradient | None = None): # noqa: UP045 # rl.Color is a function, so `rl.Color | None` fails
"""
Draw a ribbon polygon (two chains) with a triangle strip and gradient.
+2 -1
View File
@@ -1,8 +1,9 @@
import pyray as rl
from collections.abc import Sequence
class GuiStyleContext:
def __init__(self, styles: list[tuple[int, int, int]]):
def __init__(self, styles: Sequence[tuple[int, int, int]]):
"""styles is a list of tuples (control, prop, new_value)"""
self.styles = styles
self.prev_styles: list[tuple[int, int, int]] = []
+1 -1
View File
@@ -257,7 +257,7 @@ class WifiManager:
def add_callbacks(self, need_auth: Callable[[str], None] | None = None,
activated: Callable[[], None] | None = None,
forgotten: Callable[[str], None] | None = None,
forgotten: Callable[[str | None], None] | None = None,
networks_updated: Callable[[list[Network]], None] | None = None,
disconnected: Callable[[], None] | None = None):
if need_auth is not None:
@@ -4,7 +4,7 @@ Copyright (c) 2021-, Haibin Wen, sunnypilot, and a number of other contributors.
This file is part of sunnypilot and is licensed under the MIT License.
See the LICENSE.md file in the root directory for more details.
"""
from collections.abc import Callable
from collections.abc import Callable, Sequence
import pyray as rl
from openpilot.common.params import Params
@@ -126,7 +126,7 @@ class DualButtonActionSP(DualButtonAction):
class MultipleButtonActionSP(MultipleButtonAction):
def __init__(self, buttons: list[str | Callable[[], str]], button_width: int, selected_index: int = 0, callback: Callable | None = None,
def __init__(self, buttons: Sequence[str | Callable[[], str]], button_width: int, selected_index: int = 0, callback: Callable | None = None,
param: str | None = None):
MultipleButtonAction.__init__(self, buttons, button_width, selected_index, callback)
self.param_key = param
@@ -366,7 +366,7 @@ def toggle_item_sp(title: str | Callable[[], str], description: str | Callable[[
return ListItemSP(title=title, description=description, action_item=action, icon=icon)
def multiple_button_item_sp(title: str | Callable[[], str], description: str | Callable[[], str], buttons: list[str | Callable[[], str]],
def multiple_button_item_sp(title: str | Callable[[], str], description: str | Callable[[], str], buttons: Sequence[str | Callable[[], str]],
selected_index: int = 0, button_width: int = style.BUTTON_ACTION_WIDTH, callback: Callable | None = None,
icon: str = "", param: str | None = None, inline: bool = False) -> ListItemSP:
action = MultipleButtonActionSP(buttons, button_width, selected_index, callback=callback, param=param)
@@ -48,9 +48,9 @@ class TreeItemWidget(Button):
self.border_radius = 10
self.is_expanded = is_expanded
def _render(self, rect):
def _render(self, _):
indent = 60 * self.indent_level
self._rect = rl.Rectangle(rect.x + indent, rect.y, rect.width - indent, rect.height)
self._rect = rl.Rectangle(_.x + indent, _.y, _.width - indent, _.height)
if self.is_pressed:
color = BUTTON_PRESSED_BACKGROUND_COLORS[self._button_style]
elif self.selected and self.ref != "search_bar":
+5 -1
View File
@@ -37,7 +37,11 @@ class Reset(Widget):
self._reset_state = ResetState.NONE
self._cancel_button = Button("Cancel", gui_app.request_close)
self._confirm_button = Button("Confirm", self._confirm, button_style=ButtonStyle.PRIMARY)
self._reboot_button = Button("Reboot", lambda: subprocess.run("sudo reboot", shell=True))
self._reboot_button = Button("Reboot", self._reboot)
@staticmethod
def _reboot() -> None:
subprocess.run("sudo reboot", shell=True)
def _do_erase(self):
if PC:
+19 -10
View File
@@ -3,16 +3,25 @@ from __future__ import annotations
import abc
import pyray as rl
from enum import IntEnum
from typing import TypeVar
from typing import Protocol, TypeVar
from collections.abc import Callable
from openpilot.system.ui.lib.application import gui_app, MousePos, MAX_TOUCH_SLOTS, MouseEvent
try:
from openpilot.selfdrive.ui.ui_state import device
except ImportError:
class Device:
awake = True
device = Device()
class DeviceLike(Protocol):
awake: bool
def _get_device() -> DeviceLike:
try:
from openpilot.selfdrive.ui.ui_state import device
return device
except ImportError:
class Device:
awake = True
return Device()
device = _get_device()
W = TypeVar('W', bound='Widget')
@@ -185,16 +194,16 @@ class Widget(abc.ABC):
"""Optionally update the widget's non-layout state. This is called before rendering."""
@abc.abstractmethod
def _render(self, rect: rl.Rectangle) -> bool | int | None:
def _render(self, rect: rl.Rectangle, /) -> bool | int | None:
"""Render the widget within the given rectangle."""
def _update_layout_rects(self) -> None:
"""Optionally update any layout rects on Widget rect change."""
def _handle_mouse_press(self, mouse_pos: MousePos) -> None:
def _handle_mouse_press(self, mouse_pos: MousePos, /) -> None:
"""Optionally handle mouse press events."""
def _handle_mouse_release(self, mouse_pos: MousePos) -> None:
def _handle_mouse_release(self, mouse_pos: MousePos, /) -> None:
"""Optionally handle mouse release events."""
if self._click_delay is not None:
self._click_release_time = rl.get_time() + self._click_delay
+6 -4
View File
@@ -1,7 +1,6 @@
import math
from enum import IntEnum
from collections.abc import Callable
from itertools import zip_longest
from typing import Union
import pyray as rl
@@ -188,6 +187,9 @@ class Label(Widget):
if self._text_alignment_vertical == rl.GuiTextAlignmentVertical.TEXT_ALIGN_MIDDLE:
total_text_height = sum(ts.y for ts in self._text_size) or self._font_size * FONT_SCALE
text_pos = rl.Vector2(self._rect.x, (self._rect.y + (self._rect.height - total_text_height) // 2))
elif self._text_alignment_vertical == rl.GuiTextAlignmentVertical.TEXT_ALIGN_BOTTOM:
total_text_height = sum(ts.y for ts in self._text_size) or self._font_size * FONT_SCALE
text_pos = rl.Vector2(self._rect.x, self._rect.y + self._rect.height - total_text_height)
else:
text_pos = rl.Vector2(self._rect.x, self._rect.y)
@@ -196,18 +198,18 @@ class Label(Widget):
if len(self._text_wrapped) > 0:
if self._text_alignment == rl.GuiTextAlignment.TEXT_ALIGN_LEFT:
icon_x = self._rect.x + self._text_padding
text_pos.x = self._icon.width + ICON_PADDING
text_pos.x = self._rect.x + self._icon.width + ICON_PADDING
elif self._text_alignment == rl.GuiTextAlignment.TEXT_ALIGN_CENTER:
total_width = self._icon.width + ICON_PADDING + text_size.x
icon_x = self._rect.x + (self._rect.width - total_width) / 2
text_pos.x = self._icon.width + ICON_PADDING
text_pos.x = self._rect.x + self._icon.width + ICON_PADDING
else:
icon_x = (self._rect.x + self._rect.width - text_size.x - self._text_padding) - ICON_PADDING - self._icon.width
else:
icon_x = self._rect.x + (self._rect.width - self._icon.width) / 2
rl.draw_texture_v(self._icon, rl.Vector2(icon_x, icon_y), rl.WHITE)
for text, text_size, emojis in zip_longest(self._text_wrapped, self._text_size, self._emojis, fillvalue=[]):
for text, text_size, emojis in zip(self._text_wrapped, self._text_size, self._emojis, strict=True):
line_pos = rl.Vector2(text_pos.x, text_pos.y)
if self._text_alignment == rl.GuiTextAlignment.TEXT_ALIGN_LEFT:
line_pos.x += self._text_padding
+2 -2
View File
@@ -1,6 +1,6 @@
import os
import pyray as rl
from collections.abc import Callable
from collections.abc import Callable, Sequence
from abc import ABC
from openpilot.system.ui.lib.application import gui_app, FontWeight, MousePos
from openpilot.system.ui.lib.multilang import tr
@@ -207,7 +207,7 @@ class DualButtonAction(ItemAction):
class MultipleButtonAction(ItemAction):
def __init__(self, buttons: list[str | Callable[[], str]], button_width: int, selected_index: int = 0, callback: Callable | None = None):
def __init__(self, buttons: Sequence[str | Callable[[], str]], button_width: int, selected_index: int = 0, callback: Callable | None = None):
super().__init__(width=len(buttons) * button_width + (len(buttons) - 1) * RIGHT_ITEM_PADDING, enabled=True)
self.buttons = buttons
self.button_width = button_width
+1 -1
View File
@@ -78,7 +78,7 @@ class NavWidget(Widget, abc.ABC):
# the top of a vertical scroll panel to prevent erroneous swipes
return True
def set_back_callback(self, callback: Callable[[], None]) -> None:
def set_back_callback(self, callback: Callable[[], None] | None) -> None:
self._back_callback = callback
def set_shown_callback(self, callback: Callable[[], None] | None) -> None:
+11 -5
View File
@@ -1,6 +1,6 @@
from enum import IntEnum
from functools import partial
from typing import cast
from typing import Any, cast
import pyray as rl
from openpilot.system.ui.lib.application import gui_app
@@ -27,9 +27,9 @@ try:
from openpilot.selfdrive.ui.ui_state import ui_state
from openpilot.selfdrive.ui.lib.prime_state import PrimeType
except Exception:
Params = None
ui_state = None
PrimeType = None
Params: Any = None
ui_state: Any = None
PrimeType: Any = None
NM_DEVICE_STATE_NEED_AUTH = 60
MIN_PASSWORD_LENGTH = 8
@@ -111,10 +111,16 @@ class NetworkUI(Widget):
class AdvancedNetworkSettings(Widget):
def __init__(self, wifi_manager: WifiManager):
# AdvancedNetworkSettings needs the full openpilot environment, standalone apps just use WifiManagerUI
from openpilot.common.params import Params
from openpilot.selfdrive.ui.ui_state import ui_state
from openpilot.selfdrive.ui.lib.prime_state import PrimeType
super().__init__()
self._wifi_manager = wifi_manager
self._wifi_manager.add_callbacks(networks_updated=self._on_network_updated)
self._params = Params()
self._prime_state = ui_state.prime_state
self._cell_prime_types = (PrimeType.NONE, PrimeType.LITE)
self._keyboard = Keyboard(max_text_size=MAX_PASSWORD_LENGTH, min_text_size=MIN_PASSWORD_LENGTH, show_password_toggle=True)
@@ -259,7 +265,7 @@ class AdvancedNetworkSettings(Widget):
self._wifi_manager.process_callbacks()
# If not using prime SIM, show GSM settings and enable IPv4 forwarding
show_cell_settings = ui_state.prime_state.get_type() in (PrimeType.NONE, PrimeType.LITE)
show_cell_settings = self._prime_state.get_type() in self._cell_prime_types
self._wifi_manager.set_ipv4_forward(show_cell_settings)
self._roaming_btn.set_visible(show_cell_settings)
self._apn_btn.set_visible(show_cell_settings)
+6 -6
View File
@@ -1,6 +1,6 @@
import pyray as rl
import numpy as np
from collections.abc import Callable
from collections.abc import Callable, Sequence
from openpilot.common.filter_simple import FirstOrderFilter, BounceFilter
from openpilot.common.swaglog import cloudlog
@@ -40,7 +40,7 @@ class ScrollIndicator(Widget):
self._content_size = content_size
self._viewport = viewport
def _render(self, _):
def _render(self, _, /):
# scale indicator width based on content size
indicator_w = float(np.interp(self._content_size, [1000, 3000], [300, 100]))
@@ -69,7 +69,7 @@ class ScrollIndicator(Widget):
class _Scroller(Widget):
"""Should use wrapper below to reduce boilerplate"""
def __init__(self, items: list[Widget], horizontal: bool = True, snap_items: bool = False, spacing: int = ITEM_SPACING,
def __init__(self, items: Sequence[Widget], horizontal: bool = True, snap_items: bool = False, spacing: int = ITEM_SPACING,
pad: int = ITEM_SPACING, scroll_indicator: bool = True, edge_shadows: bool = True):
super().__init__()
self._items: list[Widget] = []
@@ -150,7 +150,7 @@ class _Scroller(Widget):
and not self.moving_items and (original_touch_valid_callback() if
original_touch_valid_callback else True))
def add_widgets(self, items: list[Widget]) -> None:
def add_widgets(self, items: Sequence[Widget]) -> None:
for item in items:
self.add_widget(item)
@@ -332,7 +332,7 @@ class _Scroller(Widget):
else:
item.render()
def _render(self, _):
def _render(self, _, /):
rl.begin_scissor_mode(int(self._rect.x), int(self._rect.y),
int(self._rect.width), int(self._rect.height))
@@ -397,7 +397,7 @@ class Scroller(Widget):
# pass down enabled to child widget for nav stack
self._scroller.set_enabled(lambda: self.enabled)
def _render(self, _):
def _render(self, _, /):
self._scroller.render(self._rect)
+2 -1
View File
@@ -1,4 +1,5 @@
import pyray as rl
from collections.abc import Sequence
from openpilot.system.ui.lib.scroll_panel import GuiScrollPanel
from openpilot.system.ui.widgets import Widget
@@ -23,7 +24,7 @@ class LineSeparator(Widget):
class Scroller(Widget):
def __init__(self, items: list[Widget], spacing: int = ITEM_SPACING, line_separator: bool = False, pad_end: bool = True):
def __init__(self, items: Sequence[Widget], spacing: int = ITEM_SPACING, line_separator: bool = False, pad_end: bool = True):
super().__init__()
self._items: list[Widget] = []
self._spacing = spacing
+1 -1
View File
@@ -342,7 +342,7 @@ class Updater:
setup_git_options(OVERLAY_MERGED)
output = run(["git", "ls-remote", "--heads"], OVERLAY_MERGED)
self.branches = defaultdict(lambda: None)
self.branches.clear()
for line in output.split('\n'):
ls_remotes_re = r'(?P<commit_sha>\b[0-9a-f]{5,40}\b)(\s+)(refs\/heads\/)(?P<branch_name>.*$)'
x = re.fullmatch(ls_remotes_re, line.strip())
+14 -13
View File
@@ -1,10 +1,9 @@
import asyncio
from dataclasses import dataclass
import struct
import time
import av
from teleoprtc.tracks import TiciVideoStreamTrack
from aiortc import MediaStreamError
from openpilot.cereal import messaging
from openpilot.common.realtime import DT_MDL
@@ -22,6 +21,15 @@ TIMING_SEI_UUID = bytes([
_SEI_PREFIX = b'\x00\x00\x00\x01\x06\x05\x30' + TIMING_SEI_UUID
@dataclass(frozen=True)
class EncodedVideoFrame:
data: bytes
pts: int
def __bytes__(self) -> bytes:
return self.data
class LiveStreamVideoStreamTrack(TiciVideoStreamTrack):
camera_to_sock_mapping = {
"driver": "livestreamDriverEncodeData",
@@ -55,6 +63,9 @@ class LiveStreamVideoStreamTrack(TiciVideoStreamTrack):
if not enabled:
self._seen_keyframe = False
def request_keyframe(self) -> None:
self.params.put("LivestreamRequestKeyframe", True, block=False)
def _build_frame_data(self, msg) -> bytes:
encode_data = getattr(msg, msg.which())
if not self.timing_sei_enabled:
@@ -71,9 +82,6 @@ class LiveStreamVideoStreamTrack(TiciVideoStreamTrack):
async def recv(self):
while True:
if self.readyState != "live":
raise MediaStreamError
# while video is disabled, pause here without returning
if not self.video_enabled:
await asyncio.sleep(0.005)
@@ -87,14 +95,7 @@ class LiveStreamVideoStreamTrack(TiciVideoStreamTrack):
break
await asyncio.sleep(0.005)
packet = av.Packet(self._build_frame_data(msg))
packet.time_base = self._time_base
self._pts = ((time.monotonic_ns() - self._t0_ns) * self._clock_rate) // 1_000_000_000
packet.pts = self._pts
self.log_debug("track sending frame %d", self._pts)
return packet
def codec_preference(self) -> str | None:
return "H264"
return EncodedVideoFrame(self._build_frame_data(msg), self._pts)
@@ -1,15 +1,10 @@
import asyncio
import json
import time
# for aiortc and its dependencies
import warnings
warnings.filterwarnings("ignore", category=DeprecationWarning)
warnings.filterwarnings("ignore", category=RuntimeWarning) # TODO: remove this when google-crc32c publish a python3.12 wheel
from aiortc import RTCDataChannel
from aiortc.mediastreams import VIDEO_CLOCK_RATE, VIDEO_TIME_BASE
import capnp
from openpilot.cereal import messaging, log
from teleoprtc.tracks import VIDEO_CLOCK_RATE
from openpilot.system.webrtc.webrtcd import CerealOutgoingMessageProxy, CerealIncomingMessageProxy
from openpilot.system.webrtc.device.video import LiveStreamVideoStreamTrack
@@ -31,7 +26,8 @@ class TestStreamSession:
expected_dict = {"type": "customReservedRawData0", "logMonoTime": 123, "valid": True, "data": "test"}
expected_json = json.dumps(expected_dict).encode()
channel = mocker.Mock(spec=RTCDataChannel)
channel = mocker.Mock()
channel.is_open.return_value = True
proxy = CerealOutgoingMessageProxy(["customReservedRawData0"])
def mocked_update(t):
proxy.sm.update_msgs(0, [test_msg])
@@ -59,9 +55,11 @@ class TestStreamSession:
mocked_pubmaster.send.assert_called_once()
mt, md = mocked_pubmaster.send.call_args.args
assert mt == msg["type"]
msg_type = msg["type"]
assert isinstance(msg_type, str)
assert mt == msg_type
assert isinstance(md, capnp._DynamicStructBuilder)
assert hasattr(md, msg["type"])
assert hasattr(md, msg_type)
mocked_pubmaster.reset_mock()
@@ -73,14 +71,11 @@ class TestStreamSession:
track = LiveStreamVideoStreamTrack("driver")
assert track.id.startswith("driver")
assert track.codec_preference() == "H264"
for i in range(5):
packet = self.loop.run_until_complete(track.recv())
assert packet.time_base == VIDEO_TIME_BASE
if i == 0:
start_ns = time.monotonic_ns()
start_pts = packet.pts
assert abs(i + packet.pts - (start_pts + (((time.monotonic_ns() - start_ns) * VIDEO_CLOCK_RATE) // 1_000_000_000))) < 450 #5ms
assert packet.size == 0
assert bytes(packet) == b""
+48 -68
View File
@@ -1,9 +1,11 @@
#!/usr/bin/env python3
from abc import abstractmethod
from collections.abc import Callable
import os
import socket
import time
import capnp
import argparse
import asyncio
import contextlib
@@ -14,17 +16,7 @@ import signal
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse, parse_qs
from typing import Any, TYPE_CHECKING
# aiortc and its dependencies have lots of internal warnings :(
import warnings
warnings.filterwarnings("ignore", category=DeprecationWarning)
warnings.filterwarnings("ignore", category=RuntimeWarning) # TODO: remove this when google-crc32c publish a python3.12 wheel
import capnp
if TYPE_CHECKING:
from aiortc.rtcdatachannel import RTCDataChannel
import aioice.ice
from typing import Any
from openpilot.system.webrtc.helpers import StreamRequestBody
from openpilot.system.webrtc.schema import generate_field
@@ -44,20 +36,8 @@ def _default_route_ip() -> str | None:
finally:
s.close()
# aioice patch: gather ICE candidates only on the default-route interface
_get_host_addresses = aioice.ice.get_host_addresses
def _primary_host_addresses(use_ipv4: bool, use_ipv6: bool) -> list[str]:
addresses = _get_host_addresses(use_ipv4, use_ipv6)
primary = _default_route_ip()
if primary not in addresses:
return addresses
return [primary, ]
aioice.ice.get_host_addresses = _primary_host_addresses
class AsyncTaskRunner:
def __init__(self):
self.is_running = False
self.task = None
self.logger = logging.getLogger("webrtcd")
@@ -86,10 +66,10 @@ class CerealOutgoingMessageProxy(AsyncTaskRunner):
super().__init__()
self.services = list(services)
self.sm = messaging.SubMaster(self.services)
self.channels: list[RTCDataChannel] = []
self.channels = []
self._enabled = enabled
def add_channel(self, channel: 'RTCDataChannel'):
def add_channel(self, channel):
self.channels.append(channel)
def enable(self, enable: bool):
@@ -118,20 +98,17 @@ class CerealOutgoingMessageProxy(AsyncTaskRunner):
outgoing_msg = {"type": service, "logMonoTime": mono_time, "valid": valid, "data": msg_dict}
encoded_msg = json.dumps(outgoing_msg).encode()
for channel in self.channels:
if not channel.is_open():
continue
channel.send(encoded_msg)
async def run(self):
from aiortc.exceptions import InvalidStateError
while True:
if not self._enabled:
await asyncio.sleep(0.01)
continue
try:
self.update()
except InvalidStateError:
self.logger.warning("Cereal outgoing proxy invalid state (connection closed)")
break
except Exception:
self.logger.exception("Cereal outgoing proxy failure")
await asyncio.sleep(0.01)
@@ -172,17 +149,17 @@ class LivestreamBitrateController(AsyncTaskRunner):
high_level = 0.1 # drop immediately
med_level = 0.05 # drop after # of samples
low_level = 0 # raise after # of samples
down_samples = 5 # 1s
down_samples = 5
param_name = "LivestreamEncoderBitrate"
def __init__(self, peer_connection: Any, params: Params, enabled: bool = True):
def __init__(self, get_stats: Callable[[], dict[str, Any]], params: Params, enabled: bool = True):
super().__init__()
self.pc = peer_connection
self.get_stats = get_stats
self.params = params
self.level = 2
self._publish(self.bitrates[self.level])
self.prev_lost, self.prev_sent = None, None
self.prev_stats: tuple[Any, ...] | None = None
self.counter = 0
self.up_samples = 5 # 1s
self._auto = True
@@ -199,7 +176,7 @@ class LivestreamBitrateController(AsyncTaskRunner):
if not self._auto:
continue
loss_rate = await self._sample()
loss_rate = self._sample()
if loss_rate is None:
continue
if loss_rate >= self.med_level and self.level > 0:
@@ -216,22 +193,18 @@ class LivestreamBitrateController(AsyncTaskRunner):
self.counter = 0
self._publish(self.bitrates[self.level])
async def _sample(self) -> float | None:
report = await self.pc.getStats()
packets_lost = packets_sent = 0
for s in report.values():
if s.type == "remote-inbound-rtp":
packets_lost += s.packetsLost
elif s.type == "outbound-rtp":
packets_sent += s.packetsSent
if self.prev_lost is None:
self.prev_lost, self.prev_sent = packets_lost, packets_sent
def _sample(self) -> float | None:
report = next(iter(self.get_stats().values()), None)
if report is None:
return None
lost_delta = max(0, packets_lost - self.prev_lost)
sent_delta = max(0, packets_sent - self.prev_sent)
self.prev_lost, self.prev_sent = packets_lost, packets_sent
return lost_delta / sent_delta if sent_delta else 0.0
current = (report.ssrc, report.fraction_lost, report.packets_lost, report.highest_seq_no, report.jitter, report.lsr, report.dlsr)
if self.prev_stats == current:
return None
self.prev_stats = current
loss_rate = report.fraction_lost / 256
return loss_rate
def _publish(self, bitrate: float):
self.params.put(self.param_name, bitrate)
@@ -248,17 +221,15 @@ class StreamSession:
shared_pub_master = DynamicPubMaster([])
def __init__(self, body: StreamRequestBody, debug_mode: bool = False):
if debug_mode:
from aiortc.mediastreams import VideoStreamTrack
from openpilot.system.webrtc.device.video import LiveStreamVideoStreamTrack
from teleoprtc.builder import WebRTCAnswerBuilder
self.identifier = str(uuid.uuid4())
self.params = Params()
builder = WebRTCAnswerBuilder(body.sdp)
builder = WebRTCAnswerBuilder(body.sdp, bind_address=_default_route_ip())
self.enabled = body.enabled
self.video_track = LiveStreamVideoStreamTrack(body.init_camera, self.enabled) if not debug_mode else VideoStreamTrack()
self.video_track = LiveStreamVideoStreamTrack(body.init_camera, self.enabled)
builder.add_video_stream(body.init_camera, self.video_track)
self.stream = builder.stream()
@@ -270,7 +241,7 @@ class StreamSession:
self.incoming_bridge = CerealIncomingMessageProxy(self.shared_pub_master)
if len(body.bridge_services_out) > 0:
self.outgoing_bridge = CerealOutgoingMessageProxy(body.bridge_services_out, self.enabled)
self.bitrate_controller = LivestreamBitrateController(self.stream.peer_connection, self.params, self.enabled)
self.bitrate_controller = LivestreamBitrateController(self.stream.get_receiver_report_stats, self.params, self.enabled)
self.run_task: asyncio.Task | None = None
self._cleanup_lock = asyncio.Lock()
@@ -305,13 +276,16 @@ class StreamSession:
case "livestreamCameraSwitch":
self.video_track.switch_camera(payload["data"]["camera"])
case "livestreamSettings":
self.bitrate_controller.set_quality(payload["data"]["quality"])
if self.bitrate_controller is not None:
self.bitrate_controller.set_quality(payload["data"]["quality"])
case "livestreamVideoEnable":
enabled = payload["data"]["enabled"]
self.enabled = enabled
self.video_track.enable(enabled)
self.outgoing_bridge.enable(enabled)
self.bitrate_controller.enable(enabled)
if self.outgoing_bridge is not None:
self.outgoing_bridge.enable(enabled)
if self.bitrate_controller is not None:
self.bitrate_controller.enable(enabled)
if not enabled:
self.params.put("LivestreamRequestKeyframe", True)
case "clockSync":
@@ -325,7 +299,8 @@ class StreamSession:
case _:
if payload.get("type") not in self.incoming_bridge_services:
return
self.incoming_bridge.send(message)
if self.incoming_bridge is not None:
self.incoming_bridge.send(message)
except Exception:
self.logger.exception("Cereal incoming proxy failure")
@@ -341,7 +316,8 @@ class StreamSession:
channel = self.stream.get_messaging_channel()
self.outgoing_bridge.add_channel(channel)
self.outgoing_bridge.start()
self.bitrate_controller.start()
if self.bitrate_controller is not None:
self.bitrate_controller.start()
self.logger.info("Stream session (%s) connected", self.identifier)
await self.stream.wait_for_disconnection()
@@ -357,7 +333,8 @@ class StreamSession:
return
self._cleanup_done = True
self.params.put("LivestreamRequestKeyframe", False)
await self.bitrate_controller.stop()
if self.bitrate_controller is not None:
await self.bitrate_controller.stop()
if self.outgoing_bridge is not None:
await self.outgoing_bridge.stop()
if self.video_track is not None:
@@ -417,7 +394,12 @@ async def handle_get_stream(state: ServerState, raw_body: bytes) -> tuple[int, b
session = StreamSession(body, debug_mode)
stream_dict[session.identifier] = session
try:
answer = await session.get_answer()
answer = await asyncio.wait_for(session.get_answer(), timeout=30)
except TimeoutError:
await session.stop()
stream_dict.pop(session.identifier, None)
logging.getLogger("webrtcd").exception("Timed out creating stream answer")
raise
except Exception:
await session.stop()
stream_dict.pop(session.identifier, None)
@@ -537,7 +519,7 @@ class WebrtcdHandler(BaseHTTPRequestHandler):
def do_OPTIONS(self) -> None:
self._dispatch_request()
def log_message(self, fmt, *args) -> None:
def log_message(self, format: str, *args: object) -> None: # noqa: A002 # stdlib override
# silence default access logging; errors are logged explicitly in _dispatch_request
pass
@@ -558,9 +540,6 @@ async def _shutdown(server: WebrtcdHTTPServer, state: ServerState, loop: asyncio
def prewarm_stream_session_imports(debug_mode: bool = False) -> None:
if debug_mode:
from aiortc.mediastreams import VideoStreamTrack
assert VideoStreamTrack
from openpilot.system.webrtc.device.video import LiveStreamVideoStreamTrack
from teleoprtc.builder import WebRTCAnswerBuilder
assert LiveStreamVideoStreamTrack
@@ -587,13 +566,14 @@ def webrtcd_thread(host: str, port: int, debug: bool):
http_thread.start()
shutting_down = False
shutdown_task = None
def request_shutdown() -> None:
nonlocal shutting_down
nonlocal shutting_down, shutdown_task
if shutting_down:
return
shutting_down = True
loop.create_task(_shutdown(server, state, loop))
shutdown_task = loop.create_task(_shutdown(server, state, loop))
for sig in (signal.SIGINT, signal.SIGTERM):
loop.add_signal_handler(sig, request_shutdown)