sunnypilot v2026.002.000

version: sunnypilot v2026.002.000 (staging)
date: 2026-07-21T07:44:02
master commit: 2c334ede443d7391d27575af8c854a095ba702a8
This commit is contained in:
github-actions[bot] committed 2026-07-21 07:44:02 +00:00
commit de77c89cf6
3995 files changed
+1260688

No files matched your search

+211
View File
@@ -0,0 +1,211 @@
#!/usr/bin/env python3
"""
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.
Scans all per-page YAML files in settings_ui_src/pages/ and replaces inlined
rule blocks that exactly match a macro definition in _macros.yaml with $ref
references.
Match policy:
- Whole-list match (exact): replace the entire list with [{$ref}].
- Single-rule match: any individual rule whose canonical form equals macro[0]
(when macro is exactly one rule) is replaced with {$ref}.
Re-run is idempotent (already-substituted $refs are skipped).
Usage:
python apply_macros.py [--src DIR] [--dry-run]
"""
from __future__ import annotations
import argparse
import json
import os
import sys
import yaml
DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DEFAULT_SRC = os.path.join(DIR, "settings_ui_src")
class _BlockDumper(yaml.SafeDumper):
pass
def _represent_dict(dumper, data):
return dumper.represent_mapping("tag:yaml.org,2002:map", data.items(), flow_style=False)
def _represent_list(dumper, data):
flow = all(not isinstance(x, (dict, list)) for x in data) and len(data) <= 8
return dumper.represent_sequence("tag:yaml.org,2002:seq", data, flow_style=flow)
_BlockDumper.add_representer(dict, _represent_dict)
_BlockDumper.add_representer(list, _represent_list)
def _dump_yaml(data) -> str:
return yaml.dump(data, Dumper=_BlockDumper, sort_keys=False, allow_unicode=True, width=120)
def _canon(node) -> str:
return json.dumps(node, sort_keys=True, separators=(",", ":"))
def _is_ref(node) -> bool:
return isinstance(node, dict) and len(node) == 1 and "$ref" in node
def _make_ref(name: str) -> dict:
return {"$ref": f"#/macros/{name}"}
def _substitute_rules(rules: list, whole_list: dict, single_rule: dict) -> list:
if not isinstance(rules, list) or not rules:
return rules
full_canon = _canon(rules)
if full_canon in whole_list:
return [_make_ref(whole_list[full_canon])]
out = []
for r in rules:
if _is_ref(r):
out.append(r)
continue
rc = _canon(r)
if rc in single_rule:
out.append(_make_ref(single_rule[rc]))
else:
out.append(r)
return out
def _walk_item(item: dict, whole_list: dict, single_rule: dict) -> bool:
changed = False
for ctx in ("visibility", "enablement"):
if ctx in item:
new = _substitute_rules(item[ctx], whole_list, single_rule)
if new != item[ctx]:
item[ctx] = new
changed = True
if "options" in item:
for opt in item["options"]:
if not isinstance(opt, dict):
continue
for ctx in ("visibility", "enablement"):
if ctx in opt:
new = _substitute_rules(opt[ctx], whole_list, single_rule)
if new != opt[ctx]:
opt[ctx] = new
changed = True
if "sub_items" in item:
for sub in item["sub_items"]:
if _walk_item(sub, whole_list, single_rule):
changed = True
return changed
def _walk_page(page: dict, whole_list: dict, single_rule: dict) -> bool:
changed = False
for sec in page.get("sections", []) or []:
for ctx in ("visibility", "enablement"):
if ctx in sec:
new = _substitute_rules(sec[ctx], whole_list, single_rule)
if new != sec[ctx]:
sec[ctx] = new
changed = True
for it in sec.get("items", []) or []:
if _walk_item(it, whole_list, single_rule):
changed = True
for sp in sec.get("sub_panels", []) or []:
for it in sp.get("items", []) or []:
if _walk_item(it, whole_list, single_rule):
changed = True
for it in page.get("items", []) or []:
if _walk_item(it, whole_list, single_rule):
changed = True
for sp in page.get("sub_panels", []) or []:
for it in sp.get("items", []) or []:
if _walk_item(it, whole_list, single_rule):
changed = True
return changed
def _build_macro_indices(macros: dict) -> tuple[dict, dict]:
whole_list = {}
single_rule = {}
for name, body in macros.items():
if isinstance(body, list):
whole_list[_canon(body)] = name
if len(body) == 1 and isinstance(body[0], dict):
single_rule[_canon(body[0])] = name
elif isinstance(body, dict):
single_rule[_canon(body)] = name
return whole_list, single_rule
def _read_yaml_with_header(path: str) -> tuple[list[str], dict]:
with open(path) as f:
text = f.read()
header_lines = []
for line in text.splitlines(keepends=True):
if line.startswith("#"):
header_lines.append(line)
elif line.strip() == "":
header_lines.append(line)
else:
break
doc = yaml.safe_load(text) or {}
return header_lines, doc
def _write_yaml_with_header(path: str, header: list[str], doc: dict) -> None:
with open(path, "w") as f:
if header:
# Keep at most one trailing blank line in header
while len(header) > 1 and header[-1].strip() == "" and header[-2].strip() == "":
header.pop()
for line in header:
f.write(line)
f.write(_dump_yaml(doc))
def apply(src_dir: str, dry_run: bool) -> int:
with open(os.path.join(src_dir, "_macros.yaml")) as f:
macros_doc = yaml.safe_load(f) or {}
macros = (macros_doc.get("macros") or {}) if isinstance(macros_doc, dict) else {}
whole_list, single_rule = _build_macro_indices(macros)
changed_files = 0
pages_dir = os.path.join(src_dir, "pages")
if os.path.isdir(pages_dir):
for fn in sorted(os.listdir(pages_dir)):
if not fn.endswith((".yaml", ".yml")) or fn.startswith("_"):
continue
path = os.path.join(pages_dir, fn)
header, doc = _read_yaml_with_header(path)
if _walk_page(doc, whole_list, single_rule):
changed_files += 1
if not dry_run:
_write_yaml_with_header(path, header, doc)
else:
print(f"would-rewrite: {path}")
print(f"{'Would rewrite' if dry_run else 'Rewrote'} {changed_files} files")
return 0
def _main() -> int:
parser = argparse.ArgumentParser(description="Substitute inlined rule blocks with $ref macros across pages/.")
parser.add_argument("--src", default=DEFAULT_SRC)
parser.add_argument("--dry-run", action="store_true")
args = parser.parse_args()
return apply(args.src, args.dry_run)
if __name__ == "__main__":
sys.exit(_main())
+300
View File
@@ -0,0 +1,300 @@
#!/usr/bin/env python3
"""
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.
Reads settings_ui_src/ (the dev-friendly authoring tree) and emits the
canonical settings_ui.json that the device generator + frontend consume.
Source layout:
_macros.yaml # named rule fragments
pages/<page_id>.yaml # one file per panel/page
pages/vehicle.yaml # special: emits vehicle_settings
Each page.yaml contains the full panel: metadata + sections + items + sub_panels
inline. Sub-panels are nested inside the section they belong to. Items appear
in the order written in the file.
Macro references use JSON-Schema-style $ref pointers:
enablement:
- {$ref: "#/macros/offroad"}
- {$ref: "#/macros/mads_full_platforms"}
Macros may reference other macros (max depth 3). Cycles raise an error.
Usage:
python compile_settings_ui.py [--src DIR] [--out PATH] [--check]
"""
from __future__ import annotations
import argparse
import copy
import json
import os
import sys
import yaml
DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DEFAULT_SRC = os.path.join(DIR, "settings_ui_src")
DEFAULT_OUT = os.path.join(DIR, "settings_ui.json")
SCHEMA_VERSION = "1.0"
MAX_MACRO_DEPTH = 3
class CompileError(Exception):
pass
def _load_yaml(path: str):
with open(path) as f:
return yaml.safe_load(f) or {}
def _is_ref(node) -> bool:
return isinstance(node, dict) and len(node) == 1 and "$ref" in node
def _ref_name(node: dict) -> str:
ref = node["$ref"]
if not isinstance(ref, str) or not ref.startswith("#/macros/"):
raise CompileError(f"unsupported $ref: {ref!r} (must start with '#/macros/')")
return ref[len("#/macros/"):]
def _resolve_refs(node, macros: dict, visiting: tuple[str, ...] = ()):
"""Resolve $ref nodes against macros. Recursive; depth-limited; cycle-safe.
- $ref in list-context: macro's list spliced into parent list.
- $ref in scalar-context (object value): macro's value substitutes.
- Non-list macro values may not be referenced from list contexts.
"""
if _is_ref(node):
name = _ref_name(node)
if name in visiting:
raise CompileError(f"$ref cycle: {' -> '.join(visiting + (name,))}")
if len(visiting) >= MAX_MACRO_DEPTH:
raise CompileError(f"$ref nesting exceeds depth {MAX_MACRO_DEPTH}: {' -> '.join(visiting + (name,))}")
if name not in macros:
raise CompileError(f"unknown macro: {name}")
return _resolve_refs(copy.deepcopy(macros[name]), macros, visiting + (name,))
if isinstance(node, dict):
return {k: _resolve_refs(v, macros, visiting) for k, v in node.items()}
if isinstance(node, list):
out = []
for item in node:
if _is_ref(item):
resolved = _resolve_refs(item, macros, visiting)
if not isinstance(resolved, list):
raise CompileError(
f"macro '{_ref_name(item)}' must resolve to a list when used in a list context"
)
out.extend(resolved)
else:
out.append(_resolve_refs(item, macros, visiting))
return out
return node
# Output JSON key order. Mirrors the conventions in the original hand-written
# settings_ui.json so structural diffs after extraction are minimal.
_ITEM_KEY_ORDER = [
"key",
"widget",
"needs_onroad_cycle",
"requires_attestation",
"blocked",
"title",
"description",
"details",
"title_param_suffix",
"min",
"max",
"step",
"unit",
"options",
"visibility",
"enablement",
"sub_items",
]
def _canon_item(item: dict, macros: dict) -> dict:
resolved = dict(item)
for ctx in ("visibility", "enablement"):
if ctx in resolved:
resolved[ctx] = _resolve_refs(resolved[ctx], macros)
if "options" in resolved:
new_opts = []
for opt in resolved["options"]:
if isinstance(opt, dict):
opt = dict(opt)
for ctx in ("visibility", "enablement"):
if ctx in opt:
opt[ctx] = _resolve_refs(opt[ctx], macros)
new_opts.append(opt)
resolved["options"] = new_opts
if "sub_items" in resolved:
resolved["sub_items"] = [_canon_item(s, macros) for s in resolved["sub_items"]]
out: dict = {}
for k in _ITEM_KEY_ORDER:
if k in resolved:
out[k] = resolved[k]
for k, v in resolved.items():
if k not in out:
out[k] = v
return out
def _canon_section(section: dict, macros: dict) -> dict:
out: dict = {"id": section["id"], "title": section["title"]}
if "description" in section:
out["description"] = section["description"]
for k in ("visibility", "enablement"):
if k in section:
out[k] = _resolve_refs(section[k], macros)
if "attestation_required" in section:
out["attestation_required"] = section["attestation_required"]
out["items"] = [_canon_item(i, macros) for i in section.get("items", [])]
if "sub_panels" in section and section["sub_panels"]:
out["sub_panels"] = [_canon_sub_panel(sp, macros) for sp in section["sub_panels"]]
for k, v in section.items():
if k not in out and k != "order":
out[k] = v
return out
def _canon_sub_panel(sp: dict, macros: dict) -> dict:
out: dict = {"id": sp["id"], "label": sp["label"], "trigger_key": sp["trigger_key"]}
if "trigger_condition" in sp:
out["trigger_condition"] = _resolve_refs(sp["trigger_condition"], macros)
out["items"] = [_canon_item(i, macros) for i in sp.get("items", [])]
for k, v in sp.items():
if k not in out:
out[k] = v
return out
def _canon_panel(page: dict, macros: dict) -> dict:
out: dict = {
"id": page["id"],
"label": page["label"],
"icon": page["icon"],
"order": page["order"],
}
if "remote_configurable" in page:
out["remote_configurable"] = page["remote_configurable"]
if "description" in page:
out["description"] = page["description"]
if "sections" in page and page["sections"]:
out["sections"] = [_canon_section(s, macros) for s in page["sections"]]
if "items" in page and page["items"]:
out["items"] = [_canon_item(i, macros) for i in page["items"]]
if "sub_panels" in page and page["sub_panels"]:
out["sub_panels"] = [_canon_sub_panel(sp, macros) for sp in page["sub_panels"]]
return out
def _canon_vehicle(page: dict, macros: dict) -> dict:
"""Convert page-shape vehicle.yaml to wire-format vehicle_settings dict."""
out: dict = {}
for sec in page.get("sections", []):
brand = sec["id"]
brand_out: dict = {"title": sec.get("title", "")}
if "description" in sec:
brand_out["description"] = sec["description"]
brand_out["items"] = [_canon_item(i, macros) for i in sec.get("items", [])]
out[brand] = brand_out
return out
def _load_pages(src: str) -> list[dict]:
pages_dir = os.path.join(src, "pages")
if not os.path.isdir(pages_dir):
return []
pages = []
for fn in sorted(os.listdir(pages_dir)):
if not fn.endswith((".yaml", ".yml")):
continue
if fn.startswith("_"):
continue
path = os.path.join(pages_dir, fn)
page = _load_yaml(path)
if not isinstance(page, dict):
raise CompileError(f"{path}: page YAML must be an object")
if "id" not in page:
raise CompileError(f"{path}: page missing 'id'")
page["__source"] = path
pages.append(page)
return pages
def compile_schema(src: str) -> dict:
macros_doc = _load_yaml(os.path.join(src, "_macros.yaml"))
macros = (macros_doc.get("macros") or {}) if isinstance(macros_doc, dict) else {}
pages = _load_pages(src)
panels_out = []
vehicle_out: dict = {}
# Order panels by `order` field (falling back to file position).
panel_pages = [p for p in pages if p.get("kind") != "vehicle"]
panel_pages.sort(key=lambda p: (p.get("order", 999), p["id"]))
for page in panel_pages:
panels_out.append(_canon_panel(page, macros))
for page in pages:
if page.get("kind") == "vehicle":
vehicle_out = _canon_vehicle(page, macros)
return {
"$schema": "./settings_ui.schema.json",
"schema_version": SCHEMA_VERSION,
"panels": panels_out,
"vehicle_settings": vehicle_out,
}
def _main() -> int:
parser = argparse.ArgumentParser(description="Compile settings_ui_src/ -> settings_ui.json")
parser.add_argument("--src", default=DEFAULT_SRC)
parser.add_argument("--out", default=DEFAULT_OUT)
parser.add_argument("--check", action="store_true",
help="Compile and diff against existing settings_ui.json; exit non-zero on diff.")
args = parser.parse_args()
schema = compile_schema(args.src)
rendered = json.dumps(schema, indent=2) + "\n"
if args.check:
if not os.path.exists(args.out):
print(f"--check: {args.out} does not exist", file=sys.stderr)
return 1
with open(args.out) as f:
current = f.read()
if current.strip() == rendered.strip():
print(f"--check: {args.out} matches compiled output")
return 0
print(f"--check: {args.out} differs from compiled output", file=sys.stderr)
cur_obj = json.loads(current)
if cur_obj == schema:
print("(structurally equal; only formatting differs)", file=sys.stderr)
return 1
with open(args.out, "w") as f:
f.write(rendered)
print(f"Wrote {args.out}")
return 0
if __name__ == "__main__":
sys.exit(_main())
+214
View File
@@ -0,0 +1,214 @@
#!/usr/bin/env python3
"""
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.
One-shot extractor: settings_ui.json -> settings_ui_src/pages/*.yaml.
Reads the existing monolithic settings_ui.json and produces one YAML file per
page (panel) plus pages/vehicle.yaml for the per-brand settings. Macros remain
unwritten by extract; populate _macros.yaml separately.
Usage:
python extract_settings_ui.py [--src DIR] [--definition PATH]
"""
from __future__ import annotations
import argparse
import json
import os
import shutil
import sys
import yaml
DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DEFAULT_DEFINITION = os.path.join(DIR, "settings_ui.json")
DEFAULT_SRC = os.path.join(DIR, "settings_ui_src")
class _BlockDumper(yaml.SafeDumper):
pass
def _represent_dict(dumper, data):
return dumper.represent_mapping("tag:yaml.org,2002:map", data.items(), flow_style=False)
def _represent_list(dumper, data):
flow = all(not isinstance(x, (dict, list)) for x in data) and len(data) <= 8
return dumper.represent_sequence("tag:yaml.org,2002:seq", data, flow_style=flow)
_BlockDumper.add_representer(dict, _represent_dict)
_BlockDumper.add_representer(list, _represent_list)
def _dump_yaml(data) -> str:
return yaml.dump(data, Dumper=_BlockDumper, sort_keys=False, allow_unicode=True, width=120)
# Item field order in per-page YAML (mirrors settings_ui.json conventions).
_ITEM_ORDER = [
"key",
"widget",
"needs_onroad_cycle",
"requires_attestation",
"blocked",
"title",
"description",
"details",
"title_param_suffix",
"min",
"max",
"step",
"unit",
"options",
"visibility",
"enablement",
"sub_items",
]
def _ordered_item(item: dict) -> dict:
out: dict = {}
for k in _ITEM_ORDER:
if k in item:
v = item[k]
if k == "sub_items":
v = [_ordered_item(s) for s in v]
out[k] = v
for k, v in item.items():
if k not in out:
out[k] = v
return out
def _ordered_section(section: dict) -> dict:
out: dict = {"id": section["id"], "title": section["title"]}
for f in ("description", "order", "visibility", "enablement", "attestation_required"):
if f in section:
out[f] = section[f]
if "items" in section:
out["items"] = [_ordered_item(i) for i in section["items"]]
if "sub_panels" in section and section["sub_panels"]:
out["sub_panels"] = [_ordered_sub_panel(sp) for sp in section["sub_panels"]]
for k, v in section.items():
if k not in out:
out[k] = v
return out
def _ordered_sub_panel(sp: dict) -> dict:
out: dict = {"id": sp["id"], "label": sp["label"], "trigger_key": sp["trigger_key"]}
if "trigger_condition" in sp:
out["trigger_condition"] = sp["trigger_condition"]
if "items" in sp:
out["items"] = [_ordered_item(i) for i in sp["items"]]
for k, v in sp.items():
if k not in out:
out[k] = v
return out
def _ordered_page(panel: dict) -> dict:
out: dict = {
"id": panel["id"],
"label": panel["label"],
"icon": panel["icon"],
"order": panel["order"],
}
for f in ("remote_configurable", "description"):
if f in panel:
out[f] = panel[f]
if "sections" in panel and panel["sections"]:
out["sections"] = [_ordered_section(s) for s in panel["sections"]]
if "items" in panel and panel["items"]:
out["items"] = [_ordered_item(i) for i in panel["items"]]
if "sub_panels" in panel and panel["sub_panels"]:
out["sub_panels"] = [_ordered_sub_panel(sp) for sp in panel["sub_panels"]]
for k, v in panel.items():
if k not in out:
out[k] = v
return out
def _ordered_vehicle_page(vehicle_settings: dict) -> dict:
"""Convert wire-format vehicle_settings dict into a page-shape YAML.
Brand becomes a section id. Each brand's items become section items.
"""
sections = []
for brand in sorted(vehicle_settings.keys()):
bd = vehicle_settings[brand]
items = bd.get("items", []) if isinstance(bd, dict) else bd
sec: dict = {
"id": brand,
"title": bd.get("title", "") if isinstance(bd, dict) else "",
}
if isinstance(bd, dict) and "description" in bd:
sec["description"] = bd["description"]
sec["items"] = [_ordered_item(i) for i in items]
sections.append(sec)
return {
"id": "vehicle",
"label": "Vehicle",
"icon": "vehicle",
"order": 99,
"kind": "vehicle", # signals to compiler: emit as vehicle_settings, not panels
"sections": sections,
}
def extract(definition_path: str, src_dir: str) -> None:
with open(definition_path) as f:
data = json.load(f)
pages_dir = os.path.join(src_dir, "pages")
if os.path.isdir(pages_dir):
shutil.rmtree(pages_dir)
os.makedirs(pages_dir)
count = 0
for panel in data.get("panels", []):
page = _ordered_page(panel)
path = os.path.join(pages_dir, f"{panel['id']}.yaml")
with open(path, "w") as f:
f.write(f"# Page: {panel['id']}\n")
f.write("# Edit this file. Run compile_settings_ui.py to emit settings_ui.json.\n")
f.write(_dump_yaml(page))
count += 1
vehicle_settings = data.get("vehicle_settings", {})
if vehicle_settings:
vehicle_page = _ordered_vehicle_page(vehicle_settings)
path = os.path.join(pages_dir, "vehicle.yaml")
with open(path, "w") as f:
f.write("# Page: vehicle (per-brand settings)\n")
f.write("# Compiles to settings_ui.json#vehicle_settings (brand = section id).\n")
f.write(_dump_yaml(vehicle_page))
count += 1
# Ensure _macros.yaml exists
macros_path = os.path.join(src_dir, "_macros.yaml")
if not os.path.exists(macros_path):
with open(macros_path, "w") as f:
f.write("# Named rule fragments. Reference from pages via {$ref: \"#/macros/<name>\"}.\n")
f.write(_dump_yaml({"macros": {}}))
print(f"Extracted: {count} pages -> {pages_dir}")
def _main() -> int:
parser = argparse.ArgumentParser(description="Extract settings_ui.json into per-page YAML files.")
parser.add_argument("--definition", default=DEFAULT_DEFINITION, help="path to settings_ui.json")
parser.add_argument("--src", default=DEFAULT_SRC, help="path to settings_ui_src/ output dir")
args = parser.parse_args()
extract(args.definition, args.src)
return 0
if __name__ == "__main__":
sys.exit(_main())
@@ -0,0 +1,194 @@
#!/usr/bin/env python3
"""
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 __future__ import annotations
import base64
import datetime
import gzip
import json
import os
from collections.abc import Callable
from openpilot.sunnypilot.sunnylink.capabilities import CAPABILITY_FIELDS, CAPABILITY_LABELS
SCHEMA_VERSION = "1.0"
_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DEFINITION_PATH = os.path.join(_DIR, "settings_ui.json")
TORQUE_VERSIONS_PATH = os.path.normpath(
os.path.join(_DIR, "..", "selfdrive", "controls", "lib", "latcontrol_torque_versions.json")
)
def _load_torque_versions() -> dict:
"""Load latcontrol_torque_versions.json so TorqueControlTune options stay in sync."""
try:
with open(TORQUE_VERSIONS_PATH) as f:
return json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
return {}
def _build_torque_options(versions: dict) -> list[dict]:
options: list[dict] = [{"value": "", "label": "Default"}]
parsed: list[tuple[float, str]] = []
for label, info in versions.items():
try:
parsed.append((float(info["version"]), label))
except (KeyError, TypeError, ValueError):
continue
for version, label in sorted(parsed, key=lambda kv: kv[0], reverse=True):
options.append({"value": version, "label": label})
return options
def _inject_dynamic_options(schema: dict) -> None:
versions = _load_torque_versions()
if not versions:
return
options = _build_torque_options(versions)
def visitor(item: dict) -> None:
if item.get("key") == "TorqueControlTune":
item["options"] = options
_walk_all_items(schema, visitor)
def _load_definition() -> dict:
"""Load settings_ui.json and inject dynamic options sourced from runtime data files."""
with open(DEFINITION_PATH) as f:
schema = json.load(f)
_inject_dynamic_options(schema)
return schema
# Public API
def generate_schema() -> dict:
"""Return the settings_ui.json content augmented with runtime metadata.
Adds three top-level fields the frontend consumes:
- generated_at: ISO timestamp (drives schema-cache freshness checks)
- capability_fields: declared CAPABILITY_FIELDS, used for rule validation
- capability_labels: human-readable labels for capability_fields
"""
schema = _load_definition()
schema["generated_at"] = datetime.datetime.now(datetime.UTC).isoformat()
schema["capability_fields"] = list(CAPABILITY_FIELDS)
schema["capability_labels"] = dict(CAPABILITY_LABELS)
return schema
def generate_schema_json() -> str:
"""Generate SettingsSchema as a compact JSON string."""
return json.dumps(generate_schema(), separators=(",", ":"))
def generate_schema_compressed() -> str:
"""Generate SettingsSchema as gzip-compressed, base64-encoded string.
Compression pipeline:
1. JSON serialize (compact, no whitespace)
2. UTF-8 encode
3. gzip compress
4. base64 encode
"""
raw = json.dumps(generate_schema(), separators=(",", ":")).encode("utf-8")
return base64.b64encode(gzip.compress(raw)).decode("utf-8")
# Schema introspection utilities
def _walk_rules(rules: list[dict] | None, visitor: Callable[[dict], None]) -> None:
"""Recursively walk all rules, calling visitor on each leaf rule."""
if not rules:
return
for rule in rules:
visitor(rule)
if rule.get("type") == "not" and "condition" in rule:
_walk_rules([rule["condition"]], visitor)
elif rule.get("type") in ("any", "all") and "conditions" in rule:
_walk_rules(rule["conditions"], visitor)
def _walk_all_items(schema: dict, visitor: Callable[[dict], None]) -> None:
"""Walk every item in the schema (panels, sections, sub_panels, sub_items, vehicle_settings)."""
def _visit_item(item: dict) -> None:
visitor(item)
for sub in item.get("sub_items", []):
_visit_item(sub)
for panel in schema.get("panels", []):
# Walk section items (V2)
for section in panel.get("sections", []):
for item in section.get("items", []):
_visit_item(item)
for sp in section.get("sub_panels", []):
for item in sp.get("items", []):
_visit_item(item)
# Walk flat items (V1)
for item in panel.get("items", []):
_visit_item(item)
for sp in panel.get("sub_panels", []):
for item in sp.get("items", []):
_visit_item(item)
for brand_data in schema.get("vehicle_settings", {}).values():
items = brand_data.get("items", []) if isinstance(brand_data, dict) else brand_data
for item in items:
_visit_item(item)
def collect_all_keys(schema: dict) -> set[str]:
"""Collect all param keys referenced in the schema (items + rules)."""
keys: set[str] = set()
def _visit_rule(rule: dict) -> None:
if rule.get("type") in ("param", "param_compare") and "key" in rule:
keys.add(rule["key"])
def _visit_item(item: dict) -> None:
if "key" in item:
keys.add(item["key"])
_walk_rules(item.get("visibility"), _visit_rule)
_walk_rules(item.get("enablement"), _visit_rule)
_walk_all_items(schema, _visit_item)
return keys
def collect_capability_refs(schema: dict) -> set[str]:
"""Collect all capability field names referenced in rules."""
refs: set[str] = set()
def _visit_rule(rule: dict) -> None:
if rule.get("type") == "capability" and "field" in rule:
refs.add(rule["field"])
def _visit_item(item: dict) -> None:
_walk_rules(item.get("visibility"), _visit_rule)
_walk_rules(item.get("enablement"), _visit_rule)
_walk_all_items(schema, _visit_item)
return refs
if __name__ == "__main__":
# CLI: print schema for inspection
schema = generate_schema()
print(json.dumps(schema, indent=2))
print(f"\nTotal panels: {len(schema.get('panels', []))}")
print(f"Total vehicle brands: {len(schema.get('vehicle_settings', {}))}")
keys = collect_all_keys(schema)
print(f"Total unique param keys: {len(keys)}")
# Show compression stats
raw_json = json.dumps(schema, separators=(",", ":")).encode("utf-8")
compressed = gzip.compress(raw_json)
print(f"\nRaw JSON size: {len(raw_json):,} bytes")
print(f"Compressed size: {len(compressed):,} bytes")
print(f"Compression ratio: {len(compressed)/len(raw_json):.1%}")
@@ -0,0 +1,526 @@
#!/usr/bin/env python3
"""
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.
Validates settings_ui.json against structural, semantic, and referential integrity constraints.
Usage:
python validate_settings_ui.py
python validate_settings_ui.py /path/to/settings_ui.json
"""
from __future__ import annotations
import json
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', '..'))
from openpilot.sunnypilot.sunnylink.capabilities import CAPABILITY_FIELDS
VALID_WIDGETS = {"toggle", "option", "multiple_button", "button", "info"}
VALID_COMPARE_OPS = {">", "<", ">=", "<="}
DEFAULT_PATH = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"settings_ui.json",
)
class ValidationResult:
"""Tracks pass/fail for each named check."""
def __init__(self) -> None:
self.passed: list[str] = []
self.failed: list[tuple[str, str]] = []
self.warnings: list[str] = []
def ok(self, name: str) -> None:
self.passed.append(name)
print(f"OK: {name}")
def error(self, name: str, details: str) -> None:
self.failed.append((name, details))
print(f"ERROR: {name}: {details}")
def warn(self, msg: str) -> None:
self.warnings.append(msg)
print(f"WARNING: {msg}")
@property
def success(self) -> bool:
return len(self.failed) == 0
def summary(self) -> None:
total_passed = len(self.passed)
total_failed = len(self.failed)
print(f"\n{'='*60}")
print(f"Summary: {total_passed} checks passed, {total_failed} checks failed")
if self.warnings:
print(f" {len(self.warnings)} warnings")
if self.success:
print("Result: PASS")
else:
print("Result: FAIL")
def validate_rule(rule: dict, path: str, result: ValidationResult,
capability_fields: tuple[str, ...]) -> bool:
"""Validate a single rule dict. Returns True if valid."""
if not isinstance(rule, dict):
result.error("rule well-formedness", f"{path}: rule is not a dict: {rule!r}")
return False
rule_type = rule.get("type")
if rule_type is None:
result.error("rule well-formedness", f"{path}: rule missing 'type' field")
return False
if rule_type in ("offroad_only", "not_engaged"):
# Only type required
return True
if rule_type == "capability":
valid = True
if "field" not in rule or not isinstance(rule["field"], str):
result.error("rule well-formedness", f"{path}: capability rule missing/invalid 'field'")
valid = False
if "equals" not in rule:
result.error("rule well-formedness", f"{path}: capability rule missing 'equals'")
valid = False
return valid
if rule_type == "param":
valid = True
if "key" not in rule or not isinstance(rule["key"], str):
result.error("rule well-formedness", f"{path}: param rule missing/invalid 'key'")
valid = False
if "equals" not in rule:
result.error("rule well-formedness", f"{path}: param rule missing 'equals'")
valid = False
return valid
if rule_type == "param_compare":
valid = True
if "key" not in rule or not isinstance(rule["key"], str):
result.error("rule well-formedness", f"{path}: param_compare rule missing/invalid 'key'")
valid = False
if "op" not in rule or rule["op"] not in VALID_COMPARE_OPS:
result.error("rule well-formedness",
f"{path}: param_compare rule missing/invalid 'op' (must be one of {VALID_COMPARE_OPS})")
valid = False
if "value" not in rule or not isinstance(rule["value"], (int, float)):
result.error("rule well-formedness", f"{path}: param_compare rule missing/invalid 'value' (must be number)")
valid = False
return valid
if rule_type == "not":
if "condition" not in rule or not isinstance(rule["condition"], dict):
result.error("rule well-formedness", f"{path}: 'not' rule missing/invalid 'condition'")
return False
return validate_rule(rule["condition"], f"{path}.not", result, capability_fields)
if rule_type in ("any", "all"):
if "conditions" not in rule or not isinstance(rule["conditions"], list):
result.error("rule well-formedness", f"{path}: '{rule_type}' rule missing/invalid 'conditions' array")
return False
valid = True
for i, cond in enumerate(rule["conditions"]):
if not validate_rule(cond, f"{path}.{rule_type}[{i}]", result, capability_fields):
valid = False
return valid
result.error("rule well-formedness", f"{path}: unknown rule type '{rule_type}'")
return False
def collect_rules_from_item(item: dict) -> list[tuple[str, list[dict]]]:
"""Return list of (context, rules_list) for an item's visibility + enablement."""
result = []
key = item.get("key", "?")
if "visibility" in item:
result.append((f"item '{key}' visibility", item["visibility"]))
if "enablement" in item:
result.append((f"item '{key}' enablement", item["enablement"]))
return result
def walk_rules_flat(rules: list[dict]) -> list[dict]:
"""Flatten all rules recursively into a single list."""
flat: list[dict] = []
for rule in rules:
flat.append(rule)
if rule.get("type") == "not" and "condition" in rule:
flat.extend(walk_rules_flat([rule["condition"]]))
elif rule.get("type") in ("any", "all") and "conditions" in rule:
flat.extend(walk_rules_flat(rule["conditions"]))
return flat
def collect_all_items(data: dict) -> list[tuple[str, dict]]:
"""Collect all items with their location path from the schema.
Returns (path, item_dict) tuples. Traverses sections, sub_panels, sub_items,
and vehicle_settings.
"""
items: list[tuple[str, dict]] = []
for panel in data.get("panels", []):
pid = panel.get("id", "?")
# Flat items on panel
for item in panel.get("items", []):
_collect_item(f"panel '{pid}'", item, items)
# Sections
for section in panel.get("sections", []):
sid = section.get("id", "?")
for item in section.get("items", []):
_collect_item(f"panel '{pid}' > section '{sid}'", item, items)
for sp in section.get("sub_panels", []):
spid = sp.get("id", "?")
for item in sp.get("items", []):
_collect_item(f"panel '{pid}' > section '{sid}' > sub_panel '{spid}'", item, items)
# Top-level sub_panels on panel (no section)
for sp in panel.get("sub_panels", []):
spid = sp.get("id", "?")
for item in sp.get("items", []):
_collect_item(f"panel '{pid}' > sub_panel '{spid}'", item, items)
# Vehicle settings (supports both flat list and { title, items } structure)
for brand, brand_data in data.get("vehicle_settings", {}).items():
brand_items = brand_data.get("items", []) if isinstance(brand_data, dict) else brand_data
for item in brand_items:
_collect_item(f"vehicle_settings '{brand}'", item, items)
return items
def _collect_item(path: str, item: dict, items: list[tuple[str, dict]]) -> None:
"""Recursively collect an item and its sub_items."""
items.append((path, item))
for sub in item.get("sub_items", []):
_collect_item(f"{path} > sub_item", sub, items)
def collect_panel_keys(panel: dict) -> set[str]:
"""Collect all item keys within a single panel (sections, sub_panels, sub_items)."""
keys: set[str] = set()
def _add(item: dict) -> None:
if "key" in item:
keys.add(item["key"])
for sub in item.get("sub_items", []):
_add(sub)
for item in panel.get("items", []):
_add(item)
for section in panel.get("sections", []):
for item in section.get("items", []):
_add(item)
for sp in section.get("sub_panels", []):
for item in sp.get("items", []):
_add(item)
for sp in panel.get("sub_panels", []):
for item in sp.get("items", []):
_add(item)
return keys
def check_json_parseable(path: str, result: ValidationResult) -> dict | None:
"""Check 1: JSON parseable."""
try:
with open(path) as f:
data = json.load(f)
result.ok("JSON parseable")
return data
except json.JSONDecodeError as e:
result.error("JSON parseable", str(e))
return None
except FileNotFoundError:
result.error("JSON parseable", f"file not found: {path}")
return None
def check_structural(data: dict, result: ValidationResult) -> None:
"""Check 2: Required fields on panels, sections, items, sub_panels."""
errors: list[str] = []
for i, panel in enumerate(data.get("panels", [])):
for field in ("id", "label", "icon", "order"):
if field not in panel:
errors.append(f"panels[{i}]: missing required field '{field}'")
for j, section in enumerate(panel.get("sections", [])):
for field in ("id", "title"):
if field not in section:
errors.append(f"panels[{i}].sections[{j}]: missing required field '{field}'")
for k, sp in enumerate(section.get("sub_panels", [])):
for field in ("id", "label", "trigger_key"):
if field not in sp:
errors.append(f"panels[{i}].sections[{j}].sub_panels[{k}]: missing required field '{field}'")
for k, sp in enumerate(panel.get("sub_panels", [])):
for field in ("id", "label", "trigger_key"):
if field not in sp:
errors.append(f"panels[{i}].sub_panels[{k}]: missing required field '{field}'")
# Validate items
all_items = collect_all_items(data)
for path, item in all_items:
if "key" not in item:
errors.append(f"{path}: item missing required field 'key'")
if "widget" not in item:
errors.append(f"{path}: item missing required field 'widget'")
elif item["widget"] not in VALID_WIDGETS:
errors.append(
f"{path}: item '{item.get('key', '?')}' has invalid widget '{item['widget']}'"
+ f" (must be one of {VALID_WIDGETS})"
)
if errors:
result.error("structural", "; ".join(errors))
else:
result.ok("structural")
def check_item_completeness(data: dict, result: ValidationResult) -> None:
"""Check 3: All items have required metadata (title, options for dropdowns)."""
all_items = collect_all_items(data)
issues: list[str] = []
for _path, item in all_items:
key = item.get("key", "unknown")
if "title" not in item:
issues.append(f"{key}: missing 'title'")
elif item["title"] == key:
issues.append(f"{key}: title must not equal key (use a human-readable title)")
widget = item.get("widget")
if widget in ("multiple_button", "option") and "options" in item:
opts = item["options"]
if not isinstance(opts, list):
issues.append(f"{key}: options must be a list")
else:
for opt in opts:
if not isinstance(opt, dict) or "value" not in opt or "label" not in opt:
issues.append(f"{key}: each option must have 'value' and 'label'")
break
if issues:
for issue in issues:
result.error("item completeness", issue)
else:
result.ok("item completeness")
def check_no_duplicate_keys(data: dict, result: ValidationResult) -> None:
"""Check 4: No param key appears in more than one panel."""
panel_keys: dict[str, list[str]] = {} # key -> list of panel ids
for panel in data.get("panels", []):
pid = panel.get("id", "?")
keys = collect_panel_keys(panel)
for key in keys:
panel_keys.setdefault(key, []).append(pid)
# Also check vehicle_settings keys don't collide with panel keys
for brand, brand_data in data.get("vehicle_settings", {}).items():
brand_items = brand_data.get("items", []) if isinstance(brand_data, dict) else brand_data
for item in brand_items:
key = item.get("key")
if key:
panel_keys.setdefault(key, []).append(f"vehicle_settings.{brand}")
duplicates = {k: v for k, v in panel_keys.items() if len(v) > 1}
if duplicates:
details = "; ".join(f"'{k}' in [{', '.join(v)}]" for k, v in duplicates.items())
result.error("no duplicate keys", details)
else:
result.ok("no duplicate keys")
def check_rule_wellformedness(data: dict, result: ValidationResult) -> None:
"""Check 5: All rules have valid structure."""
all_items = collect_all_items(data)
# Save current error count to detect new errors
error_count_before = len(result.failed)
for path, item in all_items:
for ctx, rules in collect_rules_from_item(item):
for i, rule in enumerate(rules):
validate_rule(rule, f"{path} > {ctx}[{i}]", result, CAPABILITY_FIELDS)
# Also validate trigger_condition rules on sub_panels
for panel in data.get("panels", []):
pid = panel.get("id", "?")
for section in panel.get("sections", []):
for sp in section.get("sub_panels", []):
if "trigger_condition" in sp:
validate_rule(sp["trigger_condition"], f"panel '{pid}' > sub_panel '{sp.get('id', '?')}' trigger_condition",
result, CAPABILITY_FIELDS)
if len(result.failed) == error_count_before:
result.ok("rule well-formedness")
def check_capability_refs(data: dict, result: ValidationResult) -> None:
"""Check 6: All capability rule field values are in CAPABILITY_FIELDS."""
all_items = collect_all_items(data)
invalid_refs: list[str] = []
cap_set = set(CAPABILITY_FIELDS)
for _path, item in all_items:
for _ctx, rules in collect_rules_from_item(item):
for rule in walk_rules_flat(rules):
if rule.get("type") == "capability":
field = rule.get("field")
if field and field not in cap_set:
invalid_refs.append(f"'{field}' in item '{item.get('key', '?')}'")
# Also check trigger_conditions
for panel in data.get("panels", []):
for section in panel.get("sections", []):
for sp in section.get("sub_panels", []):
if "trigger_condition" in sp:
for rule in walk_rules_flat([sp["trigger_condition"]]):
if rule.get("type") == "capability":
field = rule.get("field")
if field and field not in cap_set:
invalid_refs.append(f"'{field}' in sub_panel '{sp.get('id', '?')}' trigger_condition")
if invalid_refs:
result.error("capability refs", f"unknown capability fields: {', '.join(invalid_refs)}")
else:
result.ok("capability refs")
def check_no_self_reference(data: dict, result: ValidationResult) -> None:
"""Check 7: Item's rules must not reference the item's own key."""
all_items = collect_all_items(data)
self_refs: list[str] = []
for path, item in all_items:
key = item.get("key")
if not key:
continue
for _ctx, rules in collect_rules_from_item(item):
for rule in walk_rules_flat(rules):
if rule.get("type") in ("param", "param_compare") and rule.get("key") == key:
self_refs.append(f"'{key}' at {path}")
if self_refs:
result.error("no self-reference", f"items reference their own key: {', '.join(self_refs)}")
else:
result.ok("no self-reference")
def check_sub_panel_triggers(data: dict, result: ValidationResult) -> None:
"""Check 8: Sub-panel trigger_key must reference a key in the same panel."""
errors: list[str] = []
for panel in data.get("panels", []):
pid = panel.get("id", "?")
panel_keys = collect_panel_keys(panel)
# Check sub_panels at section level
for section in panel.get("sections", []):
for sp in section.get("sub_panels", []):
trigger = sp.get("trigger_key")
if trigger and trigger not in panel_keys:
errors.append(
f"sub_panel '{sp.get('id', '?')}' trigger_key '{trigger}'"
+ f" not found in panel '{pid}'"
)
# Check top-level sub_panels
for sp in panel.get("sub_panels", []):
trigger = sp.get("trigger_key")
if trigger and trigger not in panel_keys:
errors.append(
f"sub_panel '{sp.get('id', '?')}' trigger_key '{trigger}'"
+ f" not found in panel '{pid}'"
)
if errors:
result.error("sub-panel triggers", "; ".join(errors))
else:
result.ok("sub-panel triggers")
def check_ordering(data: dict, result: ValidationResult) -> None:
"""Check 9: Panel order values must be unique."""
orders: dict[int, list[str]] = {}
for panel in data.get("panels", []):
order = panel.get("order")
if order is not None:
orders.setdefault(order, []).append(panel.get("id", "?"))
duplicates = {o: ids for o, ids in orders.items() if len(ids) > 1}
if duplicates:
details = "; ".join(f"order {o}: [{', '.join(ids)}]" for o, ids in duplicates.items())
result.error("ordering", f"duplicate order values: {details}")
else:
result.ok("ordering")
def check_vehicle_brands(data: dict, result: ValidationResult) -> None:
"""Check 10: Vehicle settings keys lowercase + each brand has consistent {title, description, items} shape."""
vehicle = data.get("vehicle_settings", {})
errors: list[str] = []
for brand, brand_data in vehicle.items():
if not isinstance(brand, str) or brand != brand.lower():
errors.append(f"non-lowercase brand key: '{brand}'")
if not isinstance(brand_data, dict):
errors.append(f"brand '{brand}': expected object with {{title, description, items}}, got bare list")
continue
if "items" not in brand_data:
errors.append(f"brand '{brand}': missing 'items'")
if "title" not in brand_data:
errors.append(f"brand '{brand}': missing 'title' (use empty string for none)")
if errors:
result.error("vehicle brands", "; ".join(errors))
else:
result.ok("vehicle brands")
def validate(path: str) -> bool:
"""Run all validation checks on the given settings_ui.json file.
Returns True if all checks pass.
"""
result = ValidationResult()
# Check 1: JSON parseable
data = check_json_parseable(path, result)
if data is None:
result.summary()
return False
# Checks 2-10
check_structural(data, result)
check_item_completeness(data, result)
check_no_duplicate_keys(data, result)
check_rule_wellformedness(data, result)
check_capability_refs(data, result)
check_no_self_reference(data, result)
check_sub_panel_triggers(data, result)
check_ordering(data, result)
check_vehicle_brands(data, result)
result.summary()
return result.success
if __name__ == "__main__":
target = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_PATH
success = validate(target)
sys.exit(0 if success else 1)