mirror of
https://github.com/infiniteCable2/openpilot.git
synced 2026-09-11 02:33:41 +08:00
Jason Wen
7461f70fdb
# Conflicts: # README.md # SConstruct # conftest.py # docs/CARS.md # msgq_repo # opendbc_repo # openpilot/common/params_keys.h # openpilot/common/params_pyx.pyx # openpilot/common/tests/test_swaglog.cc # openpilot/selfdrive/car/card.py # openpilot/selfdrive/car/tests/test_car_interfaces.py # openpilot/selfdrive/car/tests/test_cruise_speed.py # openpilot/selfdrive/car/tests/test_models.py # openpilot/selfdrive/controls/controlsd.py # openpilot/selfdrive/controls/lib/latcontrol_torque.py # openpilot/selfdrive/controls/lib/longitudinal_planner.py # openpilot/selfdrive/controls/plannerd.py # openpilot/selfdrive/controls/radard.py # openpilot/selfdrive/controls/tests/test_longcontrol.py # openpilot/selfdrive/locationd/torqued.py # openpilot/selfdrive/modeld/modeld.py # openpilot/selfdrive/monitoring/dmonitoringd.py # openpilot/selfdrive/monitoring/test_monitoring.py # openpilot/selfdrive/selfdrived/selfdrived.py # openpilot/selfdrive/selfdrived/tests/test_alertmanager.py # openpilot/selfdrive/test/longitudinal_maneuvers/plant.py # openpilot/selfdrive/test/process_replay/migration.py # openpilot/selfdrive/test/process_replay/process_replay.py # openpilot/selfdrive/ui/feedback/feedbackd.py # openpilot/selfdrive/ui/layouts/settings/device.py # openpilot/selfdrive/ui/layouts/settings/toggles.py # openpilot/selfdrive/ui/mici/layouts/onboarding.py # openpilot/selfdrive/ui/onroad/augmented_road_view.py # openpilot/selfdrive/ui/tests/test_soundd.py # openpilot/selfdrive/ui/translations/app.pot # openpilot/selfdrive/ui/translations/app_de.po # openpilot/selfdrive/ui/translations/app_en.po # openpilot/selfdrive/ui/translations/app_es.po # openpilot/selfdrive/ui/translations/app_fr.po # openpilot/selfdrive/ui/translations/app_ja.po # openpilot/selfdrive/ui/translations/app_ko.po # openpilot/selfdrive/ui/translations/app_pt-BR.po # openpilot/selfdrive/ui/translations/app_th.po # openpilot/selfdrive/ui/translations/app_tr.po # openpilot/selfdrive/ui/translations/app_uk.po # openpilot/selfdrive/ui/translations/app_zh-CHS.po # openpilot/selfdrive/ui/translations/app_zh-CHT.po # openpilot/system/athena/athenad.py # openpilot/system/hardware/hardwared.py # openpilot/system/loggerd/deleter.py # openpilot/system/manager/process_config.py # openpilot/system/ui/lib/application.py # panda # pyproject.toml # tinygrad_repo # uv.lock
57 lines
1.9 KiB
Python
57 lines
1.9 KiB
Python
"""
|
|
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.
|
|
"""
|
|
# DISCLAIMER: This code is intended principally for development and debugging purposes.
|
|
# Although it provides a standalone entry point to the program, users should refer
|
|
# to the actual implementations for consumption. Usage outside of development scenarios
|
|
# is not advised and could lead to unpredictable results.
|
|
|
|
import threading
|
|
import traceback
|
|
|
|
from openpilot.cereal import messaging
|
|
from openpilot.common.gps import get_gps_location_service
|
|
from openpilot.common.params import Params
|
|
from openpilot.common.realtime import config_realtime_process
|
|
from openpilot.sunnypilot.selfdrive.controls.lib.speed_limit.common import Policy
|
|
from openpilot.sunnypilot.selfdrive.controls.lib.speed_limit.speed_limit_resolver import SpeedLimitResolver
|
|
from openpilot.sunnypilot.mapd.live_map_data import get_debug
|
|
|
|
|
|
def excepthook(args):
|
|
get_debug(f'MapD: Threading exception:\n{args}')
|
|
traceback.print_exception(args.exc_type, args.exc_value, args.exc_traceback)
|
|
|
|
|
|
def live_map_data_sp_thread():
|
|
config_realtime_process([0, 1, 2, 3], 5)
|
|
|
|
params = Params()
|
|
gps_location_service = get_gps_location_service(params)
|
|
|
|
while True:
|
|
live_map_data_sp_thread_debug(gps_location_service)
|
|
|
|
|
|
def live_map_data_sp_thread_debug(gps_location_service):
|
|
_sub_master = messaging.SubMaster(['carState', 'deviceMotion', 'liveMapDataSP', 'longitudinalPlanSP', 'carStateSP', gps_location_service])
|
|
_sub_master.update()
|
|
|
|
v_ego = _sub_master['carState'].vEgo
|
|
_resolver = SpeedLimitResolver()
|
|
_resolver.policy = Policy.car_state_priority
|
|
_resolver.update(v_ego, _sub_master)
|
|
print(_resolver.speed_limit, _resolver.distance, _resolver.source)
|
|
|
|
|
|
def main():
|
|
threading.excepthook = excepthook
|
|
live_map_data_sp_thread()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|