"""Repository-backed cabin mode profiles.""" from __future__ import annotations import hashlib import json import re from collections.abc import Callable from dataclasses import dataclass from pathlib import Path from threading import RLock from typing import Any import yaml from cognitive_agent.application.mode_schedule import ModeSchedule _MATCH_FORMATTING = re.compile(r"[\s（）()]+") _NON_ASCII_SLUG = re.compile(r"[^a-z0-9]+") _MODE_ID = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$") _MODE_REFERENCE_VERBS = ( "切换进入", "切换到", "切换为", "帮我进入", "开始执行", "进入", "启动", "开启", "打开", "启用", "执行", "运行", "调用", "应用", "使用", "恢复", "切换", "选择", "退出", "停止", "关闭", "取消", "结束", "来一个", "来个", ) _MODE_REFERENCE_TAILS = ( "吧", "呀", "啊", "哦", "呢", "一下", "看看", "试试", "，", ",", "。", ".", "！", "!", "？", "?", "；", ";", "：", ":", "然后", "接着", "同时", "并且", ) def _normalize_match_text(value: str) -> str: # Parentheses in display names are presentation-only. For example, # ``女王副驾（行驶）`` is naturally spoken as ``女王副驾行驶模式``. # Removing those separators lets the complete name match while the # short-alias boundary guard below still rejects device compounds. return _MATCH_FORMATTING.sub("", value.casefold()) def _mode_candidate_matches(utterance: str, candidate: object) -> bool: """Match a mode reference without treating device nouns as mode aliases. Names that explicitly contain ``模式`` are unambiguous enough for normal substring matching. Short aliases are different: a user-created ``阅读模式`` automatically has the alias ``阅读``, but ``打开阅读灯`` is a light command, not a request to run that mode. A short alias therefore requires an invocation verb and a bounded tail, or an explicit ``模式`` suffix. This retains natural requests such as ``启动阅读`` and ``进入阅读模式`` while failing closed for compound device names. """ needle = _normalize_match_text(str(candidate).strip()) if not needle: return False if "模式" in needle: return needle in utterance start = 0 while True: index = utterance.find(needle, start) if index < 0: return False tail = utterance[index + len(needle) :] if tail.startswith("模式"): return True prefix = utterance[:index] tail_is_bounded = not tail or any( tail.startswith(boundary) for boundary in _MODE_REFERENCE_TAILS ) if any( prefix.endswith(verb) for verb in _MODE_REFERENCE_VERBS ) and tail_is_bounded: return True start = index + 1 def _clean_name(value: str) -> str: cleaned = value.strip().strip("，。,.!！") if not cleaned: raise ValueError("mode name must not be empty") if len(cleaned) > 64: raise ValueError("mode name must be at most 64 characters") return cleaned def _clean_mode_id(value: str) -> str: cleaned = str(value).strip() if not _MODE_ID.fullmatch(cleaned): raise ValueError( "mode id must start with a lowercase letter or number and contain " "only lowercase letters, numbers, underscores, or hyphens" ) return cleaned def _clean_instruction(value: str) -> str: cleaned = str(value).strip() if len(cleaned) > 1000: raise ValueError("mode instruction must be at most 1000 characters") return cleaned def _clean_command_flow(value: Any) -> dict[str, Any] | None: if value is None: return None if not isinstance(value, dict): raise ValueError("mode command_flow must be an object") try: encoded = json.dumps(value, ensure_ascii=False, sort_keys=True) except (TypeError, ValueError) as exc: raise ValueError("mode command_flow must be JSON serializable") from exc if len(encoded.encode("utf-8")) > 1024 * 1024: raise ValueError("mode command_flow must be at most 1 MiB") return json.loads(encoded) def _mode_aliases(name: str, aliases: list[str] | None = None) -> list[str]: values = [name, *(aliases or [])] if name.endswith("模式") and len(name) > 2: values.append(name.removesuffix("模式")) else: values.append(f"{name}模式") normalized: set[str] = set() unique = [] for value in values: candidate = str(value).strip() key = _normalize_match_text(candidate) if not candidate or key in normalized: continue normalized.add(key) unique.append(candidate) return unique def _slugify_mode_id(name: str) -> str: ascii_slug = _NON_ASCII_SLUG.sub("-", name.casefold()).strip("-") digest = hashlib.sha256(name.encode("utf-8")).hexdigest()[:10] if ascii_slug: return ascii_slug if len(ascii_slug) >= 3 else f"{ascii_slug}-{digest[:6]}" return f"mode-{digest}" @dataclass(frozen=True) class ModeOperation: domain: str capability: str target_scope: str parameters: dict[str, Any] operation: str = "set" reason: str = "" @classmethod def from_dict(cls, document: dict[str, Any]) -> "ModeOperation": parameters = document.get("parameters", {}) if not isinstance(parameters, dict): raise ValueError("mode operation parameters must be an object") return cls( domain=str(document["domain"]), capability=str(document["capability"]), target_scope=str(document["target_scope"]), operation=str(document.get("operation", "set")), parameters=dict(parameters), reason=str(document.get("reason", "")), ) def to_request(self, mode_name: str) -> dict[str, Any]: target_scope = self.target_scope if self.domain == "seat" and target_scope == "front_right": # Compiled seat operations use the physical zone name, while the # model-plan adapter exposes the logical passenger target. Normalize # persisted cards so a validated target is not queried again. target_scope = "passenger" return { "domain": self.domain, "capability": self.capability, "target_scope": target_scope, "operation": self.operation, "parameters": [ {"name": name, "value": value} for name, value in self.parameters.items() ], "reason": self.reason or f"Matched mode profile {mode_name}.", } def to_yaml(self) -> dict[str, Any]: return { "domain": self.domain, "capability": self.capability, "target_scope": self.target_scope, "operation": self.operation, "parameters": dict(self.parameters), "reason": self.reason, } def public_dict(self) -> dict[str, Any]: """Return the executable fields that are safe to expose as UI metadata.""" return { "domain": self.domain, "capability": self.capability, "target_scope": self.target_scope, "operation": self.operation, "parameters": dict(self.parameters), } @dataclass(frozen=True) class ModeTimelinePhase: id: str name: str duration_ms: int | None transition: str targets: dict[str, Any] effects: dict[str, Any] completion: str timeout_ms: int | None motion: dict[str, Any] arrival_grace_ms: int = 0 @classmethod def from_dict(cls, document: dict[str, Any]) -> "ModeTimelinePhase": phase_id = str(document.get("id", "")).strip() if not phase_id: raise ValueError("mode timeline phase id must not be empty") duration_ms = document.get("duration_ms") completion = str( document.get( "completion", "elapsed_time" if duration_ms is not None else "all_targets_reached", ) ).strip() if completion not in {"elapsed_time", "all_targets_reached"}: raise ValueError( f"mode timeline phase {phase_id} completion must be elapsed_time or all_targets_reached" ) if completion == "elapsed_time": if ( not isinstance(duration_ms, int) or isinstance(duration_ms, bool) or duration_ms <= 0 ): raise ValueError( f"mode timeline phase {phase_id} duration_ms must be a positive integer" ) elif duration_ms is not None: raise ValueError( f"mode timeline phase {phase_id} duration_ms must be omitted for all_targets_reached" ) timeout_ms = document.get("timeout_ms") if timeout_ms is not None and ( not isinstance(timeout_ms, int) or isinstance(timeout_ms, bool) or timeout_ms <= 0 ): raise ValueError( f"mode timeline phase {phase_id} timeout_ms must be a positive integer" ) if completion == "all_targets_reached" and timeout_ms is None: raise ValueError( f"mode timeline phase {phase_id} timeout_ms is required for all_targets_reached" ) arrival_grace_ms = document.get("arrival_grace_ms", 0) if ( not isinstance(arrival_grace_ms, int) or isinstance(arrival_grace_ms, bool) or arrival_grace_ms < 0 ): raise ValueError( f"mode timeline phase {phase_id} arrival_grace_ms must be a non-negative integer" ) targets = document.get("targets", {}) effects = document.get("effects", {}) motion = document.get("motion", {}) if ( not isinstance(targets, dict) or not isinstance(effects, dict) or not isinstance(motion, dict) ): raise ValueError( f"mode timeline phase {phase_id} targets/effects/motion must be objects" ) return cls( id=phase_id, name=str(document.get("name", phase_id)), duration_ms=duration_ms, transition=str(document.get("transition", "hold")), targets=dict(targets), effects=dict(effects), completion=completion, timeout_ms=timeout_ms, motion=dict(motion), arrival_grace_ms=arrival_grace_ms, ) def to_yaml(self) -> dict[str, Any]: result = { "id": self.id, "name": self.name, "transition": self.transition, "targets": dict(self.targets), "effects": dict(self.effects), "completion": self.completion, } if self.duration_ms is not None: result["duration_ms"] = self.duration_ms if self.timeout_ms is not None: result["timeout_ms"] = self.timeout_ms if self.motion: result["motion"] = dict(self.motion) if self.arrival_grace_ms: result["arrival_grace_ms"] = self.arrival_grace_ms return result @dataclass(frozen=True) class ModeTimeline: repeat: str cycle_duration_ms: int | None entry: dict[str, Any] phases: tuple[ModeTimelinePhase, ...] max_runtime_ms: int | None = None @classmethod def from_dict(cls, document: dict[str, Any]) -> "ModeTimeline": repeat = str(document.get("repeat", "")).strip() if repeat not in {"once", "infinite", "until_stopped_or_timeout"}: raise ValueError( "mode timeline repeat must be once, infinite, or until_stopped_or_timeout" ) max_runtime_ms = document.get("max_runtime_ms") if max_runtime_ms is not None and ( not isinstance(max_runtime_ms, int) or isinstance(max_runtime_ms, bool) or max_runtime_ms <= 0 ): raise ValueError("mode timeline max_runtime_ms must be a positive integer") if repeat == "until_stopped_or_timeout" and max_runtime_ms is None: raise ValueError( "mode timeline max_runtime_ms is required for until_stopped_or_timeout" ) duration_ms = document.get("cycle_duration_ms") if duration_ms is not None and ( not isinstance(duration_ms, int) or isinstance(duration_ms, bool) or duration_ms <= 0 ): raise ValueError("mode timeline cycle_duration_ms must be positive") raw_entry = document.get("entry", {}) raw_phases = document.get("phases", []) if not isinstance(raw_entry, dict): raise ValueError("mode timeline entry must be an object") if not isinstance(raw_phases, list) or not raw_phases: raise ValueError("mode timeline must contain at least one phase") phases = tuple(ModeTimelinePhase.from_dict(item) for item in raw_phases) phase_ids = [phase.id for phase in phases] if len(set(phase_ids)) != len(phase_ids): raise ValueError("mode timeline phase ids must be unique") timed_phases = all(phase.duration_ms is not None for phase in phases) if timed_phases: calculated_duration = sum( int(phase.duration_ms or 0) for phase in phases ) if calculated_duration != duration_ms: raise ValueError( "mode timeline cycle_duration_ms does not equal the sum of phases: " f"declared={duration_ms}, calculated={calculated_duration}" ) elif duration_ms is not None: raise ValueError( "mode timeline cycle_duration_ms must be omitted when a phase completes by target arrival" ) return cls( repeat=repeat, cycle_duration_ms=duration_ms, entry=dict(raw_entry), phases=phases, max_runtime_ms=max_runtime_ms, ) def to_yaml(self) -> dict[str, Any]: result = { "repeat": self.repeat, "entry": dict(self.entry), "phases": [phase.to_yaml() for phase in self.phases], } if self.cycle_duration_ms is not None: result["cycle_duration_ms"] = self.cycle_duration_ms if self.max_runtime_ms is not None: result["max_runtime_ms"] = self.max_runtime_ms return result def public_dict(self) -> dict[str, Any]: return self.to_yaml() @dataclass(frozen=True) class ModeProfile: id: str name: str aliases: tuple[str, ...] description: str icon: str operations: tuple[ModeOperation, ...] source: str common: bool = True vehicle_ids: tuple[str, ...] = () timeline: ModeTimeline | None = None user_defined: bool = False instruction: str = "" command_flow: dict[str, Any] | None = None schedule: ModeSchedule | None = None @classmethod def from_yaml(cls, path: Path, profile_root: Path | None = None) -> "ModeProfile": document = yaml.safe_load(path.read_text(encoding="utf-8")) if not isinstance(document, dict): raise ValueError(f"{path} must contain a YAML object") operations = document.get("operations", []) if not isinstance(operations, list) or not operations: raise ValueError(f"{path} must contain at least one operation") name = _clean_name(str(document["name"])) common, vehicle_ids = _mode_scope(document, path, profile_root) aliases = _mode_aliases( name, [str(alias) for alias in document.get("aliases", [])], ) raw_timeline = document.get("timeline") if raw_timeline is not None and not isinstance(raw_timeline, dict): raise ValueError(f"{path} timeline must be an object") return cls( id=str(document["id"]), name=name, aliases=tuple(aliases), description=str(document.get("description", "")), icon=str(document.get("icon", "bookmark")), operations=tuple(ModeOperation.from_dict(item) for item in operations), source=str(path), common=common, vehicle_ids=vehicle_ids, timeline=( ModeTimeline.from_dict(raw_timeline) if isinstance(raw_timeline, dict) else None ), user_defined=_is_user_profile_path(path, profile_root), instruction=_clean_instruction(document.get("instruction", "")), command_flow=_clean_command_flow(document.get("command_flow")), schedule=(ModeSchedule.from_dict(document["schedule"]) if document.get("schedule") is not None else None), ) def public_dict(self) -> dict[str, Any]: is_cyclic = self.schedule is not None or self.timeline is not None or any( operation.capability == "seat.start_dynamic_mode" for operation in self.operations ) result = { "id": self.id, "name": self.name, "aliases": list(self.aliases), "description": self.description, "icon": self.icon, "domains": sorted({operation.domain for operation in self.operations}), "operation_count": len(self.operations), "operations": [operation.public_dict() for operation in self.operations], "scope": { "common": self.common, "vehicle_ids": list(self.vehicle_ids), }, "behavior": "cyclic" if is_cyclic else "static", "origin": "user" if self.user_defined else "preset", "deletable": self.user_defined, "instruction": self.instruction, "command_flow": self.command_flow, } if self.timeline is not None: result["timeline"] = self.timeline.public_dict() if self.schedule is not None: result["schedule"] = self.schedule.to_dict() result["operation_count"] = self.schedule.operation_count return result class ModeProfileRegistry: """Load preset profiles and optionally keep user profiles in a session overlay.""" def __init__(self, profile_root: Path, user_root: Path | None = None): self.profile_root = Path(profile_root) self.user_root = Path(user_root) if user_root is not None else self.profile_root self.session_user_profiles = user_root is not None self._profiles: list[ModeProfile] = [] self._change_listeners: list[Callable[["ModeProfileRegistry"], None]] = [] self._write_lock = RLock() self.reload() def reload(self) -> None: with self._write_lock: profiles: list[ModeProfile] = [] if self.profile_root.exists(): for path in sorted(self.profile_root.rglob("*.yaml")): if not path.is_file(): continue if self.session_user_profiles and _is_user_profile_path( path, self.profile_root ): continue profiles.append(ModeProfile.from_yaml(path, self.profile_root)) if self.session_user_profiles and self.user_root.exists(): profiles.extend( ModeProfile.from_yaml(path, self.user_root) for path in sorted(self.user_root.rglob("*.yaml")) if path.is_file() ) self._profiles = profiles self._notify_change_listeners() def add_change_listener( self, listener: Callable[["ModeProfileRegistry"], None], *, notify: bool = True, ) -> None: """Run ``listener`` after every successful registry reload. Mode profiles remain YAML-backed. Listeners are intended for derived, read-only projections such as the LAN scene catalog; they do not replace or alter the existing persistence path. """ with self._write_lock: if listener not in self._change_listeners: self._change_listeners.append(listener) if notify: listener(self) def _notify_change_listeners(self) -> None: for listener in tuple(self._change_listeners): listener(self) def profiles(self) -> list[ModeProfile]: return list(self._profiles) def public_modes(self, vehicle_id: str | None = None) -> list[dict[str, Any]]: profiles = self._profiles_for_vehicle(vehicle_id) profiles.sort(key=lambda profile: 0 if profile.user_defined else 1) preset_ids = { profile.id for profile in self._visible_candidates_for_vehicle(vehicle_id) if not profile.user_defined } public = [] for profile in profiles: item = profile.public_dict() if profile.user_defined and profile.id in preset_ids: item["origin"] = "customized" item["preset_override"] = True public.append(item) return public def get_profile( self, mode_id: str, vehicle_id: str | None = None, ) -> ModeProfile: """Resolve the effective preset or user override by its stable ID.""" clean_mode_id = str(mode_id).strip() profile = next( ( item for item in self._profiles_for_vehicle(vehicle_id) if item.id == clean_mode_id ), None, ) if profile is None: raise KeyError(clean_mode_id) return profile def match_text( self, utterance: str, vehicle_id: str | None = None, ) -> list[ModeProfile]: haystack = _normalize_match_text(utterance) selected = { profile.id: profile for profile in self._profiles_for_vehicle(vehicle_id) } matches = [] matched_ids: set[str] = set() for profile in self._visible_candidates_for_vehicle(vehicle_id): candidates = [profile.id, profile.name, *profile.aliases] if any( candidate and _mode_candidate_matches(haystack, candidate) for candidate in candidates ): if profile.id in matched_ids: continue matched_ids.add(profile.id) matches.append(selected[profile.id]) return matches def resolve_exact_name( self, value: str, vehicle_id: str | None = None, ) -> ModeProfile | None: """Resolve one visible mode by an exact ID, name, or alias. Scene creation must not use the substring semantics of :meth:`match_text`: a name can contain another mode alias and accidentally combine two profiles. Returning ``None`` for both no match and an ambiguous alias keeps that path fail-closed so callers can choose an explicit fallback without combining modes. """ needle = _normalize_match_text(str(value).strip()) if not needle: return None matches = [ profile for profile in self._profiles_for_vehicle(vehicle_id) if needle in { _normalize_match_text(str(candidate)) for candidate in (profile.id, profile.name, *profile.aliases) if str(candidate).strip() } ] return matches[0] if len(matches) == 1 else None @staticmethod def profile_revision(profile: ModeProfile) -> str: """Return a stable semantic revision for optimistic mode updates.""" document = { "id": profile.id, "name": profile.name, "aliases": list(profile.aliases), "description": profile.description, "icon": profile.icon, "scope": { "common": profile.common, "vehicle_ids": list(profile.vehicle_ids), }, "operations": [operation.to_yaml() for operation in profile.operations], "timeline": profile.timeline.to_yaml() if profile.timeline is not None else None, "schedule": profile.schedule.to_dict() if profile.schedule is not None else None, "instruction": profile.instruction, "command_flow": profile.command_flow, } canonical = json.dumps( document, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ) return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:16] @staticmethod def requests_for_profiles(profiles: list[ModeProfile]) -> list[dict[str, Any]]: return [ operation.to_request(profile.name) for profile in profiles for operation in profile.operations ] def save_mode( self, name: str, operations: list[dict[str, Any] | ModeOperation], mode_id: str | None = None, aliases: list[str] | None = None, vehicle_id: str | None = None, *, description: str | None = None, icon: str = "bookmark", instruction: str = "", command_flow: dict[str, Any] | None = None, timeline: ModeTimeline | None = None, schedule: ModeSchedule | None = None, expected_revision: str | None = None, create_only: bool = False, ) -> ModeProfile: clean_name = _clean_name(name) canonical_operations = [ operation if isinstance(operation, ModeOperation) else ModeOperation.from_dict(operation) for operation in operations ] if not canonical_operations: raise ValueError("mode profile must contain at least one operation") profile_id = _clean_mode_id(mode_id or _slugify_mode_id(clean_name)) clean_instruction = _clean_instruction(instruction) clean_command_flow = _clean_command_flow(command_flow) scope = ( {"vehicle_ids": [vehicle_id]} if vehicle_id else {"common": True, "vehicle_ids": []} ) document = { "id": profile_id, "name": clean_name, "aliases": _mode_aliases(clean_name, aliases), "description": str(description or "用户固化的长期模式配置。")[:500], "icon": str(icon or "bookmark")[:64], "scope": scope, "operations": [ operation.to_yaml() for operation in canonical_operations ], } if clean_instruction: document["instruction"] = clean_instruction if clean_command_flow is not None: document["command_flow"] = clean_command_flow if timeline is not None: document["timeline"] = timeline.to_yaml() if schedule is not None: document["schedule"] = schedule.to_dict() path = ( self.user_root / "vehicles" / vehicle_id / "user" / f"{profile_id}.yaml" if vehicle_id else self.user_root / "common" / "user" / f"{profile_id}.yaml" ) with self._write_lock: # Refresh inside the same critical section as the compare-and-swap. # This makes concurrent assistant saves one-winner within the server # process instead of letting two stale previews overwrite each other. self.reload() visible_profiles = self._profiles_for_vehicle(vehicle_id) current = next( (profile for profile in visible_profiles if profile.id == profile_id), None, ) normalized_name = _normalize_match_text(clean_name) name_collision = any( profile.id != profile_id and normalized_name in { _normalize_match_text(str(candidate)) for candidate in (profile.id, profile.name, *profile.aliases) if str(candidate).strip() } for profile in visible_profiles ) if name_collision: raise ValueError("该模式名称已被其他模式使用。") if create_only and current is not None: raise ValueError("该模式已由其他会话创建，请刷新后重试。") if expected_revision is not None: expected = str(expected_revision).strip() if ( not expected or current is None or self.profile_revision(current) != expected ): raise ValueError( "功能模式已被其他会话更新，请重新生成预览后再保存。" ) path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_text( yaml.safe_dump(document, allow_unicode=True, sort_keys=False), encoding="utf-8", ) temporary.replace(path) self.reload() return ModeProfile.from_yaml(path, self.user_root) def delete_user_mode( self, mode_id: str, vehicle_id: str | None = None, ) -> ModeProfile: with self._write_lock: return self._delete_user_mode_locked(mode_id, vehicle_id) def _delete_user_mode_locked( self, mode_id: str, vehicle_id: str | None = None, ) -> ModeProfile: clean_id = str(mode_id).strip() if not clean_id: raise ValueError("mode id must not be empty") profile = next( ( item for item in self._profiles_for_vehicle(vehicle_id) if item.id == clean_id ), None, ) if profile is None: raise KeyError(clean_id) if not profile.user_defined: raise PermissionError("preset mode profiles cannot be deleted") path = Path(profile.source).resolve() relative_parts = None for root in {self.profile_root.resolve(), self.user_root.resolve()}: try: relative_parts = path.relative_to(root).parts break except ValueError: continue if relative_parts is None: raise PermissionError("mode profile is outside the configured registry") if "user" not in relative_parts or path.suffix.lower() != ".yaml": raise PermissionError("only user-defined mode profiles can be deleted") if not path.is_file(): raise KeyError(clean_id) path.unlink() backup_path = _initial_backup_path(path) if backup_path.is_file(): backup_path.unlink() self.reload() return profile def update_mode( self, profile: ModeProfile, operations: list[dict[str, Any] | ModeOperation], ) -> ModeProfile: with self._write_lock: return self._update_mode_locked(profile, operations) def _update_mode_locked( self, profile: ModeProfile, operations: list[dict[str, Any] | ModeOperation], ) -> ModeProfile: canonical_updates = _canonical_operations(operations) if not canonical_updates: raise ValueError("mode profile update must contain at least one operation") merged_operations = _merge_operations(profile.operations, canonical_updates) if self.session_user_profiles and not profile.user_defined: # Runtime edits must never rewrite repository-owned presets. Save # a same-id user overlay instead; normal registry ranking makes it # the effective profile, while deleting the overlay reveals the # untouched preset again. updated = self.save_mode( profile.name, merged_operations, mode_id=profile.id, aliases=list(profile.aliases), vehicle_id=( profile.vehicle_ids[0] if len(profile.vehicle_ids) == 1 else None ), description=profile.description, icon=profile.icon, instruction=profile.instruction, command_flow=profile.command_flow, timeline=profile.timeline, ) backup_path = _initial_backup_path(Path(updated.source)) if not backup_path.exists(): _replace_yaml( backup_path, _mode_profile_document(profile, list(profile.operations)), ) return updated path = Path(profile.source) if not path.is_file(): raise ValueError(f"mode profile source does not exist: {path}") backup_path = _initial_backup_path(path) if not backup_path.exists(): backup_path.write_text(path.read_text(encoding="utf-8"), encoding="utf-8") document = _mode_profile_document(profile, merged_operations) _replace_yaml(path, document) self.reload() return ModeProfile.from_yaml(path, self._root_for_path(path)) def restore_initial_mode(self, profile: ModeProfile) -> ModeProfile: with self._write_lock: return self._restore_initial_mode_locked(profile) def _restore_initial_mode_locked(self, profile: ModeProfile) -> ModeProfile: path = Path(profile.source) backup_path = _initial_backup_path(path) if not backup_path.is_file(): raise ValueError(f"no initial mode backup found: {backup_path}") path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_text(backup_path.read_text(encoding="utf-8"), encoding="utf-8") temporary.replace(path) self.reload() return ModeProfile.from_yaml(path, self._root_for_path(path)) def _root_for_path(self, path: Path) -> Path: resolved = path.resolve() for root in (self.user_root.resolve(), self.profile_root.resolve()): try: resolved.relative_to(root) return root except ValueError: continue return self.profile_root def _profiles_for_vehicle(self, vehicle_id: str | None) -> list[ModeProfile]: selected: dict[str, ModeProfile] = {} order: list[str] = [] for profile in self._visible_candidates_for_vehicle(vehicle_id): if profile.id not in selected: order.append(profile.id) selected[profile.id] = profile continue current = selected[profile.id] if _profile_rank(profile, vehicle_id) >= _profile_rank(current, vehicle_id): selected[profile.id] = profile return [selected[profile_id] for profile_id in order] def _visible_candidates_for_vehicle( self, vehicle_id: str | None, ) -> list[ModeProfile]: return [ profile for profile in self._profiles if _is_visible_for_vehicle(profile, vehicle_id) ] def _canonical_operations( operations: list[dict[str, Any] | ModeOperation], ) -> list[ModeOperation]: return [ operation if isinstance(operation, ModeOperation) else ModeOperation.from_dict(operation) for operation in operations ] def _operation_signature(operation: ModeOperation) -> tuple[str, str, str]: return (operation.domain, operation.capability, operation.target_scope) def _merge_operations( existing: tuple[ModeOperation, ...], updates: list[ModeOperation], ) -> list[ModeOperation]: merged = list(existing) positions = { _operation_signature(operation): index for index, operation in enumerate(merged) } for operation in updates: signature = _operation_signature(operation) index = positions.get(signature) if index is None: positions[signature] = len(merged) merged.append(operation) continue merged[index] = operation return merged def _initial_backup_path(path: Path) -> Path: return path.with_suffix(path.suffix + ".initial.bak") def _mode_profile_document( profile: ModeProfile, operations: list[ModeOperation], ) -> dict[str, Any]: scope: dict[str, Any] = { "common": profile.common, "vehicle_ids": list(profile.vehicle_ids), } document = { "id": profile.id, "name": profile.name, "aliases": list(profile.aliases), "description": profile.description, "icon": profile.icon, "scope": scope, "operations": [operation.to_yaml() for operation in operations], } if profile.timeline is not None: document["timeline"] = profile.timeline.to_yaml() if profile.schedule is not None: document["schedule"] = profile.schedule.to_dict() if profile.instruction: document["instruction"] = profile.instruction if profile.command_flow is not None: document["command_flow"] = profile.command_flow return document def _replace_yaml(path: Path, document: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_text( yaml.safe_dump(document, allow_unicode=True, sort_keys=False), encoding="utf-8", ) temporary.replace(path) def _mode_scope( document: dict[str, Any], path: Path, profile_root: Path | None, ) -> tuple[bool, tuple[str, ...]]: scope = document.get("scope") if isinstance(scope, dict): vehicle_ids = tuple( str(vehicle_id).strip() for vehicle_id in scope.get("vehicle_ids", []) if str(vehicle_id).strip() ) common = bool(scope.get("common", not vehicle_ids)) return common, vehicle_ids if profile_root is not None: try: parts = path.relative_to(profile_root).parts except ValueError: parts = () if len(parts) >= 3 and parts[0] == "vehicles": return False, (parts[1],) if parts and parts[0] == "common": return True, () return True, () def _is_user_profile_path(path: Path, profile_root: Path | None) -> bool: if profile_root is None: return path.parent.name == "user" try: relative_parts = path.resolve().relative_to(profile_root.resolve()).parts except ValueError: return False return "user" in relative_parts def _is_visible_for_vehicle( profile: ModeProfile, vehicle_id: str | None, ) -> bool: if vehicle_id is None: return True return profile.common or vehicle_id in profile.vehicle_ids def _profile_rank(profile: ModeProfile, vehicle_id: str | None) -> int: if vehicle_id is not None and vehicle_id in profile.vehicle_ids: return 2 if profile.common: return 1 return 0