Добавить общие графики, декодер KONOR и порт STM32 bxCAN

python/protocan/plot.py

@@ -0,0 +1,159 @@
+"""Plot interaction port. All numerical operations use templates' set_plot.c.
+
+This module has no Qt dependency. The application supplies its SETProtocol CDLL.
+Units, clocks, acquisition, colours and rendering belong to the application.
+"""
+from __future__ import annotations
+
+import ctypes
+from dataclasses import dataclass, replace
+from enum import IntEnum
+from typing import Optional
+
+
+class Axis(IntEnum):
+    X = 1
+    Y = 2
+
+
+class Marker(IntEnum):
+    A = 0
+    B = 1
+    C = 2
+    D = 3
+    E = 4
+    F = 5
+    G = 6
+    H = 7
+
+    @property
+    def horizontal(self) -> bool:
+        return self in (Marker.E, Marker.F, Marker.G, Marker.H)
+
+
+class PlotMath:
+    """Typed operations over the versioned allocation-free C ABI."""
+    def __init__(self, library: ctypes.CDLL) -> None:
+        self.library = library
+        library.set_plot_abi_version.restype = ctypes.c_uint32
+        library.set_plot_abi_version.argtypes = []
+        if library.set_plot_abi_version() != 1:
+            raise RuntimeError("Unsupported plot ABI")
+        library.set_plot_eval.argtypes = [ctypes.c_uint32, ctypes.POINTER(ctypes.c_double),
+            ctypes.c_size_t, ctypes.POINTER(ctypes.c_double), ctypes.c_size_t]
+        library.set_plot_eval.restype = ctypes.c_size_t
+
+    def call(self, operation: int, *values: float) -> tuple:
+        inputs = (ctypes.c_double * len(values))(*values)
+        output = (ctypes.c_double * 4)()
+        count = self.library.set_plot_eval(operation, inputs, len(values), output, 4)
+        if not count:
+            raise ValueError("Invalid plot operation %s" % operation)
+        return tuple(output[:count])
+
+    def pinch_axis(self, dx: float, dy: float, slop: float) -> Optional[Axis]:
+        value = int(self.call(1, dx, dy, slop)[0])
+        return Axis(value) if value else None
+
+    def tick_step(self, span: float, pixels: float) -> float:
+        return self.call(5, span, pixels)[0]
+
+    def delta(self, a: float, b: float, multiplier: float = 1) -> float:
+        return self.call(6, a, b, multiplier)[0]
+
+
+@dataclass(frozen=True)
+class Viewport:
+    x: float = 0.0
+    y: float = 0.0
+    width: float = 1.0
+    height: float = 1.0
+
+    def transform(self, core: PlotMath, zoom_x: float = 1, zoom_y: float = 1,
+                  pan_x: float = 0, pan_y: float = 0,
+                  focus_x: float = 0.5, focus_y: float = 0.5) -> "Viewport":
+        return Viewport(*core.call(0, self.x, self.y, self.width, self.height,
+            zoom_x, zoom_y, pan_x, pan_y, focus_x, focus_y))
+
+
+@dataclass(frozen=True)
+class Bounds:
+    left: float
+    right: float
+    bottom: float
+    top: float
+
+    def fraction(self, core: PlotMath, value: float, horizontal: bool) -> float:
+        return core.call(2, value, self.bottom if horizontal else self.left,
+            self.top if horizontal else self.right, int(horizontal))[0]
+
+    def value(self, core: PlotMath, fraction: float, horizontal: bool) -> float:
+        return core.call(3, fraction, self.bottom if horizontal else self.left,
+            self.top if horizontal else self.right, int(horizontal))[0]
+
+    def visible(self, core: PlotMath, viewport: Viewport) -> "Bounds":
+        return Bounds(self.value(core, viewport.x, False),
+            self.value(core, viewport.x + viewport.width, False),
+            self.value(core, viewport.y + viewport.height, True),
+            self.value(core, viewport.y, True))
+
+
+@dataclass(frozen=True)
+class Markers:
+    x_enabled: bool = True
+    y_enabled: bool = False
+    selected: Marker = Marker.A
+    a: Optional[float] = None
+    b: Optional[float] = None
+    c: Optional[float] = None
+    d: Optional[float] = None
+    e: Optional[float] = None
+    f: Optional[float] = None
+    g: Optional[float] = None
+    h: Optional[float] = None
+
+    def value(self, marker: Marker) -> Optional[float]:
+        return getattr(self, marker.name.lower())
+
+    def enabled(self, marker: Marker) -> bool:
+        return self.y_enabled if marker.horizontal else self.x_enabled
+
+    def move(self, marker: Marker, value: float) -> "Markers":
+        return replace(self, **{marker.name.lower(): value})
+
+    def positioned(self, core: PlotMath, bounds: Bounds) -> "Markers":
+        result = self
+        for marker, fraction in ((Marker.A, .2), (Marker.B, .4), (Marker.C, .6), (Marker.D, .8),
+                                 (Marker.E, .2), (Marker.F, .4), (Marker.G, .6), (Marker.H, .8)):
+            if result.value(marker) is None:
+                result = result.move(marker, bounds.value(core, fraction, marker.horizontal))
+        return result
+
+    def reset(self, core: PlotMath, bounds: Bounds) -> "Markers":
+        return replace(self, a=None, b=None, c=None, d=None, e=None, f=None, g=None, h=None).positioned(core, bounds)
+
+    def fraction(self, core: PlotMath, marker: Marker, bounds: Bounds) -> float:
+        value = self.value(marker)
+        if value is None:
+            raise ValueError("Marker has no position")
+        return bounds.fraction(core, value, marker.horizontal)
+
+    def drag(self, core: PlotMath, marker: Marker, delta: float, length: float, bounds: Bounds) -> "Markers":
+        value = core.call(4, self.value(marker), delta, length,
+            bounds.bottom if marker.horizontal else bounds.left,
+            bounds.top if marker.horizontal else bounds.right, int(marker.horizontal))[0]
+        return replace(self.move(marker, value), selected=marker)
+
+    def hit(self, core: PlotMath, x: float, y: float, width: float, height: float,
+            radius: float, bounds: Bounds) -> Optional[Marker]:
+        if not (0 <= x <= width and 0 <= y <= height):
+            return None
+        candidates = []
+        for marker in Marker:
+            if not self.enabled(marker) or self.value(marker) is None:
+                continue
+            fraction = self.fraction(core, marker, bounds)
+            distance = abs(y - fraction * height if marker.horizontal else x - fraction * width)
+            if 0 <= fraction <= 1 and distance <= radius:
+                candidates.append((distance, marker != self.selected, marker))
+        return min(candidates)[2] if candidates else None

python/protocan/protocan.py

@@ -66,6 +66,7 @@ class SettingsResult(IntEnum):
     EEPROM_ERROR = 0x05
     INVALID_LOCATION = 0x06
     BUSY = 0x07
+    INVALID_VALUE = 0x08
 
 
 class BroadcastType(IntEnum):
@@ -145,6 +146,7 @@ SETTINGS_RESULT_RU = {
     SettingsResult.EEPROM_ERROR: "ошибка EEPROM",
     SettingsResult.INVALID_LOCATION: "неверная локация",
     SettingsResult.BUSY: "устройство занято",
+    SettingsResult.INVALID_VALUE: "значение настройки вне диапазона",
 }
 
 BROADCAST_RU = {
@@ -680,6 +682,30 @@ def _decode_settings(res: Decoded) -> None:
         ("  SETTINGS.Position Y", "%d (0x%02X)" % (position, position)),
     ]
 
+    # Зарезервированная локация прибора KONOR: runtime-настройки не являются
+    # ROM датчика, хотя используют тот же однокадровый SETTINGS-транспорт.
+    if assembly == 0xFF and position == 0x01:
+        if res.dlc == 0 and res.id.pm == Route.FROM_PM:
+            res.summary = "SETTINGS RUNTIME GET"
+            return
+        if res.dlc == 8:
+            pulse_seconds = res.data[0]
+            led_hz = int.from_bytes(res.data[1:3], "little")
+            res.fields += [
+                ("  SETTINGS.Pulse period",
+                 "%d s" % pulse_seconds if pulse_seconds else "выключен"),
+                ("  SETTINGS.LED frequency", "%d Hz" % led_hz),
+            ]
+            if led_hz > 500:
+                res.warnings.append("Частота LED должна быть 0..500 Гц")
+            if any(res.data[3:]):
+                res.warnings.append("Резерв SETTINGS Data[3:8] должен быть нулевым")
+            direction = "WRITE" if res.id.pm == Route.FROM_PM else "RESPONSE"
+            period_text = "%d s" % pulse_seconds if pulse_seconds else "выключен"
+            res.summary = ("SETTINGS RUNTIME %s: PULSE %s, LED %d Hz"
+                           % (direction, period_text, led_hz))
+            return
+
     if res.id.pm == Route.FROM_PM:
         if res.dlc == 0:
             operation = "GET"

python/protocan/spectrum.py

@@ -0,0 +1,71 @@
+"""Shared FFT adapter for desktop GUIs; no numpy/Qt dependency or duplicated DSP.
+
+Pass NativeProtocol.lib (rebuilt with set_spectrum.c). Timestamp inputs are seconds.
+See c/set-protocol/include/set_spectrum.h for filtering and normalization semantics.
+"""
+from __future__ import annotations
+import ctypes
+from dataclasses import dataclass
+from enum import IntEnum
+
+
+class Window(IntEnum):
+    RECT = 0
+    HANN = 1
+    HAMMING = 2
+    BLACKMAN = 3
+    FLATTOP = 4
+
+
+class Filter(IntEnum):
+    NONE = 0
+    LOW_PASS = 1
+    HIGH_PASS = 2
+    BAND_PASS = 3
+    NOTCH = 4
+
+
+@dataclass(frozen=True)
+class Spectrum:
+    size: int
+    sample_rate: float
+    jitter: float
+    amplitudes: tuple[float, ...]
+
+    @property
+    def frequencies(self) -> tuple[float, ...]:
+        return tuple(i * self.sample_rate / self.size for i in range(len(self.amplitudes)))
+
+
+class NativeSpectrum:
+    def __init__(self, library: ctypes.CDLL):
+        self.lib = library
+        self._analyze = library.set_spectrum_analyze
+        pointer = ctypes.POINTER(ctypes.c_double)
+        self._analyze.argtypes = [pointer, pointer, ctypes.c_size_t, ctypes.c_size_t,
+            ctypes.c_int, ctypes.c_int, ctypes.c_double, ctypes.c_double, ctypes.c_int,
+            pointer, ctypes.c_size_t, pointer]
+        self._analyze.restype = ctypes.c_int
+
+    def analyze(self, times, values, *, max_size=4096, window=Window.HANN, filter=Filter.NONE,
+                low_hz=10.0, high_hz=100.0, remove_mean=True) -> Spectrum:
+        if type(max_size) is not int or not 16 <= max_size <= 16384 or max_size & (max_size - 1):
+            raise ValueError("FFT size must be a power of two in 16..16384")
+        if len(times) != len(values):
+            raise ValueError("Timestamp and value counts differ")
+        count = min(len(times), max_size)
+        origin = times[len(times) - count] if count else 0.0
+        t = (ctypes.c_double * count)(*(value - origin for value in times[-count:])) if count else (ctypes.c_double * 0)()
+        v = (ctypes.c_double * count)(*values[-count:]) if count else (ctypes.c_double * 0)()
+        capacity = max_size // 2 + 1
+        output, meta = (ctypes.c_double * capacity)(), (ctypes.c_double * 3)()
+        status = self._analyze(t, v, count, max_size, int(window), int(filter),
+            low_hz, high_hz, remove_mean, output, capacity, meta)
+        if status:
+            message = {1: "At least 16 samples are required", 2: "Invalid spectrum input",
+                3: "Gaps, duplicate or irregular timestamps (>50% interval deviation)",
+                4: f"Filter frequencies must be between 0 and Fs/2 ({meta[1] / 2:g} Hz)",
+                5: "Cannot allocate FFT workspace"}
+            raise ValueError(message.get(status, "FFT failed"))
+        n = int(meta[0])
+        return Spectrum(n, meta[1], meta[2], tuple(output[:n // 2 + 1]))

python/protocan/trends.py

@@ -0,0 +1,192 @@
+"""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")
+    root = json.loads(text.removeprefix("\ufeff"))
+    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)