Files
openpilot-evo/sunnypilot/sunnylink/utils.py
T
DevTekVE 0871abcf55 bugfix: fix fetching params for sunnylink and backup (#1177)
* Hotfix for the params stuff until I rework this properly and leverage the new data types

* Revert "Hotfix for the params stuff until I rework this properly and leverage the new data types"

This reverts commit c6031b29d92d3ff5b679ffce3ba53611bb2dba0e.

* refactor: enhance getParams function to support JSON and bytes types with optional compression

* refactor: add TODO for enhancing server support of metadata in sunnylinkd.py

* lint and clean

* refactor: update value handling in getParams to return decoded values for JSON serialization

* refactor: simplify params_dict initialization by removing type hint

* refactor: update response handling in getParams to include JSON serialization of params

* refactor: update response handling in getParams to include JSON serialization of params

* Add to dic types

* refactor: extract get_param_as_byte function for improved parameter handling and fix backup inconsistencies

* cleanup

* ensure error propagates on backup fail
2025-09-04 14:45:37 +02:00

71 lines
2.3 KiB
Python

import json
from sunnypilot.sunnylink.api import SunnylinkApi, UNREGISTERED_SUNNYLINK_DONGLE_ID
from openpilot.common.params import Params, ParamKeyType
from openpilot.system.version import is_prebuilt
def get_sunnylink_status(params=None) -> tuple[bool, bool]:
"""Get the status of Sunnylink on the device. Returns a tuple of (is_sunnylink_enabled, is_registered)."""
params = params or Params()
is_sunnylink_enabled = params.get_bool("SunnylinkEnabled")
is_registered = params.get("SunnylinkDongleId") not in (None, UNREGISTERED_SUNNYLINK_DONGLE_ID)
return is_sunnylink_enabled, is_registered
def sunnylink_ready(params=None) -> bool:
"""Check if the device is ready to communicate with Sunnylink. That means it is enabled and registered."""
params = params or Params()
is_sunnylink_enabled, is_registered = get_sunnylink_status(params)
return is_sunnylink_enabled and is_registered
def use_sunnylink_uploader(params) -> bool:
"""Check if the device is ready to use Sunnylink and the uploader is enabled."""
return sunnylink_ready(params) and params.get_bool("EnableSunnylinkUploader")
def sunnylink_need_register(params=None) -> bool:
"""Check if the device needs to be registered with Sunnylink."""
params = params or Params()
is_sunnylink_enabled, is_registered = get_sunnylink_status(params)
return is_sunnylink_enabled and not is_registered
def register_sunnylink():
"""Register the device with Sunnylink if it is enabled."""
extra_args = {}
if not Params().get_bool("SunnylinkEnabled"):
print("Sunnylink is not enabled. Exiting.")
exit(0)
if not is_prebuilt():
extra_args = {
"verbose": True,
"timeout": 60
}
sunnylink_id = SunnylinkApi(None).register_device(None, **extra_args)
print(f"SunnyLinkId: {sunnylink_id}")
def get_api_token():
"""Get the API token for the device."""
params = Params()
sunnylink_dongle_id = params.get("SunnylinkDongleId")
sunnylink_api = SunnylinkApi(sunnylink_dongle_id)
token = sunnylink_api.get_token()
print(f"API Token: {token}")
def get_param_as_byte(param_name: str) -> bytes:
params = Params()
param = params.get(param_name)
param_type = params.get_type(param_name)
if param_type == ParamKeyType.BYTES:
return bytes(param)
elif param_type == ParamKeyType.JSON:
return json.dumps(param).encode('utf-8')
return str(param).encode('utf-8')