"""Portable trend configuration shared by SETGUI and Android (no Qt/Android). Numeric CAN decoding uses set_trends.c through NativeTrends. JSON adapters on Python/Kotlin implement the same documented version-1 interchange contract. """ from __future__ import annotations import ctypes import json import math import re import uuid from collections import deque from dataclasses import asdict, dataclass from typing import Mapping MAX_SIGNALS = 64 MAX_POINTS = 600 MAX_FILE_BYTES = 1024 * 1024 PROFILE_SOURCES = { "TMS2812": ("TMS_MEMORY",), "SET_V1": ("SET_GAS", "SET_SENSOR"), "BALZAM_CAN": ("CAN_RAW",), **{name: ("CAN_GAS", "CAN_RAW") for name in ("CAN_BRIDGE", "GS_USB_CAN", "SLCAN", "CANGAROO_SLCAN")}, } def parse_address(value: str, maximum: int) -> int: text = value.strip() if not re.fullmatch(r"(?:0[xX][0-9a-fA-F]+|[0-9]+)", text): raise ValueError("Address must be decimal or explicit 0x hexadecimal") number = int(text[2:], 16) if text.lower().startswith("0x") else int(text, 10) if not 0 <= number <= maximum: raise ValueError("Address out of range") return number def normalize_rom(value: str) -> str: return value.strip().replace("-", "").replace(" ", "").upper() @dataclass(frozen=True) class TrendSignal: # Field spellings intentionally match the JSON/Kotlin contract. id: str order: int = 1 name: str = "Тренд 1" source: str = "TMS_MEMORY" address: str = "0x00000100" color: str = "#2F91FF" visible: bool = True valueType: str = "UINT16" deviceType: int = 7 device: int = 13 byteOffset: int = 0 extended: bool = True @classmethod def new(cls, profile: str, existing: tuple = ()) -> "TrendSignal": order = next(number for number in range(1, 10000) if all(item.order != number for item in existing)) source = PROFILE_SOURCES[profile][0] address = {"TMS_MEMORY": "0x00000100", "SET_GAS": "0x0000", "CAN_GAS": "0x0000", "CAN_RAW": "0x00BA0010"}[source] colors = ("#2F91FF", "#FFB547", "#52D6A4", "#F8798D", "#B79AFF", "#4AD9E8", "#E8DA68", "#E7ECF3") return cls(str(uuid.uuid4()), order, f"Тренд {order}", source, address, colors[(order - 1) % len(colors)]) def validate(self, profile: str) -> None: for field in ("id", "name", "source", "address", "color", "valueType"): if type(getattr(self, field)) is not str: raise ValueError(f"{field} must be a string") for field in ("order", "deviceType", "device", "byteOffset"): if type(getattr(self, field)) is not int: raise ValueError(f"{field} must be an integer") if type(self.visible) is not bool or type(self.extended) is not bool: raise ValueError("Visibility and CAN format must be boolean") if not self.id.strip() or len(self.id) > 80 or not 1 <= self.order <= 9999: raise ValueError("Invalid ID/order") if not self.name.strip() or len(self.name) > 100 or len(self.address) > 64: raise ValueError("Invalid name/address length") if self.source not in PROFILE_SOURCES.get(profile, ()): raise ValueError("Source does not match connection protocol") if not re.fullmatch(r"#[0-9a-fA-F]{6}", self.color): raise ValueError("Color must be #RRGGBB") if self.valueType not in ("UINT16", "INT16"): raise ValueError("Unsupported word type") if self.source == "SET_SENSOR": if not re.fullmatch(r"[0-9A-F]{16}", normalize_rom(self.address)): raise ValueError("ROM must contain 16 hexadecimal digits") else: maximum = {"TMS_MEMORY": 0xFFFFFFFF, "CAN_GAS": 0xFFFF, "SET_GAS": 0xFFFF, "CAN_RAW": 0x1FFFFFFF if self.extended else 0x7FF}[self.source] parse_address(self.address, maximum) if self.source == "CAN_GAS" and not (0 <= self.deviceType <= 7 and 0 <= self.device <= 15): raise ValueError("Invalid ProtoCAN device") if self.source == "CAN_RAW" and not 0 <= self.byteOffset <= 6: raise ValueError("Word offset must be 0..6") def word_value(self, word: int) -> float: if not 0 <= word <= 65535: raise ValueError("Not a 16-bit word") return float(word - 65536 if self.valueType == "INT16" and word >= 32768 else word) def validate_settings(settings: Mapping[str, list[TrendSignal]]) -> None: ids = set() for profile, signals in settings.items(): if profile not in PROFILE_SOURCES or len(signals) > MAX_SIGNALS: raise ValueError("Unknown profile or too many signals") orders = set() for signal in signals: signal.validate(profile) if signal.id in ids or signal.order in orders: raise ValueError("Duplicate trend ID or order") ids.add(signal.id) orders.add(signal.order) def encode_settings(settings: Mapping[str, list[TrendSignal]]) -> str: validate_settings(settings) return json.dumps({"format": "setflash-trends", "version": 1, "profiles": { profile: [asdict(signal) for signal in sorted(signals, key=lambda item: item.order)] for profile, signals in settings.items()}}, ensure_ascii=False, indent=2) def decode_settings(text: str) -> dict[str, list[TrendSignal]]: if len(text.encode("utf-8")) > MAX_FILE_BYTES: raise ValueError("Trend configuration exceeds 1 MiB") # Python 3.8 is used by the Windows 7 port; remove at most one BOM. root = json.loads(text[1:] if text.startswith("\ufeff") else text) if not isinstance(root, dict) or root.get("format") != "setflash-trends" or type(root.get("version")) is not int or root["version"] != 1: raise ValueError("Unknown trend configuration format/version") profiles = root.get("profiles") if not isinstance(profiles, dict): raise ValueError("profiles must be an object") result = {} fields = set(TrendSignal.__dataclass_fields__) for profile, items in profiles.items(): if not isinstance(items, list) or len(items) > MAX_SIGNALS: raise ValueError("Expected up to 64 signals") signals = [] for item in items: if not isinstance(item, dict) or not fields <= item.keys(): raise ValueError("Missing signal fields") signals.append(TrendSignal(**{key: item[key] for key in fields})) result[profile] = signals validate_settings(result) return result class TrendHistory: """Bounded per-signal monotonic-time history; visibility belongs to the renderer.""" def __init__(self) -> None: self.series: dict[str, deque] = {} def append(self, values: Mapping[str, float], timestamp_ms: int) -> None: for key, value in values.items(): if math.isfinite(value): self.series.setdefault(key, deque(maxlen=MAX_POINTS)).append((timestamp_ms, value)) class NativeTrends: """Thin ctypes port: accepts NativeProtocol.lib or any loaded SETProtocol CDLL. Existing older DLLs keep working for other protocol functions. This optional feature reports a clear error until the shared library is rebuilt. """ def __init__(self, library: ctypes.CDLL) -> None: self.lib = library try: self.decode = library.set_trend_can_value except AttributeError as error: raise RuntimeError("Rebuild SETProtocol with set_trends.c to use trends") from error self.decode.argtypes = [ctypes.c_uint8, ctypes.c_uint32, ctypes.c_uint8, ctypes.c_uint8, ctypes.c_uint8, ctypes.c_uint8, ctypes.c_uint8, ctypes.c_uint32, ctypes.c_uint8, ctypes.c_void_p, ctypes.c_size_t] self.decode.restype = ctypes.c_int32 def can_value(self, signal: TrendSignal, can_id: int, flags: int, data: bytes) -> float | None: if signal.source not in ("CAN_GAS", "CAN_RAW"): return None signal.validate("CAN_BRIDGE") if not 0 <= can_id <= 0x1FFFFFFF or not 0 <= flags <= 255 or len(data) > 8: return None payload = (ctypes.c_uint8 * len(data)).from_buffer_copy(data) value = self.decode(1 if signal.source == "CAN_GAS" else 2, parse_address(signal.address, 0x1FFFFFFF), signal.deviceType, signal.device, signal.byteOffset, signal.extended, signal.valueType == "INT16", can_id, flags, payload, len(data)) return None if value == -2147483648 else float(value)