feat(params): add support for parameter metadata retrieval

- Introduced `getKeyMetadata` method for accessing metadata associated with params.
- Enhanced `getParamsAllKeysV1` to include metadata parsing and optional dynamic enum generation.
- Extended unit tests to verify metadata parsing, enum mapping, and edge cases.
This commit is contained in:
DevTekVE
2025-11-27 21:56:32 +01:00
parent c229eb4f38
commit 865b695ff9
6 changed files with 166 additions and 11 deletions
+4
View File
@@ -129,6 +129,10 @@ std::optional<std::string> Params::getKeyDefaultValue(const std::string &key) {
return keys[key].default_value;
}
std::optional<std::string> Params::getKeyMetadata(const std::string &key) {
return keys[key].metadata;
}
int Params::put(const char* key, const char* value, size_t value_size) {
// Information about safely and atomically writing a file: https://lwn.net/Articles/457667/
// 1) Create temp file
+2
View File
@@ -36,6 +36,7 @@ struct ParamKeyAttributes {
uint32_t flags;
ParamKeyType type;
std::optional<std::string> default_value = std::nullopt;
std::optional<std::string> metadata = std::nullopt;
};
class Params {
@@ -51,6 +52,7 @@ public:
ParamKeyFlag getKeyFlag(const std::string &key);
ParamKeyType getKeyType(const std::string &key);
std::optional<std::string> getKeyDefaultValue(const std::string &key);
std::optional<std::string> getKeyMetadata(const std::string &key);
inline std::string getParamPath(const std::string &key = {}) {
return params_path + params_prefix + (key.empty() ? "" : "/" + key);
}
+2 -2
View File
@@ -82,7 +82,7 @@ inline static std::unordered_map<std::string, ParamKeyAttributes> keys = {
{"LiveTorqueParameters", {PERSISTENT | DONT_LOG, BYTES}},
{"LocationFilterInitialState", {PERSISTENT, BYTES}},
{"LongitudinalManeuverMode", {CLEAR_ON_MANAGER_START | CLEAR_ON_OFFROAD_TRANSITION, BOOL}},
{"LongitudinalPersonality", {PERSISTENT | BACKUP, INT, std::to_string(static_cast<int>(cereal::LongitudinalPersonality::STANDARD))}},
{"LongitudinalPersonality", {PERSISTENT | BACKUP, INT, std::to_string(static_cast<int>(cereal::LongitudinalPersonality::STANDARD)), R"({"enum": "LongitudinalPersonality", "title": "Driving Personality", "description": "Adjusts the driving style of the openpilot longitudinal control."})"}},
{"NetworkMetered", {PERSISTENT | BACKUP, BOOL}},
{"ObdMultiplexingChanged", {CLEAR_ON_MANAGER_START | CLEAR_ON_ONROAD_TRANSITION, BOOL}},
{"ObdMultiplexingEnabled", {CLEAR_ON_MANAGER_START | CLEAR_ON_ONROAD_TRANSITION, BOOL}},
@@ -135,7 +135,7 @@ inline static std::unordered_map<std::string, ParamKeyAttributes> keys = {
{"ApiCache_DriveStats", {PERSISTENT, JSON}},
{"AutoLaneChangeBsmDelay", {PERSISTENT | BACKUP, BOOL, "0"}},
{"AutoLaneChangeTimer", {PERSISTENT | BACKUP, INT, "0"}},
{"BlinkerMinLateralControlSpeed", {PERSISTENT | BACKUP, INT, "20"}}, // MPH or km/h
{"BlinkerMinLateralControlSpeed",{PERSISTENT | BACKUP, INT, "20", R"({"min": 0, "max": 255, "step": 1, "title": "Blinker Min Lat Control Speed", "units": "mph", "description": "Minimum speed for lateral control during blinker usage."})"}},
{"BlinkerPauseLateralControl", {PERSISTENT | BACKUP, INT, "0"}},
{"Brightness", {PERSISTENT | BACKUP, INT, "0"}},
{"CarParamsSP", {CLEAR_ON_MANAGER_START | CLEAR_ON_ONROAD_TRANSITION, BYTES}},
+6
View File
@@ -42,6 +42,7 @@ cdef extern from "common/params.h":
bool checkKey(string) nogil
ParamKeyType getKeyType(string) nogil
optional[string] getKeyDefaultValue(string) nogil
optional[string] getKeyMetadata(string) nogil
string getParamPath(string) nogil
void clearAll(ParamKeyFlag)
vector[string] allKeys(ParamKeyFlag)
@@ -191,6 +192,11 @@ cdef class Params:
cdef optional[string] default = self.p.getKeyDefaultValue(k)
return self._cpp2python(t, default.value(), None, key) if default.has_value() else None
def get_key_metadata(self, key):
cdef string k = self.check_key(key)
cdef optional[string] metadata = self.p.getKeyMetadata(k)
return metadata.value().decode("utf-8") if metadata.has_value() else None
def cpp2python(self, key, value):
cdef string k = self.check_key(key)
cdef ParamKeyType t = self.p.getKeyType(k)
+29 -9
View File
@@ -13,6 +13,7 @@ import time
from jsonrpc import dispatcher
from functools import partial
from cereal import log
from openpilot.common.params import Params
from openpilot.common.realtime import set_core_affinity
from openpilot.common.swaglog import cloudlog
@@ -180,16 +181,39 @@ def getParamsAllKeys() -> list[str]:
@dispatcher.add_method
def getParamsAllKeysV1() -> dict[str, str]:
available_keys: list[str] = [k.decode('utf-8') for k in Params().all_keys()]
params = Params()
available_keys: list[str] = [k.decode('utf-8') for k in params.all_keys()]
params_dict: dict[str, list[dict[str, str | bool | int | None]]] = {"params": []}
for key in available_keys:
value = get_param_as_byte(key, get_default=True)
params_dict["params"].append({
metadata = None
metadata_json = params.get_key_metadata(key)
if metadata_json:
try:
metadata = json.loads(metadata_json)
if "enum" in metadata and log is not None:
enum_name = metadata["enum"]
if hasattr(log, enum_name):
enum_cls = getattr(log, enum_name)
if hasattr(enum_cls, "schema") and hasattr(enum_cls.schema, "enumerants"):
options = []
for name, val in enum_cls.schema.enumerants.items():
options.append({"value": val, "label": name})
# Sort by value
options.sort(key=lambda x: x["value"])
metadata["options"] = options
except json.JSONDecodeError:
pass
param_data = {
"key": key,
"type": int(params.get_type(key).value),
"default_value": base64.b64encode(value).decode('utf-8') if value else None,
})
"_extra": metadata,
}
params_dict["params"].append(param_data)
return {"keys": json.dumps(params_dict.get("params", []))}
@@ -238,10 +262,7 @@ def startLocalProxy(global_end_event: threading.Event, remote_ws_uri: str, local
cloudlog.debug("athena.startLocalProxy.starting")
ws = create_connection(
remote_ws_uri,
header={"Authorization": f"Bearer {sunnylink_api.get_token()}"},
enable_multithread=True,
sslopt={"cert_reqs": ssl.CERT_NONE}
remote_ws_uri, header={"Authorization": f"Bearer {sunnylink_api.get_token()}"}, enable_multithread=True, sslopt={"cert_reqs": ssl.CERT_NONE}
)
return start_local_proxy_shim(global_end_event, local_port, ws)
@@ -277,8 +298,7 @@ def main(exit_event: threading.Event = None):
sslopt={"cert_reqs": ssl.CERT_NONE if "localhost" in ws_uri else ssl.CERT_REQUIRED},
timeout=SUNNYLINK_RECONNECT_TIMEOUT_S,
)
cloudlog.event("sunnylinkd.main.connected_ws", ws_uri=ws_uri, retries=conn_retries,
duration=time.monotonic() - conn_start)
cloudlog.event("sunnylinkd.main.connected_ws", ws_uri=ws_uri, retries=conn_retries, duration=time.monotonic() - conn_start)
conn_start = None
conn_retries = 0
@@ -0,0 +1,123 @@
#!/usr/bin/env python3
import sys
import os
import json
import unittest
from unittest.mock import MagicMock, patch
# Add openpilot root to python path
sys.path.append(os.getcwd())
# Mock ALL dependencies before importing sunnylinkd
jsonrpc_mock = MagicMock()
def add_method(func):
return func
jsonrpc_mock.dispatcher.add_method = add_method
sys.modules["jsonrpc"] = jsonrpc_mock
sys.modules["jsonrpc.dispatcher"] = jsonrpc_mock.dispatcher
sys.modules["openpilot.common.params"] = MagicMock()
sys.modules["openpilot.common.realtime"] = MagicMock()
sys.modules["openpilot.common.swaglog"] = MagicMock()
sys.modules["openpilot.system.hardware.hw"] = MagicMock()
sys.modules["openpilot.system.athena.athenad"] = MagicMock()
sys.modules["websocket"] = MagicMock()
sys.modules["cereal"] = MagicMock()
sys.modules["cereal.messaging"] = MagicMock()
sys.modules["openpilot.sunnypilot.sunnylink.api"] = MagicMock()
sys.modules["openpilot.sunnypilot.sunnylink.utils"] = MagicMock()
# Now import sunnylinkd
from sunnypilot.sunnylink.athena import sunnylinkd
class TestParamsMetadata(unittest.TestCase):
@patch("sunnypilot.sunnylink.athena.sunnylinkd.Params")
@patch("sunnypilot.sunnylink.athena.sunnylinkd.get_param_as_byte")
def test_metadata_parsing(self, mock_get_param, mock_params):
# Setup mock Params
mock_params_instance = mock_params.return_value
mock_params_instance.all_keys.return_value = [b"LongitudinalPersonality", b"BlinkerMinLateralControlSpeed", b"DongleId"]
mock_params_instance.get_type.return_value.value = 1 # INT
# Setup mock get_key_metadata
def get_key_metadata_side_effect(key):
if key == "LongitudinalPersonality":
return json.dumps(
{"enum": "LongitudinalPersonality", "title": "Driving Personality", "description": "Adjusts the driving style of the openpilot longitudinal control."}
)
elif key == "BlinkerMinLateralControlSpeed":
return json.dumps(
{
"min": 0,
"max": 255,
"step": 1,
"title": "Blinker Min Lat Control Speed",
"units": "mph",
"description": "Minimum speed for lateral control during blinker usage.",
}
)
return None
mock_params_instance.get_key_metadata.side_effect = get_key_metadata_side_effect
# Setup mock get_param_as_byte
mock_get_param.return_value = b"1"
# Mock cereal.log
mock_log = MagicMock()
sunnylinkd.log = mock_log
# Setup LongitudinalPersonality enum mock
mock_enum = MagicMock()
mock_enum.schema.enumerants = {"Aggressive": 0, "Standard": 1, "Relaxed": 2}
mock_log.LongitudinalPersonality = mock_enum
# Call the function
response = sunnylinkd.getParamsAllKeysV1()
self.assertIn("keys", response)
params = json.loads(response["keys"])
# Check LongitudinalPersonality (Enum)
lp_param = next((p for p in params if p["key"] == "LongitudinalPersonality"), None)
self.assertIsNotNone(lp_param)
# Debug print
print(f"LongitudinalPersonality param: {lp_param}")
self.assertIn("_extra", lp_param)
extra = lp_param["_extra"]
self.assertIsNotNone(extra)
self.assertEqual(extra.get("title"), "Driving Personality")
self.assertEqual(extra.get("enum"), "LongitudinalPersonality")
# Verify options are populated from our mock
self.assertIn("options", extra)
options = extra["options"]
self.assertEqual(len(options), 3)
self.assertEqual(options[0]["label"], "Aggressive")
self.assertEqual(options[0]["value"], 0)
# Check BlinkerMinLateralControlSpeed (Numeric)
blinker_param = next((p for p in params if p["key"] == "BlinkerMinLateralControlSpeed"), None)
self.assertIsNotNone(blinker_param)
extra = blinker_param["_extra"]
self.assertIsNotNone(extra)
self.assertEqual(extra.get("title"), "Blinker Min Lat Control Speed")
self.assertEqual(extra.get("min"), 0)
self.assertEqual(extra.get("max"), 255)
self.assertEqual(extra.get("units"), "mph")
# Check a param without metadata
other_param = next((p for p in params if p["key"] == "DongleId"), None)
self.assertIsNotNone(other_param)
self.assertIsNone(other_param.get("_extra"))
if __name__ == "__main__":
unittest.main()