194 lines
8.5 KiB
Python
194 lines
8.5 KiB
Python
"""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)
|