355 lines
15 KiB
Python
355 lines
15 KiB
Python
"""Digital capture readers. Store edges, never expand captures to sample arrays.
|
|
|
|
DSView layout: DreamSourceLab/DSView libsigrok4DSL/session_driver.c.
|
|
SAL v1 layout is validated against example/digitOSC; v3 RLE layout reference:
|
|
https://github.com/nemanjan00/sigrok2sal/blob/master/SAL_FORMAT.md
|
|
Unknown binary versions fail explicitly instead of guessing their layout.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from array import array
|
|
from bisect import bisect_right
|
|
import configparser
|
|
import csv
|
|
from dataclasses import dataclass
|
|
import json
|
|
import math
|
|
from pathlib import Path
|
|
import re
|
|
import struct
|
|
import zipfile
|
|
|
|
|
|
class ImportCancelled(Exception):
|
|
pass
|
|
|
|
|
|
@dataclass
|
|
class DigitalChannel:
|
|
name: str
|
|
initial: int
|
|
edges: array
|
|
|
|
def level_at(self, time):
|
|
return self.initial ^ (bisect_right(self.edges, time) & 1)
|
|
|
|
|
|
@dataclass
|
|
class DigitalCapture:
|
|
source: str
|
|
format: str
|
|
channels: list
|
|
start: float
|
|
end: float
|
|
sample_rate: float = 0
|
|
trigger_time: float = None
|
|
|
|
|
|
def _check(cancel):
|
|
if cancel():
|
|
raise ImportCancelled()
|
|
|
|
|
|
def read_capture(path, progress=lambda value: None, cancel=lambda: False):
|
|
path = Path(path)
|
|
try:
|
|
if path.suffix.lower() == '.csv':
|
|
result = _csv(path, progress, cancel)
|
|
elif path.suffix.lower() == '.json':
|
|
result = _json(path, progress, cancel)
|
|
elif path.suffix.lower() in ('.sal', '.dsl'):
|
|
with zipfile.ZipFile(path) as archive:
|
|
result = (_sal if path.suffix.lower() == '.sal' else _dsl)(
|
|
path, archive, progress, cancel)
|
|
else:
|
|
raise ValueError('Выберите цифровую запись CSV, JSON, SAL или DSL.')
|
|
except (KeyError, struct.error, zipfile.BadZipFile, EOFError,
|
|
configparser.Error, UnicodeError) as exc:
|
|
raise ValueError('Повреждённый или неподдерживаемый файл: %s' % exc) from exc
|
|
_check(cancel)
|
|
if not result.channels or not math.isfinite(result.start) or not math.isfinite(result.end):
|
|
raise ValueError('Файл не содержит цифровой записи.')
|
|
progress(100)
|
|
return result
|
|
|
|
|
|
def _json(path, progress, cancel):
|
|
def unique_object(pairs):
|
|
result = {}
|
|
for key, value in pairs:
|
|
if key in result:
|
|
raise ValueError('Повторяющееся поле JSON: %s.' % key)
|
|
result[key] = value
|
|
return result
|
|
|
|
_check(cancel)
|
|
with path.open(encoding='utf-8-sig') as stream:
|
|
rows = json.load(stream, object_pairs_hook=unique_object)
|
|
_check(cancel)
|
|
if not isinstance(rows, list) or not rows or not isinstance(rows[0], dict):
|
|
raise ValueError('Ожидается JSON: непустой массив объектов с Time [s] и каналами 0/1.')
|
|
names = list(rows[0])
|
|
if ('Time [s]' not in names or not 2 <= len(names) <= 65
|
|
or any(not name.strip() for name in names)):
|
|
raise ValueError('JSON должен содержать Time [s] и от 1 до 64 цифровых каналов.')
|
|
channels = [DigitalChannel(name, 0, array('d')) for name in names if name != 'Time [s]']
|
|
fields = set(names)
|
|
start = end = None
|
|
previous = []
|
|
for index, row in enumerate(rows):
|
|
if index % 4096 == 0:
|
|
_check(cancel)
|
|
progress(index * 99 // len(rows))
|
|
try:
|
|
if not isinstance(row, dict) or set(row) != fields:
|
|
raise ValueError('набор полей должен быть одинаковым во всех строках')
|
|
time = row['Time [s]']
|
|
if (type(time) not in (int, float) or not math.isfinite(time)
|
|
or (end is not None and time <= end)):
|
|
raise ValueError('время должно быть числом, строго возрастать и быть конечным')
|
|
values = [row[c.name] for c in channels]
|
|
if any(type(v) not in (int, float, bool) or v not in (0, 1) for v in values):
|
|
raise ValueError('цифровые уровни должны быть 0 или 1')
|
|
if start is None:
|
|
start = time
|
|
for channel, value in zip(channels, values):
|
|
channel.initial = int(value)
|
|
else:
|
|
for channel, old, value in zip(channels, previous, values):
|
|
if old != value:
|
|
channel.edges.append(time)
|
|
previous = values
|
|
end = time
|
|
except (ValueError, OverflowError) as exc:
|
|
raise ValueError('JSON, строка %d: %s' % (index + 1, exc)) from exc
|
|
return DigitalCapture(str(path), 'JSON', channels, start, end)
|
|
|
|
|
|
def _csv(path, progress, cancel):
|
|
size = max(1, path.stat().st_size)
|
|
consumed = 0
|
|
channels = None
|
|
previous = []
|
|
start = end = None
|
|
dsview = False
|
|
with path.open(encoding='utf-8-sig', newline='') as stream:
|
|
for number, line in enumerate(stream, 1):
|
|
consumed += len(line)
|
|
if number % 4096 == 0:
|
|
_check(cancel)
|
|
progress(min(99, int(consumed * 100 / size)))
|
|
if not line.strip():
|
|
continue
|
|
if line.lstrip().startswith((';', '#')):
|
|
dsview |= 'DSView' in line
|
|
continue
|
|
if channels is None:
|
|
delimiter = max((',', ';', '\t'), key=line.count)
|
|
names = [v.strip() for v in next(csv.reader([line], delimiter=delimiter))]
|
|
match = re.fullmatch(r'(?:time|timestamp)\s*(?:[\[(](s|ms|us|µs|ns)[\])])?', names[0], re.I)
|
|
if not match or not 2 <= len(names) <= 65 or any(not n for n in names):
|
|
raise ValueError('Ожидается CSV: Time [s], затем цифровые каналы (0/1).')
|
|
if len(set(names)) != len(names):
|
|
raise ValueError('Повторяющиеся имена каналов CSV.')
|
|
factor = {'s': 1, 'ms': 1e-3, 'us': 1e-6, 'µs': 1e-6, 'ns': 1e-9}[(match[1] or 's').lower()]
|
|
channels = [DigitalChannel(name, 0, array('d')) for name in names[1:]]
|
|
continue
|
|
row = next(csv.reader([line], delimiter=delimiter))
|
|
try:
|
|
time = float(row[0].strip().replace(',', '.')) * factor
|
|
values = [v.strip() for v in row[1:]]
|
|
if len(values) != len(channels) or any(v not in ('0', '1') for v in values):
|
|
raise ValueError('цифровые уровни должны быть 0 или 1')
|
|
if not math.isfinite(time) or (end is not None and time <= end):
|
|
raise ValueError('время должно строго возрастать и быть конечным')
|
|
except (ValueError, IndexError) as exc:
|
|
raise ValueError('CSV, строка %d: %s' % (number, exc)) from exc
|
|
if start is None:
|
|
start = time
|
|
for channel, value in zip(channels, values):
|
|
channel.initial = int(value)
|
|
else:
|
|
for i, channel in enumerate(channels):
|
|
if values[i] != previous[i]:
|
|
channel.edges.append(time)
|
|
previous = values
|
|
end = time
|
|
if start is None:
|
|
raise ValueError('CSV не содержит данных.')
|
|
return DigitalCapture(str(path), 'DSLogic CSV' if dsview else 'Logic 2 CSV', channels, start, end)
|
|
|
|
|
|
class _Binary:
|
|
def __init__(self, data):
|
|
self.data, self.pos = data, 0
|
|
|
|
def take(self, size):
|
|
if size < 0 or self.pos + size > len(self.data):
|
|
raise ValueError('Обрезанный двоичный блок.')
|
|
value = self.data[self.pos:self.pos + size]
|
|
self.pos += size
|
|
return value
|
|
|
|
def unpack(self, fmt):
|
|
return struct.unpack('<' + fmt, self.take(struct.calcsize('<' + fmt)))
|
|
|
|
|
|
def _runs(data):
|
|
pos = 0
|
|
while pos < len(data):
|
|
value = data[pos]
|
|
pos += 1
|
|
if value >= 128:
|
|
raise ValueError('Некорректный RLE SAL.')
|
|
if value & 64:
|
|
value &= 63
|
|
for _ in range(9):
|
|
if pos == len(data):
|
|
raise ValueError('Обрезанный RLE SAL.')
|
|
byte = data[pos]
|
|
pos += 1
|
|
value = (value << 7) | (byte & 127)
|
|
if not byte & 128:
|
|
break
|
|
else:
|
|
raise ValueError('Слишком длинный RLE SAL.')
|
|
yield value + 1
|
|
|
|
|
|
def _sal_channel(data, name, cancel):
|
|
reader = _Binary(data)
|
|
if reader.take(8) != b'<SALEAE>':
|
|
raise ValueError('Неверная сигнатура SAL.')
|
|
version, kind = reader.unpack('II')
|
|
if version not in (1, 3, 4) or kind != 100:
|
|
raise ValueError('Не поддерживается цифровой SAL версии %d, тип %d.' % (version, kind))
|
|
schema, rate, unix_ms, fractional_ms = reader.unpack('BdQd')
|
|
if schema != 1 or not math.isfinite(rate) or rate <= 0:
|
|
raise ValueError('Некорректный заголовок канала SAL.')
|
|
for _ in range(2):
|
|
optional, = reader.unpack('B')
|
|
if optional not in (0, 1):
|
|
raise ValueError('Неизвестный заголовок SAL.')
|
|
if optional:
|
|
reader.take(8)
|
|
count, = reader.unpack('Q')
|
|
if not 0 < count <= len(data) // 26:
|
|
raise ValueError('Неверное число блоков SAL.')
|
|
channel = DigitalChannel(name, 0, array('d'))
|
|
previous_end = None
|
|
level = 0
|
|
for chunk in range(count):
|
|
_check(cancel)
|
|
if version == 1:
|
|
begin, end, samples, chunk_rate, scale, length = reader.unpack('6Q')
|
|
if samples != end - begin or chunk_rate != rate or scale != 1:
|
|
raise ValueError('Неподдерживаемая шкала блока SAL.')
|
|
encoded = reader.take(length)
|
|
entries, = reader.unpack('Q')
|
|
if not entries:
|
|
raise ValueError('В SAL отсутствует индекс начального уровня.')
|
|
index = reader.take(entries * 20)
|
|
offset, byte_offset, initial = struct.unpack_from('<QQI', index)
|
|
if offset or byte_offset:
|
|
raise ValueError('Некорректный индекс SAL.')
|
|
else:
|
|
begin, end, initial, flags, length = reader.unpack('QQBBQ')
|
|
if flags:
|
|
raise ValueError('Неизвестные флаги SAL.')
|
|
encoded = reader.take(length)
|
|
if initial not in (0, 1) or end <= begin or (previous_end is not None and begin != previous_end):
|
|
raise ValueError('Некорректные границы или уровень блока SAL.')
|
|
if chunk == 0:
|
|
first = begin
|
|
channel.initial = initial
|
|
elif initial != level:
|
|
channel.edges.append(begin / rate)
|
|
level = initial
|
|
position = begin
|
|
for run_index, run in enumerate(_runs(encoded)):
|
|
if run_index % 8192 == 0:
|
|
_check(cancel)
|
|
position += run
|
|
if position > end:
|
|
raise ValueError('RLE выходит за границы блока SAL.')
|
|
if position < end:
|
|
channel.edges.append(position / rate)
|
|
level ^= 1
|
|
if position != end:
|
|
raise ValueError('Неполный RLE блока SAL.')
|
|
previous_end = end
|
|
if reader.pos != len(data):
|
|
raise ValueError('Неизвестные дополнительные данные SAL.')
|
|
return channel, first / rate, previous_end / rate, rate
|
|
|
|
|
|
def _sal(path, archive, progress, cancel):
|
|
meta = json.loads(archive.read('meta.json'))
|
|
names = {}
|
|
for row in meta.get('data', {}).get('rowsSettings', []):
|
|
channel = row.get('channel', {})
|
|
if channel.get('type') == 'Digital':
|
|
names[channel.get('deviceChannel')] = row.get('name')
|
|
files = sorted((int(m[1]), name) for name in archive.namelist()
|
|
for m in [re.fullmatch(r'digital-(\d+)\.bin', name)] if m)
|
|
if not files:
|
|
raise ValueError('В SAL нет цифровых каналов.')
|
|
channels, bounds = [], []
|
|
for i, (index, filename) in enumerate(files):
|
|
_check(cancel)
|
|
channel, start, end, rate = _sal_channel(archive.read(filename), names.get(index) or 'D%d' % index, cancel)
|
|
channels.append(channel)
|
|
bounds.append((start, end, rate))
|
|
progress(int((i + 1) * 100 / len(files)))
|
|
if any(bound != bounds[0] for bound in bounds):
|
|
raise ValueError('SAL с разными временными границами каналов не поддерживается.')
|
|
return DigitalCapture(str(path), 'Logic 2 SAL', channels, *bounds[0])
|
|
|
|
|
|
def _dsl(path, archive, progress, cancel):
|
|
config = configparser.ConfigParser(interpolation=None)
|
|
config.read_string(archive.read('header').decode('utf-8-sig'))
|
|
version = config.getint('version', 'version')
|
|
header = config['header']
|
|
if version not in (2, 3) or header.getint('device mode', 0) != 0:
|
|
raise ValueError('Поддерживаются цифровые сессии DSView DSL версий 2 и 3.')
|
|
match = re.fullmatch(r'\s*([\d.]+)\s*([kKmMgG]?)(?:Hz)?\s*', header['samplerate'])
|
|
if not match:
|
|
raise ValueError('Неверная частота DSL.')
|
|
rate = float(match[1]) * {'': 1, 'k': 1e3, 'm': 1e6, 'g': 1e9}[match[2].lower()]
|
|
samples = header.getint('total samples')
|
|
blocks = header.getint('total blocks')
|
|
if not math.isfinite(rate) or rate <= 0 or samples <= 0 or blocks <= 0:
|
|
raise ValueError('Некорректный размер или частота DSL.')
|
|
probes = sorted((int(m[1]), name) for key, name in header.items()
|
|
for m in [re.fullmatch(r'probe(\d+)', key)] if m)
|
|
channels = []
|
|
for i, (index, name) in enumerate(probes):
|
|
channel = DigitalChannel(name, 0, array('d'))
|
|
position, level = 0, 0
|
|
for block in range(blocks):
|
|
_check(cancel)
|
|
# DSView v3 stores enabled channels in sequential directories.
|
|
directory = i if version == 3 else index
|
|
data = archive.read('L-%d/%d' % (directory, block))
|
|
for offset, byte in enumerate(data):
|
|
if offset % 65536 == 0:
|
|
_check(cancel)
|
|
if position >= samples:
|
|
break
|
|
if position == 0:
|
|
channel.initial = level = byte & 1
|
|
changes = byte ^ ((byte << 1) & 255 | level)
|
|
while changes:
|
|
bit = (changes & -changes).bit_length() - 1
|
|
if position + bit < samples:
|
|
channel.edges.append((position + bit) / rate)
|
|
changes &= changes - 1
|
|
level = byte >> 7
|
|
position += 8
|
|
if position < samples:
|
|
raise ValueError('Обрезанные данные DSL, канал %s.' % name)
|
|
channels.append(channel)
|
|
progress(int((i + 1) * 100 / max(1, len(probes))))
|
|
return DigitalCapture(str(path), 'DSLogic DSL', channels, 0, samples / rate, rate)
|