246 lines
12 KiB
Python
246 lines
12 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
|
||
multiplier: float = 1.0
|
||
iq: 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", "iq"):
|
||
if type(getattr(self, field)) is not int:
|
||
raise ValueError(f"{field} must be an integer")
|
||
if type(self.multiplier) not in (int, float) or not math.isfinite(self.multiplier):
|
||
raise ValueError("multiplier must be a finite number")
|
||
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")
|
||
if not 0 <= self.iq <= 30:
|
||
raise ValueError("IQ must be 0..30")
|
||
|
||
def display_value(self, raw_value: float) -> float:
|
||
return raw_value * self.multiplier / (1 << self.iq)
|
||
|
||
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__)
|
||
required_fields = fields - {"multiplier", "iq"}
|
||
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 required_fields <= item.keys():
|
||
raise ValueError("Missing signal fields")
|
||
signals.append(TrendSignal(**{key: item[key] for key in fields if key in item}))
|
||
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
|
||
self._word = library.set_trend_word_value
|
||
self._word.argtypes = [ctypes.c_uint16, ctypes.c_uint8]
|
||
self._word.restype = ctypes.c_int32
|
||
self._watch_request = library.set_trend_watch_request
|
||
self._watch_request.argtypes = [ctypes.c_uint16, ctypes.POINTER(ctypes.c_uint16),
|
||
ctypes.c_size_t, ctypes.c_void_p, ctypes.c_size_t]
|
||
self._watch_request.restype = ctypes.c_size_t
|
||
self._watch_ack = library.set_trend_watch_ack
|
||
self._watch_ack.argtypes = [ctypes.c_void_p, ctypes.c_size_t, ctypes.c_uint16, ctypes.c_size_t]
|
||
self._watch_ack.restype = ctypes.c_int
|
||
self._watch_decode = library.set_trend_watch_decode
|
||
self._watch_decode.argtypes = [ctypes.c_void_p, ctypes.c_size_t,
|
||
ctypes.POINTER(ctypes.c_uint32), ctypes.POINTER(ctypes.c_uint16), ctypes.c_size_t]
|
||
self._watch_decode.restype = ctypes.c_int
|
||
|
||
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)
|
||
|
||
def word_value(self, signal: TrendSignal, word: int) -> float:
|
||
if not 0 <= word <= 65535:
|
||
raise ValueError("Not a 16-bit word")
|
||
return float(self._word(word, signal.valueType == "INT16"))
|
||
|
||
def watch_request(self, period_ms: int, addresses: list[int]) -> bytes:
|
||
if not 0 <= period_ms <= 65535 or len(addresses) > MAX_SIGNALS or any(
|
||
type(address) is not int or not 0 <= address <= 65535 for address in addresses):
|
||
raise ValueError("Invalid GAS watch request")
|
||
source = (ctypes.c_uint16 * len(addresses))(*addresses)
|
||
output = (ctypes.c_uint8 * (4 + len(addresses) * 2))()
|
||
size = self._watch_request(period_ms, source, len(addresses), output, len(output))
|
||
if not size:
|
||
raise ValueError("Invalid GAS watch request")
|
||
return bytes(output[:size])
|
||
|
||
def validate_watch_ack(self, payload: bytes, period_ms: int, count: int) -> None:
|
||
data = (ctypes.c_uint8 * len(payload)).from_buffer_copy(payload)
|
||
if not self._watch_ack(data, len(payload), period_ms, count):
|
||
raise ValueError("Device did not accept the complete GAS subscription")
|
||
|
||
def watch_values(self, payload: bytes, expected_count: int | None = None) -> tuple[int, list[int]]:
|
||
data = (ctypes.c_uint8 * len(payload)).from_buffer_copy(payload)
|
||
timestamp = ctypes.c_uint32()
|
||
words = (ctypes.c_uint16 * MAX_SIGNALS)()
|
||
count = self._watch_decode(data, len(payload), ctypes.byref(timestamp), words, MAX_SIGNALS)
|
||
if count < 0 or expected_count is not None and count != expected_count:
|
||
raise ValueError("Invalid GAS watch data")
|
||
return timestamp.value, list(words[:count])
|