109 lines
5.2 KiB
Python
109 lines
5.2 KiB
Python
"""SETCAN streaming FFI and immutable display/export models. No wire codec."""
|
|
from __future__ import annotations
|
|
import csv
|
|
import ctypes as C
|
|
from dataclasses import dataclass
|
|
from .native import NativeAnalyzer
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class StreamSnapshot:
|
|
indices: tuple[int, ...]
|
|
samples: tuple[int, ...]
|
|
breaks: tuple[int, ...]
|
|
period_ns: int
|
|
session: int
|
|
demo: bool = False
|
|
traces: tuple = ()
|
|
|
|
def save_csv(self, path):
|
|
with open(path, "w", newline="", encoding="utf-8") as file:
|
|
writer = csv.writer(file)
|
|
writer.writerow(["session", "sample_index", "time_s", "gap_before", "sample_hex", "demo"] +
|
|
["D%d" % n for n in range(16)])
|
|
for index, sample, gap in zip(self.indices, self.samples, self.breaks):
|
|
writer.writerow([self.session, index, index*self.period_ns/1e9, gap,
|
|
"%04X" % sample, int(self.demo)] +
|
|
[(sample >> bit)&1 for bit in range(16)])
|
|
|
|
|
|
class NativeStream:
|
|
def __init__(self, device_id=None):
|
|
self._owner = NativeAnalyzer()
|
|
self.lib = self._owner.lib
|
|
signatures = {
|
|
"las_context_size": ([], C.c_size_t),
|
|
"las_init": ([C.c_void_p, C.c_uint32], C.c_int),
|
|
"las_can": ([C.c_void_p, C.c_uint32, C.c_void_p, C.c_size_t, C.c_uint32, C.c_uint32], None),
|
|
"las_uart": ([C.c_void_p, C.c_void_p, C.c_size_t], None),
|
|
"las_get": ([C.c_void_p, C.c_uint32], C.c_uint32),
|
|
"las_snapshot": ([C.c_void_p, C.c_void_p, C.c_void_p, C.c_void_p, C.c_size_t], C.c_size_t),
|
|
"las_trace": ([C.c_void_p, C.c_uint32, C.c_void_p, C.c_void_p, C.c_void_p, C.c_size_t], C.c_size_t),
|
|
"las_device_type": ([], C.c_uint32),
|
|
"las_device_id": ([], C.c_uint32),
|
|
"las_device_name": ([], C.c_char_p),
|
|
"las_demo_step": ([C.c_void_p, C.c_uint32], None),
|
|
"las_metadata": ([C.c_uint32]*4+[C.POINTER(C.c_uint32), C.c_void_p, C.c_size_t], C.c_size_t),
|
|
"las_data": ([C.c_uint32]*5+[C.POINTER(C.c_uint32), C.c_void_p, C.c_size_t], C.c_size_t),
|
|
}
|
|
try:
|
|
for name, (args, result) in signatures.items():
|
|
fn = getattr(self.lib, name)
|
|
fn.argtypes, fn.restype = args, result
|
|
except AttributeError as exc:
|
|
raise RuntimeError("Обновите DLL templates: отсутствует SETCAN Altera stream") from exc
|
|
self.device_type = int(self.lib.las_device_type())
|
|
self.device_id = int(self.lib.las_device_id()) if device_id is None else device_id
|
|
self.device_name = self.lib.las_device_name().decode("utf-8")
|
|
self.ctx = (C.c_uint64*((self.lib.las_context_size()+7)//8))()
|
|
if not 0 <= self.device_id <= 15 or not self.lib.las_init(self.ctx, self.device_id):
|
|
raise ValueError("DeviceID должен быть 0…15")
|
|
|
|
@property
|
|
def stats(self):
|
|
fields = ("count", "period_ns", "session", "received", "missing", "duplicates",
|
|
"invalid", "ignored", "crc_errors", "has_meta")
|
|
return {name: int(self.lib.las_get(self.ctx, i)) for i, name in enumerate(fields)}
|
|
|
|
def feed_can(self, identifier, data, extended=True, remote=False):
|
|
if not 0 <= identifier <= 0xffffffff:
|
|
return
|
|
self.lib.las_can(self.ctx, identifier, data, len(data), int(extended), int(remote))
|
|
|
|
def feed_uart(self, data):
|
|
self.lib.las_uart(self.ctx, data, len(data))
|
|
|
|
def snapshot(self, demo=False):
|
|
stats = self.stats
|
|
count = stats["count"]
|
|
indices, samples, breaks = (C.c_uint64*count)(), (C.c_uint16*count)(), (C.c_ubyte*count)()
|
|
n = self.lib.las_snapshot(self.ctx, indices, samples, breaks, count)
|
|
traces = []
|
|
xs, ys, moves = (C.c_uint64*(2*count))(), (C.c_ubyte*(2*count))(), (C.c_ubyte*(2*count))()
|
|
for channel in range(16):
|
|
size = self.lib.las_trace(self.ctx, channel, xs, ys, moves, 2*count)
|
|
traces.append(tuple(zip(xs[:size], ys[:size], moves[:size])))
|
|
return StreamSnapshot(tuple(indices[:n]), tuple(samples[:n]), tuple(breaks[:n]),
|
|
stats["period_ns"], stats["session"], demo, tuple(traces))
|
|
|
|
def demo_step(self, count=50):
|
|
self.lib.las_demo_step(self.ctx, count)
|
|
|
|
def metadata_packet(self, session, period_ns, uart=False):
|
|
if not 0 <= session <= 65535 or not 1 <= period_ns <= 0xffffffff:
|
|
raise ValueError("Invalid stream metadata")
|
|
return self._packet(self.lib.las_metadata, session, period_ns, int(uart))
|
|
|
|
def data_packet(self, session, index, sample, uart=False):
|
|
if not 0 <= index <= 0xffffffff or any(not 0 <= v <= 65535 for v in (session, sample)):
|
|
raise ValueError("Invalid stream samples")
|
|
return self._packet(self.lib.las_data, session, index, sample, int(uart))
|
|
|
|
def _packet(self, fn, *args):
|
|
identifier = C.c_uint32()
|
|
out = (C.c_ubyte*32)()
|
|
n = fn(self.device_id, *args, C.byref(identifier), out, len(out))
|
|
if not n:
|
|
raise ValueError("Invalid SETCAN packet arguments")
|
|
return identifier.value, bytes(out[:n])
|