132 lines
5.4 KiB
Python
132 lines
5.4 KiB
Python
"""ctypes binding only: no Python packet codec or protocol fallback."""
|
||
from __future__ import annotations
|
||
|
||
import csv
|
||
import ctypes as C
|
||
import os
|
||
import sys
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Capture:
|
||
samples: tuple[int, ...]
|
||
trigger_index: int
|
||
sample_rate: float
|
||
demo: bool = False
|
||
|
||
def save_csv(self, path):
|
||
with open(path, "w", newline="", encoding="utf-8") as stream:
|
||
writer = csv.writer(stream)
|
||
writer.writerow(["index", "time_s", "sample_hex", "demo"] +
|
||
["D%d" % n for n in range(16)])
|
||
for i, sample in enumerate(self.samples):
|
||
writer.writerow([i, (i-self.trigger_index)/self.sample_rate,
|
||
"%04X" % sample, int(self.demo)] +
|
||
[(sample >> n) & 1 for n in range(16)])
|
||
|
||
|
||
class NativeAnalyzer:
|
||
CONNECTING, READY, CAPTURING, DONE, ERROR = range(5)
|
||
ERROR_TEXT = {
|
||
1: "Неверный формат ответа FPGA",
|
||
2: "Ошибка контрольной суммы ответа",
|
||
3: "FPGA отклонила команду",
|
||
4: "Тайм-аут UART. При потере синхронизации нужен RESET_N и повторная настройка",
|
||
5: "Неподдерживаемая конфигурация FPGA: ожидается 16 каналов, 4096 выборок, версия 1",
|
||
6: "Недопустимые или противоречивые настройки триггера",
|
||
7: "Ошибка последовательного порта",
|
||
}
|
||
|
||
def __init__(self, library=None):
|
||
if library is None:
|
||
explicit = os.environ.get("ALTERA_LOGIC_LIBRARY")
|
||
name = "setprotocol.dll" if sys.platform == "win32" else "libsetprotocol.so"
|
||
bundled = Path(getattr(sys, "_MEIPASS", "")) / "gui_desktop/native" / name
|
||
local = Path(__file__).resolve().parent / "native" / name
|
||
path = Path(explicit) if explicit else (bundled if bundled.is_file() else local)
|
||
try:
|
||
library = C.CDLL(str(path))
|
||
except OSError as exc:
|
||
raise RuntimeError("Не загружено C-ядро Altera Logic. Соберите templates/"
|
||
"c/set-protocol/tools/build_host.py; " + str(exc)) from exc
|
||
self.lib = library
|
||
signatures = {
|
||
"la_context_size": ([], C.c_size_t),
|
||
"la_init": ([C.c_void_p], None),
|
||
"la_start": ([C.c_void_p] + [C.c_uint32]*5, C.c_int),
|
||
"la_next": ([C.c_void_p, C.c_void_p, C.c_size_t], C.c_size_t),
|
||
"la_feed": ([C.c_void_p, C.c_void_p, C.c_size_t], None),
|
||
"la_tick": ([C.c_void_p, C.c_uint32], None),
|
||
"la_fail": ([C.c_void_p, C.c_uint32], None),
|
||
"la_get": ([C.c_void_p, C.c_uint32], C.c_uint32),
|
||
"la_samples": ([C.c_void_p, C.c_void_p, C.c_size_t], C.c_size_t),
|
||
"la_demo_init": ([C.c_void_p], None),
|
||
"la_demo_capture": ([C.c_void_p, C.c_uint32], C.c_int),
|
||
}
|
||
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("C-ядро устарело: пересоберите templates с Altera Logic") from exc
|
||
# Explicitly aligned caller-owned storage; C never allocates memory.
|
||
self.ctx = (C.c_uint64 * ((self.lib.la_context_size()+7)//8))()
|
||
self.demo = False
|
||
self.reset()
|
||
|
||
def reset(self, demo=False):
|
||
self.demo = demo
|
||
(self.lib.la_demo_init if demo else self.lib.la_init)(self.ctx)
|
||
|
||
def get(self, field):
|
||
return int(self.lib.la_get(self.ctx, field))
|
||
|
||
@property
|
||
def state(self):
|
||
return self.get(0)
|
||
|
||
@property
|
||
def error(self):
|
||
return self.ERROR_TEXT.get(self.get(1), "Ошибка обмена")
|
||
|
||
@property
|
||
def progress(self):
|
||
return self.get(7)
|
||
|
||
@property
|
||
def flags(self):
|
||
return self.get(5)
|
||
|
||
def start(self, divider, mask=0, value=0, edge_mask=0, edge_value=0):
|
||
values = (divider, mask, value, edge_mask, edge_value)
|
||
if any(not isinstance(v, int) or not 0 <= v <= 65535 for v in values):
|
||
raise ValueError(self.ERROR_TEXT[6])
|
||
result = (self.lib.la_demo_capture(self.ctx, divider) if self.demo else
|
||
self.lib.la_start(self.ctx, *values))
|
||
if result:
|
||
raise ValueError(self.ERROR_TEXT.get(result, "Захват уже выполняется"))
|
||
|
||
def next_request(self):
|
||
out = (C.c_ubyte*6)()
|
||
size = self.lib.la_next(self.ctx, out, len(out))
|
||
return bytes(out[:size])
|
||
|
||
def feed(self, data):
|
||
self.lib.la_feed(self.ctx, data, len(data))
|
||
|
||
def tick(self, elapsed_ms):
|
||
self.lib.la_tick(self.ctx, max(0, min(int(elapsed_ms), 0xffffffff)))
|
||
|
||
def fail_io(self):
|
||
self.lib.la_fail(self.ctx, 7)
|
||
|
||
def capture(self):
|
||
out = (C.c_uint16*4096)()
|
||
count = self.lib.la_samples(self.ctx, out, len(out))
|
||
if not count:
|
||
raise RuntimeError("Запись ещё не завершена")
|
||
return Capture(tuple(out[:count]), self.get(6),
|
||
self.get(4)/(self.get(8)+1), self.demo)
|