mirror of
https://github.com/sunnypilot/sunnypilot.git
synced 2026-10-01 16:53:41 +08:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| fd2e9a687e |
@@ -162,10 +162,6 @@ inline static std::unordered_map<std::string, ParamKeyAttributes> keys = {
|
||||
{"DeviceBootMode", {PERSISTENT | BACKUP, INT, "0"}},
|
||||
{"DevUIInfo", {PERSISTENT | BACKUP, INT, "0"}},
|
||||
{"EnableCopyparty", {PERSISTENT | BACKUP, BOOL}},
|
||||
{"SunnydriveEnabled", {PERSISTENT | BACKUP, BOOL}},
|
||||
{"SunnydriveAutoPairPrompt", {PERSISTENT | BACKUP, BOOL, "1"}},
|
||||
{"SunnydrivePairedClients", {PERSISTENT, JSON}},
|
||||
{"SunnydrivePairingRequests", {CLEAR_ON_MANAGER_START, JSON}},
|
||||
{"EnableGithubRunner", {PERSISTENT | BACKUP, BOOL}},
|
||||
{"GreenLightAlert", {PERSISTENT | BACKUP, BOOL, "0"}},
|
||||
{"GithubRunnerSufficientVoltage", {CLEAR_ON_MANAGER_START , BOOL}},
|
||||
|
||||
@@ -154,7 +154,7 @@ class SelfdriveD(CruiseHelper):
|
||||
self.state_machine = StateMachine()
|
||||
self.rk = Ratekeeper(100, print_delay_threshold=None)
|
||||
|
||||
self.ignored_processes = {'mapd', 'sunnydrived'}
|
||||
self.ignored_processes = {'mapd', }
|
||||
|
||||
# Determine startup event
|
||||
is_remote = build_metadata.openpilot.comma_remote or build_metadata.openpilot.sunnypilot_remote
|
||||
|
||||
@@ -4,10 +4,6 @@ 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.
|
||||
"""
|
||||
import math
|
||||
|
||||
import pyray as rl
|
||||
|
||||
from openpilot.selfdrive.ui.mici.layouts.settings import settings as OP
|
||||
from openpilot.selfdrive.ui.mici.layouts.settings.settings import SettingsBigButton
|
||||
from openpilot.selfdrive.ui.mici.layouts.settings.device import DeviceLayoutMici
|
||||
@@ -15,7 +11,6 @@ from openpilot.selfdrive.ui.mici.widgets.button import BigCircleButton
|
||||
from openpilot.selfdrive.ui.mici.widgets.dialog import BigConfirmationDialog, BigDialog
|
||||
from openpilot.selfdrive.ui.sunnypilot.mici.layouts.sunnylink import SunnylinkLayoutMici
|
||||
from openpilot.selfdrive.ui.sunnypilot.mici.layouts.models import ModelsLayoutMici
|
||||
from openpilot.selfdrive.ui.sunnypilot.mici.layouts.sunnydrive import SunnydriveLayoutMici
|
||||
from openpilot.selfdrive.ui.ui_state import ui_state
|
||||
from openpilot.system.ui.lib.application import gui_app, FontWeight
|
||||
from openpilot.system.ui.lib.multilang import tr
|
||||
@@ -34,40 +29,6 @@ class SunnylinkBigButton(SettingsBigButton):
|
||||
return 56
|
||||
|
||||
|
||||
def sunnydrive_icon(width, height):
|
||||
"""The sunnydrive icon (a sun over a car roof), drawn once in code so it needs no image asset.
|
||||
Drawn at 4x on a 336x312 grid, then scaled down for smooth edges."""
|
||||
s = 4
|
||||
img = rl.gen_image_color(336 * s, 312 * s, rl.BLANK)
|
||||
p = rl.ffi.addressof(img)
|
||||
|
||||
def dot(x, y, r, color):
|
||||
rl.image_draw_circle(p, int(x * s), int(y * s), int(r * s), color)
|
||||
|
||||
def roof(r, color): # a shallow circular arc, apex under the sun, ends reaching down beside the wheels
|
||||
for i in range(-52, 53):
|
||||
t = math.radians(i)
|
||||
dot(168 + 194 * math.sin(t), 399 - 194 * math.cos(t), r, color)
|
||||
|
||||
dot(168, 100, 100, rl.WHITE) # the sun
|
||||
roof(28, rl.BLANK) # a gap between the sun and the roof
|
||||
roof(14, rl.WHITE) # the roof
|
||||
rl.image_draw_rectangle(p, 70 * s, 284 * s, 196 * s, 28 * s, rl.WHITE) # the car's body between its wheels
|
||||
for x in (84, 252):
|
||||
dot(x, 300, 18, rl.WHITE)
|
||||
dot(x, 300, 11, rl.BLANK) # wheel
|
||||
rl.image_resize(p, width, height)
|
||||
texture = rl.load_texture_from_image(img)
|
||||
rl.unload_image(img)
|
||||
rl.set_texture_filter(texture, rl.TextureFilter.TEXTURE_FILTER_BILINEAR)
|
||||
return texture
|
||||
|
||||
|
||||
class SunnydriveBigButton(SettingsBigButton):
|
||||
def _get_label_font_size(self):
|
||||
return 54 # "sunnydrive" wraps under the icon at 64
|
||||
|
||||
|
||||
class SettingsLayoutSP(OP.SettingsLayout):
|
||||
def __init__(self):
|
||||
OP.SettingsLayout.__init__(self)
|
||||
@@ -89,10 +50,6 @@ class SettingsLayoutSP(OP.SettingsLayout):
|
||||
models_btn = SettingsBigButton(tr("models"), "", gui_app.texture("../../sunnypilot/selfdrive/assets/offroad/icon_models.png", ICON_SIZE, ICON_SIZE))
|
||||
models_btn.set_click_callback(lambda: gui_app.push_widget(models_panel))
|
||||
|
||||
sunnydrive_panel = SunnydriveLayoutMici()
|
||||
sunnydrive_btn = SunnydriveBigButton(tr("sunnydrive"), "", sunnydrive_icon(64, 59))
|
||||
sunnydrive_btn.set_click_callback(lambda: gui_app.push_widget(sunnydrive_panel))
|
||||
|
||||
# onroad: enable button sits at the front (left of toggles)
|
||||
self._enable_offroad_btn_onroad = BigCircleButton(self.icon_offroad_enable, red=True)
|
||||
self._enable_offroad_btn_onroad.set_click_callback(lambda: self._handle_always_offroad(True))
|
||||
@@ -115,7 +72,6 @@ class SettingsLayoutSP(OP.SettingsLayout):
|
||||
# front slots (only one ever visible at a time): exit-always-offroad, then enable-onroad
|
||||
items.insert(0, self._enable_offroad_btn_onroad)
|
||||
items.insert(0, self._disable_offroad_btn)
|
||||
items.insert(0, sunnydrive_btn)
|
||||
# end slot: enable-offroad (right of developer)
|
||||
items.append(self._enable_offroad_btn_offroad)
|
||||
|
||||
|
||||
@@ -1,71 +0,0 @@
|
||||
import time
|
||||
|
||||
from openpilot.selfdrive.ui.mici.widgets.button import BigButton, BigParamControl
|
||||
from openpilot.selfdrive.ui.mici.widgets.dialog import BigConfirmationDialog
|
||||
from openpilot.selfdrive.ui.ui_state import ui_state
|
||||
from openpilot.sunnypilot.sunnydrive import pairing
|
||||
from openpilot.system.ui.lib.application import gui_app
|
||||
from openpilot.system.ui.widgets.scroller import NavScroller
|
||||
|
||||
|
||||
class SunnydriveLayoutMici(NavScroller):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self._pending_id = ""
|
||||
self._prompted_id = ""
|
||||
self._approval_dialog = None
|
||||
self._opened_for_prompt = False
|
||||
self._next_prompt_check = 0.0
|
||||
self._enable = BigParamControl("enabled", "SunnydriveEnabled")
|
||||
self._pairing_prompts = BigParamControl("allow pairing", "SunnydriveAutoPairPrompt")
|
||||
self._unpair = BigButton("paired phones", "none")
|
||||
self._unpair.set_click_callback(self._confirm_unpair_all)
|
||||
self._scroller.add_widgets([self._enable, self._pairing_prompts, self._unpair])
|
||||
gui_app.add_nav_stack_tick(self._pairing_tick)
|
||||
|
||||
def _update_state(self):
|
||||
super()._update_state()
|
||||
self._enable.refresh()
|
||||
self._pairing_prompts.refresh()
|
||||
requests = pairing.pairing_requests()
|
||||
pending = next(((request_id, request) for request_id, request in reversed(list(requests.items())) if request.get("status") == "pending"), None)
|
||||
self._pending_id = pending[0] if pending else ""
|
||||
count = len(pairing.paired_clients())
|
||||
self._unpair.set_value(f"{count} paired" if count else "none")
|
||||
self._unpair.set_visible(count > 0)
|
||||
|
||||
def _pairing_tick(self):
|
||||
if time.monotonic() < self._next_prompt_check:
|
||||
return
|
||||
self._next_prompt_check = time.monotonic() + 0.5
|
||||
if not ui_state.params.get_bool("SunnydriveAutoPairPrompt"):
|
||||
return
|
||||
requests = reversed(list(pairing.pairing_requests().items()))
|
||||
pending = next(((request_id, request) for request_id, request in requests if request.get("status") == "pending"), None)
|
||||
answered = not pending or pending[0] == self._prompted_id # approved, expired, or its slide was dismissed
|
||||
if self._opened_for_prompt and answered and gui_app.get_active_widget() is self: # back to whatever was showing
|
||||
self._opened_for_prompt = False
|
||||
self.dismiss()
|
||||
if pending and self._prompted_id != pending[0]:
|
||||
self._pending_id = self._prompted_id = pending[0]
|
||||
if self._approval_dialog and gui_app.widget_in_stack(self._approval_dialog):
|
||||
self._approval_dialog.dismiss(self._confirm_pending)
|
||||
else:
|
||||
self._confirm_pending()
|
||||
|
||||
def _confirm_pending(self):
|
||||
if not self._pending_id or (self._approval_dialog and gui_app.widget_in_stack(self._approval_dialog)):
|
||||
return
|
||||
request_id = self._pending_id
|
||||
icon = gui_app.texture("icons_mici/settings/device/pair.png", 64, 64)
|
||||
if not gui_app.widget_in_stack(self):
|
||||
# The screen right under a dialog keeps handling touches, and onroad that's the driving view,
|
||||
# whose swipe-left bookmark is the same gesture as this slider. Showing this panel underneath keeps it out.
|
||||
gui_app.push_widget(self)
|
||||
self._opened_for_prompt = True
|
||||
self._approval_dialog = BigConfirmationDialog("slide to approve phone", icon, lambda: pairing.approve_request(request_id))
|
||||
gui_app.push_widget(self._approval_dialog)
|
||||
|
||||
def _confirm_unpair_all(self):
|
||||
icon = gui_app.texture("icons_mici/settings/network/new/trash.png", 54, 64)
|
||||
gui_app.push_widget(BigConfirmationDialog("slide to unpair all phones", icon, lambda: pairing.unpair(), red=True))
|
||||
@@ -1,132 +0,0 @@
|
||||
"""Local Sunnydrive pairing shared by sunnydrived and the comma UI."""
|
||||
|
||||
import hashlib
|
||||
import re
|
||||
import secrets
|
||||
import time
|
||||
from datetime import UTC, datetime
|
||||
|
||||
from openpilot.common.params import Params
|
||||
|
||||
|
||||
PAIRED_KEY = "SunnydrivePairedClients"
|
||||
REQUESTS_KEY = "SunnydrivePairingRequests"
|
||||
PAIRING_SECONDS = 5 * 60
|
||||
CLIENT_ID_RE = re.compile(r"[A-Za-z0-9_-]{16,128}")
|
||||
|
||||
|
||||
def _params(params=None):
|
||||
return params or Params()
|
||||
|
||||
|
||||
def _dict(params, key):
|
||||
value = params.get(key)
|
||||
return value if isinstance(value, dict) else {}
|
||||
|
||||
|
||||
def device_id(params=None):
|
||||
params = _params(params)
|
||||
return params.get("DongleId") or params.get("HardwareSerial") or "unregistered"
|
||||
|
||||
|
||||
def device_name(params=None):
|
||||
identity = device_id(params)
|
||||
return f"comma {identity[-8:]}" if identity != "unregistered" else "comma"
|
||||
|
||||
|
||||
def is_offroad(params=None):
|
||||
return _params(params).get_bool("IsOffroad")
|
||||
|
||||
|
||||
def _require_allowed(params):
|
||||
if not params.get_bool("SunnydriveAutoPairPrompt"): # the comma's "allow pairing" switch
|
||||
raise PermissionError("pairing is turned off on the comma")
|
||||
|
||||
|
||||
def paired_clients(params=None):
|
||||
return _dict(_params(params), PAIRED_KEY)
|
||||
|
||||
|
||||
def is_paired(client_id, params=None):
|
||||
return client_id in paired_clients(params)
|
||||
|
||||
|
||||
def authorized(token, params=None):
|
||||
if not token:
|
||||
return False
|
||||
digest = hashlib.sha256(token.encode()).hexdigest()
|
||||
return any(secrets.compare_digest(digest, str(client.get("token_hash", ""))) for client in paired_clients(params).values())
|
||||
|
||||
|
||||
def _pair(client_id, name, params):
|
||||
if not CLIENT_ID_RE.fullmatch(client_id):
|
||||
raise ValueError("invalid client id")
|
||||
token = secrets.token_urlsafe(32)
|
||||
clients = paired_clients(params)
|
||||
clients[client_id] = {"name": str(name or "Sunnydrive phone")[:80], "token_hash": hashlib.sha256(token.encode()).hexdigest(),
|
||||
"paired_at": int(datetime.now(UTC).timestamp())}
|
||||
params.put(PAIRED_KEY, clients, block=True)
|
||||
return token
|
||||
|
||||
|
||||
def request_pairing(client_id, name, params=None):
|
||||
if not CLIENT_ID_RE.fullmatch(client_id):
|
||||
raise ValueError("invalid client id")
|
||||
params = _params(params)
|
||||
_require_allowed(params)
|
||||
now = int(time.monotonic())
|
||||
requests = {key: value for key, value in _dict(params, REQUESTS_KEY).items()
|
||||
if int(value.get("expires", 0)) > now and value.get("client_id") != client_id}
|
||||
request_id = secrets.token_urlsafe(18)
|
||||
requests[request_id] = {"client_id": client_id, "name": str(name or "Sunnydrive phone")[:80], "status": "pending", "expires": now + PAIRING_SECONDS}
|
||||
params.put(REQUESTS_KEY, requests, block=True)
|
||||
return request_id
|
||||
|
||||
|
||||
def pairing_requests(params=None):
|
||||
params = _params(params)
|
||||
now = int(time.monotonic())
|
||||
requests = _dict(params, REQUESTS_KEY)
|
||||
fresh = {key: value for key, value in requests.items() if int(value.get("expires", 0)) > now}
|
||||
if fresh != requests:
|
||||
params.put(REQUESTS_KEY, fresh, block=True) if fresh else params.remove(REQUESTS_KEY)
|
||||
return fresh
|
||||
|
||||
|
||||
def approve_request(request_id, params=None):
|
||||
params = _params(params)
|
||||
requests = pairing_requests(params)
|
||||
request = requests.get(request_id)
|
||||
if not request or request.get("status") != "pending":
|
||||
return False
|
||||
request["token"] = _pair(request["client_id"], request.get("name"), params)
|
||||
request["status"] = "approved"
|
||||
requests[request_id] = request
|
||||
params.put(REQUESTS_KEY, requests, block=True)
|
||||
return True
|
||||
|
||||
|
||||
def consume_request(request_id, params=None):
|
||||
params = _params(params)
|
||||
requests = pairing_requests(params)
|
||||
request = requests.get(request_id)
|
||||
if not request:
|
||||
return {"status": "expired"}
|
||||
result = {"status": request.get("status", "pending")}
|
||||
if request.get("status") == "approved":
|
||||
result["token"] = request.get("token", "")
|
||||
requests.pop(request_id, None)
|
||||
params.put(REQUESTS_KEY, requests, block=True) if requests else params.remove(REQUESTS_KEY)
|
||||
return result
|
||||
|
||||
|
||||
def unpair(client_id=None, params=None):
|
||||
params = _params(params)
|
||||
clients = paired_clients(params)
|
||||
if client_id is None:
|
||||
params.remove(PAIRED_KEY)
|
||||
return bool(clients)
|
||||
removed = clients.pop(client_id, None) is not None
|
||||
if removed:
|
||||
params.put(PAIRED_KEY, clients, block=True) if clients else params.remove(PAIRED_KEY)
|
||||
return removed
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,172 +0,0 @@
|
||||
import json
|
||||
import threading
|
||||
import time
|
||||
import unittest
|
||||
import unittest.mock
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from tempfile import TemporaryDirectory
|
||||
from unittest.mock import patch
|
||||
from urllib.error import HTTPError
|
||||
from urllib.request import Request, urlopen
|
||||
|
||||
from openpilot.sunnypilot.sunnydrive import sunnydrived
|
||||
from openpilot.sunnypilot.sunnydrive import pairing
|
||||
from openpilot.sunnypilot.sunnydrive.sunnydrived import SunnydriveServer
|
||||
|
||||
|
||||
class SunnydriveApiTest(unittest.TestCase):
|
||||
class FakeParams:
|
||||
def __init__(self):
|
||||
self.values = {"DongleId": "comma-test-12345678", "IsOffroad": True, "SunnydriveAutoPairPrompt": True}
|
||||
|
||||
def get(self, key):
|
||||
return self.values.get(key)
|
||||
|
||||
def get_bool(self, key):
|
||||
return bool(self.values.get(key))
|
||||
|
||||
def put(self, key, value, block=False):
|
||||
self.values[key] = value
|
||||
|
||||
def remove(self, key):
|
||||
self.values.pop(key, None)
|
||||
|
||||
def test_slide_approval_pairing(self):
|
||||
params = self.FakeParams()
|
||||
request = pairing.request_pairing("phone_abcdefghijklmnop", "Second phone", params)
|
||||
replacement = pairing.request_pairing("phone_abcdefghijklmnop", "Second phone", params)
|
||||
self.assertNotEqual(replacement, request)
|
||||
request = replacement
|
||||
self.assertEqual(len(pairing.pairing_requests(params)), 1)
|
||||
self.assertTrue(pairing.approve_request(request, params))
|
||||
approved = pairing.consume_request(request, params)
|
||||
self.assertEqual(approved["status"], "approved")
|
||||
self.assertTrue(pairing.authorized(approved["token"], params))
|
||||
self.assertEqual(pairing.consume_request(request, params)["status"], "expired")
|
||||
|
||||
params.values["SunnydriveAutoPairPrompt"] = False
|
||||
with self.assertRaisesRegex(PermissionError, "turned off"):
|
||||
pairing.request_pairing("phone_switched_off_123456", "Off", params)
|
||||
params.values["SunnydriveAutoPairPrompt"] = True
|
||||
|
||||
params.values["IsOffroad"] = False # pairing works onroad too
|
||||
self.assertTrue(pairing.approve_request(pairing.request_pairing("phone_onroad_123456789", "Onroad", params), params))
|
||||
|
||||
def test_subscribed_services_exist(self): # a renamed service crashes the sample loop and freezes telemetry
|
||||
from openpilot.cereal.services import SERVICE_LIST
|
||||
self.assertEqual([name for name in sunnydrived.SERVICES if name not in SERVICE_LIST], [])
|
||||
|
||||
def test_discovery_is_metadata_only(self):
|
||||
packet = sunnydrived.DISCOVERY_PREFIX + b'{"v":1,"nonce":"12345678","clientId":"phone_1234567890123456"}'
|
||||
with patch.object(pairing, "device_id", return_value="comma-id"), \
|
||||
patch.object(pairing, "device_name", return_value="comma test"), \
|
||||
patch.object(pairing, "is_paired", return_value=False), \
|
||||
patch.object(pairing, "is_offroad", return_value=True):
|
||||
reply = sunnydrived.discovery_response(packet)
|
||||
body = json.loads(reply[len(sunnydrived.DISCOVERY_PREFIX):])
|
||||
self.assertEqual(set(body), {"v", "nonce", "deviceId", "name", "httpPort", "apiVersion", "paired", "offroad"})
|
||||
self.assertNotIn("token", body)
|
||||
|
||||
def test_settings_follow_sunnylink_safety_rules(self):
|
||||
with patch("openpilot.common.params.Params") as params_class, patch.object(sunnydrived, "_setting_engaged", return_value=False) as engaged:
|
||||
params_class.return_value.get_bool.return_value = False
|
||||
params_class.return_value.get_type.return_value = 1
|
||||
sunnydrived.sunnylink_set("AlphaLongitudinalEnabled", True)
|
||||
params_class.return_value.put.assert_called_once()
|
||||
with self.assertRaisesRegex(PermissionError, "onroad or engaged"):
|
||||
sunnydrived.sunnylink_set("Mads", True)
|
||||
engaged.return_value = True
|
||||
with self.assertRaisesRegex(PermissionError, "onroad or engaged"):
|
||||
sunnydrived.sunnylink_set("AlphaLongitudinalEnabled", True)
|
||||
params_class.return_value.put.assert_called_once()
|
||||
params_class.return_value.get.return_value = b"1" # TorqueParamsOverrideEnabled satisfies the schema's onroad alternative
|
||||
params_class.return_value.get_type.return_value = 3
|
||||
sunnydrived.sunnylink_set("TorqueParamsOverrideFriction", 0.1)
|
||||
self.assertEqual(params_class.return_value.put.call_count, 2)
|
||||
params_class.return_value.remove.assert_not_called()
|
||||
|
||||
def test_live_frames_start_at_keyframe(self):
|
||||
def msg(key, data):
|
||||
frame = SimpleNamespace(idx=SimpleNamespace(flags=0x8 if key else 0), header=b"H" if key else b"", data=data)
|
||||
return SimpleNamespace(which=lambda: "livestreamCabinEncodeData", livestreamCabinEncodeData=frame)
|
||||
params = unittest.mock.MagicMock()
|
||||
with patch.object(sunnydrived.messaging, "sub_sock"), \
|
||||
patch.object(sunnydrived.messaging, "recv_one", side_effect=[msg(False, b"p0"), None, msg(True, b"k1"), msg(False, b"p2")]):
|
||||
frames = sunnydrived.live_frames("driver", params)
|
||||
self.assertEqual([next(frames) for _ in range(3)], [None, (True, b"Hk1"), (False, b"p2")])
|
||||
self.assertEqual([c.args for c in params.put_bool.call_args_list], [("LivestreamRequestKeyframe", True), ("LivestreamRequestKeyframe", False)])
|
||||
|
||||
def test_route_list_cache(self):
|
||||
with TemporaryDirectory() as folder, patch.object(sunnydrived, "REALDATA", Path(folder)):
|
||||
sunnydrived._connect_routes.cache_clear()
|
||||
try:
|
||||
self.assertEqual(sunnydrived.connect_routes(), [])
|
||||
self.assertEqual(sunnydrived.connect_routes(), [])
|
||||
self.assertEqual(sunnydrived._connect_routes.cache_info().hits, 1)
|
||||
finally:
|
||||
sunnydrived._connect_routes.cache_clear()
|
||||
|
||||
def test_api_only(self):
|
||||
server = SunnydriveServer(("127.0.0.1", 0))
|
||||
server.publish_telemetry({"timestampMs": 123})
|
||||
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||||
thread.start()
|
||||
try:
|
||||
base = f"http://127.0.0.1:{server.server_address[1]}"
|
||||
request = Request(base + "/telemetry", headers={"Origin": "https://ai.sunnypilot.sunnydrive"})
|
||||
with urlopen(request, timeout=2) as response:
|
||||
self.assertEqual(json.load(response), server.telemetry)
|
||||
self.assertEqual(response.headers["Access-Control-Allow-Origin"], "https://ai.sunnypilot.sunnydrive")
|
||||
with urlopen(base + "/telemetry/stream", timeout=2) as first, urlopen(base + "/telemetry/stream", timeout=2) as second:
|
||||
self.assertEqual(first.readline(), b'data: {"timestampMs":123}\n')
|
||||
self.assertEqual(second.readline(), b'data: {"timestampMs":123}\n')
|
||||
first.readline()
|
||||
second.readline()
|
||||
server.publish_telemetry({"timestampMs": 456})
|
||||
self.assertEqual(first.readline(), b'data: {"timestampMs":456}\n')
|
||||
self.assertEqual(second.readline(), b'data: {"timestampMs":456}\n')
|
||||
for path in ("/", "/index.html", "/replay", "/youtube-playlists"):
|
||||
with self.assertRaises(HTTPError) as error:
|
||||
urlopen(base + path, timeout=2)
|
||||
self.assertEqual(error.exception.code, 404)
|
||||
finally:
|
||||
server.shutdown()
|
||||
server.server_close()
|
||||
thread.join(timeout=2)
|
||||
|
||||
def test_unpaired_phone_cannot_read_api(self):
|
||||
server = SunnydriveServer(("127.0.0.1", 0), allow_loopback=False)
|
||||
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||||
thread.start()
|
||||
try:
|
||||
base = f"http://127.0.0.1:{server.server_address[1]}"
|
||||
with patch.object(pairing, "authorized", return_value=False):
|
||||
with self.assertRaises(HTTPError) as error:
|
||||
urlopen(base + "/telemetry", timeout=2)
|
||||
self.assertEqual(error.exception.code, 401)
|
||||
with patch.object(pairing, "device_id", return_value="comma-id"), \
|
||||
patch.object(pairing, "device_name", return_value="comma test"), \
|
||||
patch.object(pairing, "is_paired", return_value=False), \
|
||||
patch.object(pairing, "is_offroad", return_value=True):
|
||||
with urlopen(base + "/pair/info?client_id=phone_1234567890123456", timeout=2) as response:
|
||||
self.assertEqual(json.load(response)["deviceId"], "comma-id")
|
||||
with patch.object(pairing, "authorized", return_value=True):
|
||||
with urlopen(base + "/telemetry?auth=paired", timeout=2) as response:
|
||||
self.assertIn("timestampMs", json.load(response))
|
||||
with patch.object(pairing, "authorized", return_value=True) as authorized: # unpairing cuts an open stream within a second
|
||||
with urlopen(base + "/telemetry/stream?auth=paired", timeout=5) as stream:
|
||||
self.assertTrue(stream.readline().startswith(b"data: "))
|
||||
stream.readline() # the event's blank line
|
||||
authorized.return_value = False
|
||||
time.sleep(1.1)
|
||||
server.publish_telemetry({"timestampMs": 789})
|
||||
self.assertEqual(stream.read(), b"")
|
||||
finally:
|
||||
server.shutdown()
|
||||
server.server_close()
|
||||
thread.join(timeout=2)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -71,9 +71,6 @@ def livestream(started: bool, params: Params, CP: car.CarParams) -> bool:
|
||||
def use_copyparty(started, params, CP: car.CarParams) -> bool:
|
||||
return bool(params.get_bool("EnableCopyparty"))
|
||||
|
||||
def sunnydrive_enabled(started: bool, params: Params, CP: car.CarParams) -> bool:
|
||||
return params.get_bool("SunnydriveEnabled")
|
||||
|
||||
def sunnylink_ready_shim(started, params, CP: car.CarParams) -> bool:
|
||||
"""Shim for sunnylink_ready to match the process manager signature."""
|
||||
return sunnylink_ready(params)
|
||||
@@ -170,7 +167,6 @@ procs = [
|
||||
|
||||
# sunnypilot
|
||||
procs += [
|
||||
PythonProcess("sunnydrived", "openpilot.sunnypilot.sunnydrive.sunnydrived", sunnydrive_enabled),
|
||||
# Models
|
||||
PythonProcess("models_manager", "openpilot.sunnypilot.models.manager", only_offroad),
|
||||
NativeProcess("modeld_tinygrad", "openpilot/sunnypilot/modeld_v2", ["./modeld"], and_(only_onroad, is_tinygrad_model)),
|
||||
|
||||
Reference in New Issue
Block a user