Commit f1ee9876 authored by Federico Vaga's avatar Federico Vaga

Merge remote-tracking branch 'sw/develop' into develop

parents afa65e47 4def3e9f
......@@ -12,4 +12,15 @@ transcript
work/
NullFile
*.orig
*.html
\ No newline at end of file
*.html
*.o
*.ko
*~
*.mod.c
/.tmp_versions
.*cmd
modules.order
Module.symvers
*TAGS
GPATH
Makefile.specific
......@@ -19,4 +19,6 @@
[submodule "hdl/ip_cores/svec"]
path = hdl/ip_cores/svec
url = https://ohwr.org/project/svec.git
[submodule "zio"]
path = software/zio
url = git://ohwr.org/misc/zio.git
..
SPDX-License-Identifier: CC-0.0
SPDX-FileCopyrightText: 2019 CERN
=========
Changelog
=========
Unreleased
==========
Changed
-------
- drv,lib: API change fmctdc_buffer_mode() -> fmctdc_transfer_mode()
This diff is collapsed.
"""
SPDX-License-Identifier: GPL-3.0-or-later
SPDX-FileCopyrightText: 2020 CERN
"""
import pytest
import subprocess
import time
import re
import os
from PyFmcTdc import FmcTdc
class PulseGenerator(object):
def __init__(self, id):
self.id = id
def disable(self, ch):
pass
def generate_pulse(self, ch, rel_time_us,
period_ns, count, sync):
pass
class SCPI(PulseGenerator):
def __init__(self, scpi_id):
super(SCPI, self).__init__(scpi_id)
import pyvisa
self.mgr = pyvisa.ResourceManager()
self.instr = self.mgr.open_resource(self.id)
self.instr.query_delay=0
self.instr.timeout = 10000
self.instr.read_termination = '\n'
self.instr.write_termination = '\n'
self.instr.write("*RST")
self.instr.query_ascii_values("*OPC?")
self.instr.write("*CLS")
self.instr.write("INITIATE:CONTINUOUS OFF")
self.instr.write("OUTPUT:STATE OFF")
def disable(self, ch):
self.instr.write("OUTPUT:STATE OFF")
def generate_pulse(self, ch, rel_time_us,
period_ns, count, sync):
self.instr.write("OUTPUT:STATE OFF")
# START Custom Agilent 33600A commands
self.instr.write("SOURCE:BURST:STATE OFF")
# END Custom Agilent 33600A commands
self.instr.write("SOURCE:VOLTAGE:LEVEL:IMMEDIATE:AMPLITUDE 2.5V")
self.instr.write("SOURCE:VOLTAGE:LEVEL:IMMEDIATE:OFFSET 1.25V")
self.instr.write("SOURCE:FUNCTION:SHAPE PULSE")
self.instr.write("SOURCE:PULSE:WIDTH 101ns")
self.instr.write("SOURCE:PULSE:PERIOD {:d}ns".format(period_ns))
# START Custom Agilent 33600A commands
self.instr.write("TRIGGER:DELAY {:d}e-6".format(rel_time_us))
burst_period_ns = int(count/(1/period_ns)) + 500
self.instr.write("SOURCE:BURST:INTERNAL:PERIOD {:d}ns".format(burst_period_ns))
self.instr.write("SOURCE:BURST:NCYCLES {:d}".format(count))
self.instr.write("SOURCE:BURST:STATE ON")
# END Custom Agilent 33600A commands
self.instr.write("OUTPUT:STATE ON")
self.instr.query_ascii_values("*OPC?")
self.instr.write("INITIATE:IMMEDIATE")
if sync:
self.instr.query_ascii_values("*OPC?")
class FmcFineDelay(PulseGenerator):
CHANNEL_NUMBER = 4
def __init__(self, fd_id):
super(FmcFineDelay, self).__init__(fd_id)
def disable(self, ch):
cmd = ["/usr/local/bin/fmc-fdelay-pulse",
"-d", "0x{:x}".format(self.id),
"-o", str(ch),
"-m", "disable",
]
proc = subprocess.Popen(cmd)
proc.wait()
def generate_pulse(self, ch, rel_time_us,
period_ns, count, sync):
cmd = ["/usr/local/bin/fmc-fdelay-pulse",
"-d", "0x{:x}".format(self.id),
"-o", str(ch),
"-m", "pulse",
"-r", "{:d}u".format(rel_time_us),
"-T", "{:d}n".format(period_ns),
"-c", str(count),
"-t"
]
proc = subprocess.Popen(cmd)
proc.wait()
if sync:
time.sleep(1 + 2 * (period_ns * count) / 1000000000.0)
@pytest.fixture(scope="module")
def fmcfd():
if pytest.fd_id is not None:
gen = FmcFineDelay(pytest.fd_id)
elif pytest.scpi is not None:
gen = SCPI(pytest.scpi)
yield gen
if isinstance(gen, FmcFineDelay):
for ch in range(FmcFineDelay.CHANNEL_NUMBER):
gen.disable(ch + 1)
elif isinstance(gen, SCPI):
gen.disable(0)
@pytest.fixture(scope="function")
def fmctdc():
tdc = FmcTdc(pytest.tdc_id)
for ch in tdc.chan:
ch.enable = False
ch.termination = False
ch.timestamp_mode = "post"
ch.flush()
yield tdc
def pytest_addoption(parser):
parser.addoption("--tdc-id", type=lambda x : int(x, 16),
required=True, help="Fmc TDC Linux Identifier")
parser.addoption("--fd-id", type=lambda x : int(x, 16), default=None,
help="Fmc Fine-Delay Linux Identifier")
parser.addoption("--scpi", type=str, default=None,
help="SCPI Connection String")
parser.addoption("--dump-range", type=int, default=10,
help="Timestamps to show before and after an error")
parser.addoption("--channel", type=int, default=[],
action="append", choices=range(FmcTdc.CHANNEL_NUMBER),
help="Channel(s) to be used for acquisition tests. Default all channels")
parser.addoption("--usr-acq-count", type=int, default=0,
help="Number of pulses to generate during a acquisition test.")
parser.addoption("--usr-acq-period-ns", type=int, default=0,
help="Pulses period (ns) during a acquisition test.")
def pytest_configure(config):
pytest.tdc_id = config.getoption("--tdc-id")
pytest.fd_id = config.getoption("--fd-id")
pytest.scpi = config.getoption("--scpi")
if pytest.scpi is None and pytest.fd_id is None:
raise Exception("You must set --fd-id or --scpi")
pytest.channels = config.getoption("--channel")
if len(pytest.channels) == 0:
pytest.channels = range(FmcTdc.CHANNEL_NUMBER)
if len(pytest.channels) != 1 and pytest.scpi is not None:
raise Exception("With --scpi we can test only the channel connected to the Waveform generator. Set --channel")
pytest.usr_acq = (config.getoption("--usr-acq-period-ns"),
config.getoption("--usr-acq-count"))
pytest.dump_range = config.getoption("--dump-range")
pytest.transfer_mode = None
with open("/sys/bus/zio/devices/tdc-1n5c-{:04x}/transfer-mode".format(pytest.tdc_id)) as f_mode:
mode = int(f_mode.read().rstrip())
for k, v in FmcTdc.TRANSFER_MODE.items():
if mode == v:
pytest.transfer_mode = k
pytest.carrier = None
full_path = os.readlink("/sys/bus/zio/devices/tdc-1n5c-{:04x}".format(pytest.tdc_id))
for carr in ["spec", "svec"]:
is_carr = re.search(carr, full_path)
if is_carr is not None:
pytest.carrier = carr
# SPDX-License-Identifier: GPL-3.0-or-later
# SPDX-FileCopyrightText: 2020 CERN
[pytest]
addopts = -v -p no:cacheprovider
\ No newline at end of file
"""
SPDX-License-Identifier: GPL-3.0-or-later
SPDX-FileCopyrightText: 2020 CERN
"""
import pytest
import random
from PyFmcTdc import FmcTdc
class TestFmctdcGetterSetter(object):
@pytest.mark.parametrize("i", range(FmcTdc.CHANNEL_NUMBER))
@pytest.mark.parametrize("term", [True, False])
def test_termination(self, fmctdc, i, term):
"""Set temination and read it back"""
fmctdc.chan[i].termination = term
assert term == fmctdc.chan[i].termination
@pytest.mark.parametrize("i", range(FmcTdc.CHANNEL_NUMBER))
@pytest.mark.parametrize("term", [True, False])
def test_termination(self, fmctdc, i, term):
"""Set temination and read it back"""
fmctdc.chan[i].termination = term
assert term == fmctdc.chan[i].termination
# TODO vmalloc error EBUSY
# @pytest.mark.parametrize("buffer_type", FmcTdc.BUFFER_TYPE.keys())
# def test_buffer_type(self, fmctdc, buffer_type):
# """Set buffer type and read it back"""
# fmctdc.buffer_type = buffer_type
# assert buffer_type == fmctdc.buffer_type
def test_transfer_mode(self, fmctdc):
"""Set buffer type and read it back"""
assert fmctdc.transfer_mode in FmcTdc.TRANSFER_MODE
@pytest.mark.parametrize("i", range(FmcTdc.CHANNEL_NUMBER))
@pytest.mark.parametrize("term", [True, False])
def test_termination(self, fmctdc, i, term):
"""Set temination and read it back"""
fmctdc.chan[i].termination = term
assert term == fmctdc.chan[i].termination
@pytest.mark.parametrize("ch", range(FmcTdc.CHANNEL_NUMBER))
@pytest.mark.parametrize("enable", [True, False])
def test_enable(self, fmctdc, ch, enable):
"""Set enable status and read it back"""
"""When a channel gets enabled only that one become enabled all
the other remains disables"""
fmctdc.chan[ch].enable = True
for i in range(FmcTdc.CHANNEL_NUMBER):
assert fmctdc.chan[i].enable == (i == ch)
fmctdc.chan[ch].enable = False
for i in range(FmcTdc.CHANNEL_NUMBER):
assert fmctdc.chan[i].enable == False
@pytest.mark.parametrize("i", range(FmcTdc.CHANNEL_NUMBER))
@pytest.mark.parametrize("buffer_mode", FmcTdc.FmcTdcChannel.BUFFER_MODE.keys())
def test_buffer_mode(self, fmctdc, i, buffer_mode):
"""Set buffer mode and read it back"""
fmctdc.chan[i].buffer_mode = buffer_mode
assert buffer_mode == fmctdc.chan[i].buffer_mode
# TODO vmalloc problems first
# @pytest.mark.parametrize("i", range(FmcTdc.CHANNEL_NUMBER))
# @pytest.mark.parametrize("buffer_len", range(1, 10))
# def test_buffer_len(self, fmctdc, i, buffer_len):
# """Set buffer length and read it back"""
# fmctdc.chan[i].buffer_len = buffer_len
# assert buffer_len == fmctdc.chan[i].buffer_len
@pytest.mark.parametrize("i", range(FmcTdc.CHANNEL_NUMBER))
def test_fileno(self, fmctdc, i):
"""file descriptors are always positive numbers"""
assert 0 < fmctdc.chan[i].fileno
@pytest.mark.parametrize("i", range(FmcTdc.CHANNEL_NUMBER))
@pytest.mark.parametrize("offset", random.sample(range(1000000), 10))
def test_offset(self, fmctdc, i, offset):
"""Set user offset and read it back"""
fmctdc.chan[i].offset = offset
assert offset == fmctdc.chan[i].offset
@pytest.mark.parametrize("i", range(FmcTdc.CHANNEL_NUMBER))
def test_stats(self, fmctdc, i):
"""Set user offset and read it back"""
st = fmctdc.chan[i].stats
@pytest.mark.parametrize("i", range(FmcTdc.CHANNEL_NUMBER))
@pytest.mark.parametrize("timestamp_mode",
FmcTdc.FmcTdcChannel.TIMESTAMP_MODE.keys())
def test_timestamp_mode(self, fmctdc, i, timestamp_mode):
"""Set timestamp mode and read it back"""
fmctdc.chan[i].timestamp_mode = timestamp_mode
assert timestamp_mode == fmctdc.chan[i].timestamp_mode
"""
SPDX-License-Identifier: GPL-3.0-or-later
SPDX-FileCopyrightText: 2020 CERN
"""
import pytest
import random
import select
import time
import sys
import os
from PyFmcTdc import FmcTdc, FmcTdcTime
TDC_FD_CABLING = [1, 2, 3, 4, 4]
min_period = 5 if pytest.transfer_mode == "fifo" else 4
fmctdc_acq_100ms = [(200, 65000),
(250, 65000),
(500, 65000),
(1000, 65000),
(1700, 65000),
# Let's keep the test within 100ms duration
# vvvvvvvvvvv
(1875, 60000),
(2500, 40000),
(5000, 20000),
(10000, 10000),
(100000, 1000),
(1000000, 100),
(10000000, 10)]
@pytest.fixture(scope="function", params=pytest.channels)
def fmctdc_chan(request):
tdc = FmcTdc(pytest.tdc_id)
for ch in tdc.chan:
ch.enable = False
tdc.chan[request.param].termination = False
tdc.chan[request.param].timestamp_mode = "post"
tdc.chan[request.param].flush()
tdc.chan[request.param].coalescing_timeout = 1
tdc.whiterabbit_mode = False
tdc.time = FmcTdcTime(0, 0, 0, 0, 0)
tdc.chan[request.param].enable = True
yield tdc.chan[request.param]
tdc.chan[request.param].enable = False
del tdc
class TestFmctdcAcquisition(object):
def test_acq_single_channel_disable(self, fmctdc_chan, fmcfd):
"""Acquistion does not start if the channel is not enable"""
fmctdc_chan.enable = False
fmcfd.generate_pulse(TDC_FD_CABLING[fmctdc_chan.idx], 0, 1000000000, 1, True)
with pytest.raises(OSError):
ts = fmctdc_chan.read(1, os.O_NONBLOCK)
@pytest.mark.parametrize("period_ns,count", fmctdc_acq_100ms)
def test_acq_chan_stats(self, fmctdc_chan, fmcfd, period_ns, count):
"""Check that unders a controlled acquisiton statistics increase
correctly. Test 100 milli-second acquisition at different
frequencies"""
stats_before = fmctdc_chan.stats
fmctdc_chan.buffer_len = max(count + 1, 64)
fmcfd.generate_pulse(TDC_FD_CABLING[fmctdc_chan.idx], 1000,
period_ns, count, True)
stats_after = fmctdc_chan.stats
assert stats_before[0] + count == stats_after[0]
@pytest.mark.parametrize("period_ns,count", fmctdc_acq_100ms)
def test_acq_chan_read_count(self, fmctdc_chan, fmcfd, period_ns, count):
"""Check that unders a controlled acquisiton the number of read
timestamps is correct. Test 100 milli-second acquisition at different
frequencies"""
fmctdc_chan.buffer_len = max(count + 1, 64)
fmcfd.generate_pulse(TDC_FD_CABLING[fmctdc_chan.idx], 1000,
period_ns, count, True)
ts = fmctdc_chan.read(count, os.O_NONBLOCK)
assert len(ts) == count
@pytest.mark.parametrize("period_ns,count", fmctdc_acq_100ms)
def test_acq_timestamp_seq_num(self, fmctdc_chan, fmcfd, period_ns, count):
"""Check that unders a controlled acquisiton the sequence
number of each timestamps increase by 1. Test 100 milli-second
acquisition at different frequencies"""
fmctdc_chan.buffer_len = max(count + 1, 64)
prev = None
fmcfd.generate_pulse(TDC_FD_CABLING[fmctdc_chan.idx], 1000,
period_ns, count, True)
ts = fmctdc_chan.read(count, os.O_NONBLOCK)
assert len(ts) == count
for i in range(len(ts)):
if prev == None:
prev = ts[i]
continue
assert ts[i].seq_id == (prev.seq_id + 1) & 0xFFFFFFF, \
"Missed {:d} timestamps (idx: {:d}, max: {:d}, prev: {{ {:s}, curr: {:s} }}, full dump;\n{:s}".format(ts[i].seq_id - prev.seq_id + 1,
i,
len(ts),
str(prev),
str(ts[i]),
"\n".join([str(x) for x in ts[max(0, i - pytest.dump_range):min(i + pytest.dump_range, len(ts) -1)]]))
prev = ts[i]
@pytest.mark.skipif(0 in pytest.usr_acq,
reason="Missing user acquisition option")
@pytest.mark.skipif(pytest.carrier == "spec" and \
pytest.transfer_mode == "fifo" and \
pytest.usr_acq[0] < 7000,
reason="On SPEC with FIFO acquisition we can't do more than 100kHz")
@pytest.mark.parametrize("period_ns,count", [pytest.usr_acq])
def test_acq_timestamp_single_channel(self, capsys, fmctdc_chan, fmcfd,
period_ns, count):
"""Run an acquisition with users parameters for period and count.
The Fine-Delay can generate a burst of maximum 65536 pulses, so we
compute and approximated timeout to stop the test and we let
the Fine-Delay generating an infinite train of pulses.
Since the test can be very long, periodically this test will print the
timestamp sequence number, you should see it increasing.
"""
poll = select.poll()
poll.register(fmctdc_chan.fileno, select.POLLIN)
pending = count
prev = None
# be able to buffer for 1 second
fmctdc_chan.buffer_len = int(1/(period_ns/1000000000.0)) + 1
stats_o = fmctdc_chan.stats
trans_b = stats_o[1]
fmcfd.generate_pulse(TDC_FD_CABLING[fmctdc_chan.idx], 1000,
period_ns, 0, False)
timeout = time.time() + 1 + (period_ns * count) / 1000000000.0
while pending > 0:
t = time.time()
if t >= timeout:
break
ret = poll.poll(1)
if len(ret) == 0:
continue
ts = fmctdc_chan.read(1000, os.O_NONBLOCK)
assert len(ts) <= 1000
for i in range(len(ts)):
if prev == None:
prev = ts[i]
continue
assert ts[i].seq_id == (prev.seq_id + 1) & 0xFFFFFFF, \
"Missed {:d} timestamps (idx: {:d}, max: {:d}, prev: {{ {:s}, curr: {:s} }}, full dump;\n{:s}".format(ts[i].seq_id - prev.seq_id + 1,
i,
len(ts),
str(prev),
str(ts[i]),
"\n".join([str(x) for x in ts[max(0, i - pytest.dump_range):min(i + pytest.dump_range, len(ts) -1)]]))
prev = ts[i]
pending -= len(ts)
poll.unregister(fmctdc_chan.fileno)
fmcfd.disable(TDC_FD_CABLING[fmctdc_chan.idx])
assert stats_o[0] == stats_o[1]
assert fmctdc_chan.stats[0] == fmctdc_chan.stats[1]
assert fmctdc_chan.stats[0] - stats_o[0] >= count
assert pending <= 0, "Some timestamp could be missing"
"""
SPDX-License-Identifier: GPL-3.0-or-later
SPDX-FileCopyrightText: 2020 CERN
"""
import pytest
class TestFmctdcTemperature(object):
def test_temperature_read(self, fmctdc):
assert 0 < fmctdc.temperature
"""
SPDX-License-Identifier: GPL-3.0-or-later
SPDX-FileCopyrightText: 2020 CERN
"""
import pytest
import random
import time
from PyFmcTdc import FmcTdcTime
class TestFmctdcTime(object):
def test_whiterabbit_mode(self, fmctdc):
"""It must be possible to toggle the White-Rabbit status"""
fmctdc.whiterabbit_mode = True
assert fmctdc.whiterabbit_mode == True
fmctdc.whiterabbit_mode = False
assert fmctdc.whiterabbit_mode == False
def test_time_set_fail_wr(self, fmctdc):
"""Time can't be changed when White-Rabbit is enabled"""
fmctdc.whiterabbit_mode = True
with pytest.raises(OSError):
fmctdc.time = FmcTdcTime(10, 0, 0, 0, 0)
@pytest.mark.parametrize("t", random.sample(range(1000000), 10))
def test_time_set(self, fmctdc, t):
"""Time can be changed when White-Rabbit is disabled"""
fmctdc.whiterabbit_mode = False
t_base = FmcTdcTime(t, 0, 0, 0, 0)
fmctdc.time = t_base
assert t_base.seconds == fmctdc.time.seconds
@pytest.mark.parametrize("whiterabbit", [False, True])
def test_time_flows(self, fmctdc, whiterabbit):
"""Just check that the time flows more or less correctly second by
second for a minute"""
fmctdc.whiterabbit_mode = whiterabbit
for i in range(20):
t_prev = fmctdc.time
time.sleep(1)
t_diff = abs(float(fmctdc.time) - float(t_prev))
assert 0.999 < t_diff < 1.001
# include parent_common.mk for buildsystem's defines
# use absolute path for REPO_PARENT
REPO_PARENT ?= $(shell /bin/pwd)/..
-include $(REPO_PARENT)/parent_common.mk
all: kernel lib tools
DIRS = kernel lib tools
$(SPEC_SW_ABS):
kernel:
lib:
tools: lib
DESTDIR ?= /usr/local
.PHONY: all clean modules install modules_install $(DIRS)
install modules_install:
all clean modules install modules_install: $(DIRS)
clean: TARGET = clean
modules: TARGET = modules
install: TARGET = install
modules_install: TARGET = modules_install
$(DIRS):
$(MAKE) -C $@ $(TARGET)
cppcheck:
for d in $(DIRS); do $(MAKE) -C $$d cppcheck || exit 1; done
WARNING!
This is work in progress. It is buggy and can be incomplete!
Please check doc/
In ./doc/, fmc-tdc.in is the source file, and you can "make" to get
pdf and other formats provided you have the proper tools installed
(mainly texinfo and tex).
doxygen-fmctdc-output
*.pdf
_build
\ No newline at end of file
#
# Makefile for the documentation directory
#
# Copyright 1994,2000,2010,2011 Alessandro Rubini <rubini@linux.it>
#
#################
# There is not basenames here, all *.in are considered input
INPUT = $(wildcard *.in)
TEXI = $(INPUT:.in=.texi)
INFO = $(INPUT:.in=.info)
HTML = $(INPUT:.in=.html)
TXT = $(INPUT:.in=.txt)
PDF = $(INPUT:.in=.pdf)
ALL = $(PDF)
MAKEINFO ?= makeinfo
%.texi: %.in
@rm -f $@
sed -f ./infofilter $< > $@
emacs -batch --no-site-file -l fixinfo $@
chmod -w $@
%.pdf: %.texi
texi2pdf --batch $<
%.info: %.texi
$(MAKEINFO) $< -o $@
%.html: %.texi
$(MAKEINFO) --html --no-split -o $@ $<
%.txt: %.texi
$(MAKEINFO) --no-headers $< > $@
##############################################
.PHONY: all images check terse clean install
.INTERMEDIATE: $(TEXI)
# Exports for doxygen
export GIT_VERSION = $(shell cd $(src); git describe --dirty --long --tags)
export EXCLUDE_FILES = "../lib/fmctdc-lib-private.h"
export BRIEF = "API Documentation"
export OUTPUT ?= doxy-fmctdc
all: doxygen images $(ALL)
$(MAKE) terse
doxygen:
doxygen ./doxygen-fmctdc-config
images::
if [ -d images ]; then $(MAKE) -C images || exit 1; fi
check: _err.ps
gs -sDEVICE=linux -r320x200x16 $<
terse:
for n in cp fn ky pg toc tp vr aux log; do rm -f *.$$n; done
rm -f *~
clean: terse
rm -f $(ALL) $(TEXI)
rm -rf $(OUTPUT)
install:
# add the other unused targets, so the rule in ../Makefile works
modules modules_install:
PROJECT_NAME = "FMC TDC - Software"
PROJECT_NUMBER = $(GIT_VERSION)
PROJECT_BRIEF = $(BRIEF)
PROJECT_LOGO =
OUTPUT_DIRECTORY = $(OUTPUT)
CREATE_SUBDIRS = YES
TAB_SIZE = 8
OPTIMIZE_OUTPUT_FOR_C = YES
EXTRACT_STATIC = NO
CASE_SENSE_NAMES = YES
WARN_NO_PARAMDOC = YES
INPUT = ../lib readme.md
RECURSIVE = YES
EXCLUDE = $(EXCLUDE_FILES)
GENERATE_HTML = YES
GENERATE_LATEX = YES