Merge branch 'developer' into dev/rx_disable_datastream_port
Build on RHEL9 docker image / build (push) Successful in 3m37s
Build on RHEL8 docker image / build (push) Successful in 5m4s
Run Simulator Tests on local RHEL9 / build (push) Failing after 6m0s
Run Simulator Tests on local RHEL8 / build (push) Failing after 7m51s

This commit is contained in:
2026-05-11 12:58:13 +02:00
15 changed files with 141 additions and 1273 deletions
+10 -42
View File
@@ -101,7 +101,7 @@ To run only tests requiring virtual detectors use the following command:
#in build
python -m pytest -m detectorintegration ../python/tests/
There is a helper test fixture in ``slsDetectorSoftware/python/tests/conftest.py`` called ``test_with_simulators`` that sets up virtual detectors and yields the test for all detectors. The set up is done for every test automatically.
There is a helper test fixture in ``slsDetectorSoftware/python/tests/conftest.py`` called ``session_simulator`` that sets up virtual detectors and yields the test for all detectors. The set up is done for every test automatically. Note that the fixture persist over the entire session e.g. the fixture is setup one detector at a time and runs all tests using this fixture before cleaning up and moving on to the next detector. It saves time if the setup and cleanup is expensive.
Example usage:
@@ -110,10 +110,13 @@ Example usage:
import pytest
@pytest.mark.detectorintegration
def test_example_with_simulator(test_with_simulators):
def test_example_with_simulator(session_simulator):
# your test code here
If you want to run the test only for a specific test use the parametrized test fixture:
.. Note::
As the detector is set up only once makes sure to not change the state of the detector in a way that affects other tests. If you want to change the state of the detector make sure to reset it at the end of your test.
If you want to run the test only for a specific detector use the parametrized test fixture:
Example usage:
@@ -122,45 +125,10 @@ Example usage:
import pytest
@pytest.mark.detectorintegration
@pytest.mark.parametrize("setup_parameters", [(["<my_detector>"], <num_modules>)], indirect=True)
def test_example_with_specific_simulators(test_with_simulators, setup_parameters):
@pytest.mark.parametrize("session_simulator", [("<my_detector>", <num_interfaces>, <num_modules>), ("<another_detector>", <num_interfaces>, <num_modules>)], indirect=True)
def test_example_with_specific_simulators(session_simulator):
# your test code here
.. Note::
The parametrized test fixture is setup per file and not for the entire session.
There is another helper test fixture in ``slsDetectorSoftware/python/tests/conftest.py`` called ``session_simulator`` that sets up virtual detectors and yields the test for all detectors. The difference with the previous fixture ``test_with_simulators`` is that this fixture will set up one detector at a time and run all the tests using this fixture before cleaning up and moving on to the next detector. It saves time if the setup and cleanup is expensive.
Example usage:
.. code-block:: python
import pytest
@pytest.mark.detectorintegration
def test_define_reg(session_simulator, request):
""" Test setting define_reg for ctb and xilinx_ctb."""
det_type, num_interfaces, num_mods, d = session_simulator
assert d is not None
from slsdet import RegisterAddress
if det_type in ['ctb', 'xilinx_ctb']:
# your test code here
For more specific parameters, you can parametrize the fixture like below:
.. code-block:: python
import pytest
@pytest.mark.detectorintegration
@pytest.mark.parametrize(
"session_simulator",
[
("ctb", 1, 1),
("xilinx_ctb", 1, 1),
],
indirect=True,
)
def test_define_reg(session_simulator):
det_type, num_interfaces, num_mods, d = session_simulator
# your test code here
-110
View File
@@ -1,110 +0,0 @@
// SPDX-License-Identifier: LGPL-3.0-or-other
// Copyright (C) 2021 Contributors to the SLS Detector Package
#include <errno.h>
#include <fcntl.h>
#include <math.h>
#include <stdarg.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/utsname.h>
#include <unistd.h>
int main(int argc, char *argv[]) {
int iarg;
char fname[10000];
uint64_t word;
int val[64];
int bit[64];
FILE *fdin;
int nb = 2;
int off = 0;
int ioff = 0;
int dr = 24;
int idr = 0;
int ib = 0;
int iw = 0;
bit[0] = 19;
bit[1] = 8;
// for (iarg=0; iarg<argc; iarg++) printf("%d %s\n",iarg, argv[iarg]);
if (argc < 2)
printf("Error: usage is %s fname [dr off b0 b1 bn]\n");
if (argc > 2)
dr = atoi(argv[2]);
if (argc > 3)
off = atoi(argv[3]);
if (argc > 4) {
for (ib = 0; ib < 64; ib++) {
if (argc > 4 + ib) {
bit[ib] = atoi(argv[4 + ib]);
nb++;
}
}
}
idr = 0;
for (ib = 0; ib < nb; ib++) {
val[ib] = 0;
}
fdin = fopen(argv[1], "rb");
if (fdin == NULL) {
printf("Cannot open input file %s for reading\n", argv[1]);
return 200;
}
while (fread((void *)&word, 8, 1, fdin)) {
// printf("%llx\n",word);
if (ioff < off)
ioff++;
else {
for (ib = 0; ib < nb; ib++) {
if (word & (1 << bit[ib]))
val[ib] |= (1 << idr);
}
idr++;
if (idr == dr) {
idr = 0;
fprintf(stdout, "%d\t", iw++);
for (ib = 0; ib < nb; ib++) {
#ifdef HEX
fprintf(stdout, "%08llx\t", val[ib]);
#else
fprintf(stdout, "%lld\t", val[ib]);
#endif
val[ib] = 0;
}
fprintf(stdout, "\n");
}
}
}
if (idr != 0) {
fprintf(stdout, "%d\t", iw++);
for (ib = 0; ib < nb; ib++) {
#ifdef HEX
fprintf(stdout, "%08llx\t", val[ib]);
#else
fprintf(stdout, "%lld\t", val[ib]);
#endif
val[ib] = 0;
}
fprintf(stdout, "\n");
}
fclose(fdin);
return 0;
}
@@ -1,4 +0,0 @@
BIT0 output0 1
BIT1 output1 1
BIT2 output2 1
BIT3 output3 1
@@ -1,97 +0,0 @@
//define signals and directions (Input, outputs, clocks)
#define output0 0
setoutput(output0);
#define output1 1
setoutput(output1);
#define output2 2
setoutput(output2);
#define output3 3
setoutput(output3);
#define input0 4
setinput(input0);
#define input1 5
setinput(input1);
#define input2 6
setinput(input2);
#define input3 7
setinput(input3);
#define PW pw()
#define SB(x) setbit(x)
#define CB(x) clearbit(x)
#define CLOCK clearbit(CLKBIT); pw();setbit(CLKBIT);pw()
#define LCLOCK clearbit(CLKBIT); pw();setbit(CLKBIT);pw();clearbit(CLKBIT); pw()
#define CLOCKS(x) for (i=0;i<x;i++) {clearbit(CLKBIT);pw(); setbit(CLKBIT); pw();}
#define STOP setstop();
#define START setstart();
#define REPEAT(x) for (i=0;i<(x);i++) {pw();}
#define DOFOR(x) for (j=0;j<(x);j++) {
// }
#define INST0 CB(output3);CB(output2);CB(output1);CB(output0);PW;
#define INST1 CB(output3);CB(output2);CB(output1);SB(output0);PW;
#define INST2 CB(output3);CB(output2);SB(output1);CB(output0);PW;
#define INST3 CB(output3);CB(output2);SB(output1);SB(output0);PW;
#define INST4 CB(output3);SB(output2);CB(output1);CB(output0);PW;
#define INST5 CB(output3);SB(output2);CB(output1);SB(output0);PW;
#define INST6 CB(output3);SB(output2);SB(output1);CB(output0);PW;
#define INST7 CB(output3);SB(output2);SB(output1);SB(output0);PW;
#define INST8 SB(output3);CB(output2);CB(output1);CB(output0);PW;
#define INST9 SB(output3);CB(output2);CB(output1);SB(output0);PW;
#define INST10 SB(output3);CB(output2);SB(output1);CB(output0);PW;
#define INST11 SB(output3);CB(output2);SB(output1);SB(output0);PW;
#define INST12 SB(output3);SB(output2);CB(output1);CB(output0);PW;
START;
INST0;
setwaitpoint(0);
setwaittime(0,5);
INST1;
setstartloop(5);
setnloop(5,2);
INST2;
setstartloop(0);
setnloop(0,2);
INST3;
INST4;
setstoploop(0);
setstoploop(5);
INST5;
INST6;
setwaitpoint(4);
setwaittime(1,0);
INST7;
INST8;
setstartloop(2);
setnloop(2,0);
INST9;
INST10;
setstoploop(2);
INST11;
STOP;
INST12;
@@ -1,25 +0,0 @@
patword 0x0000 0x0000000000000000
patword 0x0001 0x0000000000000001
patword 0x0002 0x0000000000000002
patword 0x0003 0x0000000000000003
patword 0x0004 0x0000000000000004
patword 0x0005 0x0000000000000005
patword 0x0006 0x0000000000000006
patword 0x0007 0x0000000000000007
patword 0x0008 0x0000000000000008
patword 0x0009 0x0000000000000009
patword 0x000a 0x000000000000000a
patword 0x000b 0x000000000000000b
patword 0x000c 0x000000000000000c
patlimits 0x0000 0x000c
patioctrl 0x000000000000000f
patloop 0 0x0003 0x0005
patnloop 0 2
patloop 2 0x0009 0x000b
patnloop 2 0
patloop 5 0x0002 0x0005
patnloop 5 2
patwait 0 0x0001
patwaittime 0 5
patwait 4 0x0007
patwaittime 4 0
Binary file not shown.

Before

Width:  |  Height:  |  Size: 299 KiB

-34
View File
@@ -1,34 +0,0 @@
# SPDX-License-Identifier: LGPL-3.0-or-other
# Copyright (C) 2021 Contributors to the SLS Detector Package
if [ "$#" -eq 0 ]; then
echo "Wrong number of arguments: usage should be $0 patname"
exit 1
fi
infile=$1
outfile=$infile"at"
outfilebin=$infile"bin"
if [ "$#" -ge 2 ]; then
outfile=$2
fi
exe=$infile"exe"
if [ "$#" -ge 4 ]; then
exe=$4
fi
if [ "$#" -ge 3 ]; then
outfilebin=$3
fi
if [ -f "$infile" ]
then
dir=$(dirname $infile)
gcc -DINFILE="\"$infile\"" -DOUTFILE="\"$outfile\"" -DOUTFILEBIN="\"$outfilebin\"" -o $exe generator.c -I$dir;
echo compiling
echo gcc -DINFILE="\"$infile\"" -DOUTFILE="\"$outfile\"" -DOUTFILEBIN="\"$outfilebin\"" -o $exe generator.c -I$dir;
$exe ;
echo cleaning
rm $exe
echo done
else
echo "$infile not found."
fi
-183
View File
@@ -1,183 +0,0 @@
// SPDX-License-Identifier: LGPL-3.0-or-other
// Copyright (C) 2021 Contributors to the SLS Detector Package
/****************************************************************************
usage to generate a patter test.pat from test.p
gcc -DINFILE="\"test.p\"" -DOUTFILE="\"test.pat\"" -o test.exe generator.c ;
./test.exe ; rm test.exe
*************************************************************************/
#include <errno.h>
#include <fcntl.h>
#include <math.h>
#include <stdarg.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/utsname.h>
#include <unistd.h>
#define MAXLOOPS 6
#define MAXTIMERS 6
#define MAXWORDS 8191
uint64_t pat = 0;
uint64_t iopat = 0;
uint64_t clkpat = 0;
unsigned iaddr = 0;
unsigned waitaddr[MAXTIMERS] = {MAXWORDS, MAXWORDS, MAXWORDS,
MAXWORDS, MAXWORDS, MAXWORDS};
unsigned startloopaddr[MAXLOOPS] = {MAXWORDS, MAXWORDS, MAXWORDS,
MAXWORDS, MAXWORDS, MAXWORDS};
unsigned stoploopaddr[MAXLOOPS] = {MAXWORDS, MAXWORDS, MAXWORDS,
MAXWORDS, MAXWORDS, MAXWORDS};
unsigned start = 0, stop = 0;
uint64_t waittime[MAXTIMERS] = {0, 0, 0, 0, 0, 0};
unsigned nloop[MAXLOOPS] = {0, 0, 0, 0, 0, 0};
char infile[10000], outfile[10000];
FILE *fd, *fd1;
uint64_t PAT[MAXWORDS];
int iopat_enable = 0;
int i, ii, iii, j, jj, jjj, pixx, pixy, memx, memy, muxout, memclk, colclk,
rowclk, muxclk, memcol, memrow, loopcounter;
void setstart() { start = iaddr; }
void setstop() { stop = iaddr; }
void setinput(int bit) {
uint64_t mask = 1;
mask = mask << bit;
iopat &= ~mask;
iopat_enable = 1;
}
void setoutput(int bit) {
uint64_t mask = 1;
mask = mask << bit;
iopat |= mask;
iopat_enable = 1;
}
void clearbit(int bit) {
uint64_t mask = 1;
mask = mask << bit;
pat &= ~mask;
}
void setbit(int bit) {
uint64_t mask = 1;
mask = mask << bit;
pat |= mask;
}
int checkbit(int bit) {
uint64_t mask = 1;
mask = mask << bit;
return (pat & mask) >> bit;
}
void setstartloop(int iloop) {
if (iloop >= 0 && iloop < MAXLOOPS) {
startloopaddr[iloop] = iaddr;
}
}
void setstoploop(int iloop) {
if (iloop >= 0 && iloop < MAXLOOPS) {
stoploopaddr[iloop] = iaddr;
}
}
void setnloop(int iloop, int n) {
if (iloop >= 0 && iloop < MAXLOOPS) {
nloop[iloop] = n;
}
}
void setwaitpoint(int iloop) {
if (iloop >= 0 && iloop < MAXTIMERS) {
waitaddr[iloop] = iaddr;
}
}
void setwaittime(int iloop, uint64_t t) {
if (iloop >= 0 && iloop < MAXTIMERS) {
waittime[iloop] = t;
}
}
void pw() {
if (iaddr < MAXWORDS) {
PAT[iaddr] = pat;
}
fprintf(fd, "patword 0x%04x 0x%016llx\n", iaddr, pat);
iaddr++;
if (iaddr >= MAXWORDS) {
printf("ERROR: too many word in the pattern (%d instead of %d)!", iaddr,
MAXWORDS);
}
}
int parseCommand(int clk, int cmdbit, int cmd, int length) {
int ibit;
clearbit(clk);
for (ibit = 0; ibit < length; ibit++) {
if (cmd & (1 >> ibit)) {
setbit(cmdbit);
} else {
clearbit(cmdbit);
}
pw();
/******/
setbit(clk);
pw();
/******/
}
};
int main() {
int iloop = 0;
fd = fopen(OUTFILE, "w");
#include INFILE
fprintf(fd, "patlimits 0x%04x 0x%04x\n", start, stop);
if (iopat_enable == 1) {
fprintf(fd, "patioctrl 0x%016llx\n", iopat);
}
for (iloop = 0; iloop < MAXLOOPS; iloop++) {
if ((startloopaddr[iloop] != MAXWORDS) &&
(stoploopaddr[iloop] != MAXWORDS)) {
fprintf(fd, "patloop %d 0x%04x 0x%04x\n", iloop,
startloopaddr[iloop], stoploopaddr[iloop]);
if (stoploopaddr[iloop] <= startloopaddr[iloop]) {
nloop[iloop] = 0;
}
fprintf(fd, "patnloop %d %u\n", iloop, nloop[iloop]);
}
}
for (iloop = 0; iloop < MAXTIMERS; iloop++) {
if (waitaddr[iloop] != MAXWORDS) {
fprintf(fd, "patwait %d 0x%04x\n", iloop, waitaddr[iloop]);
fprintf(fd, "patwaittime %d %llu\n", iloop, waittime[iloop]);
}
}
fclose(fd);
fd1 = fopen(OUTFILEBIN, "w");
fwrite(PAT, sizeof(uint64_t), iaddr, fd1);
fclose(fd1);
return 0;
}
-610
View File
@@ -1,610 +0,0 @@
#!/usr/bin/env python3
"""
Created on Wed May 24 09:44:53 2017
Plot the pattern for New Chip Test Box (.pat)
Changes:
- 2017-11-21 Adapt it to python-3
- 2017-09-25 All can be plotted
- 2017-09-22 Can be plotted but the loop and wait not work yet
@author: zhang_j1
"""
import matplotlib.pyplot as plt
from numpy import *
from matplotlib.pyplot import *
from matplotlib.patches import Rectangle
import os
import argparse
###############################################################################
# COLORS AND LINE STYLES
# alternating colors of the plots (2 needed)
colors_plot = ['tab:blue', 'tab:orange']
# Wait colors and line styles (6 needed from 0 to 5)
colors_wait = ['b', 'g', 'r', 'c', 'm', 'y']
linestyles_wait = ['--', '--', '--', '--', '--', '--']
alpha_wait = [0.5, 0.5, 0.5, 0.5, 0.5, 0.5]
alpha_wait_rect = [0.2, 0.2, 0.2, 0.2, 0.2, 0.2]
# Loop colors and line styles (6 needed from 0 to 5)
colors_loop = ['tab:green', 'tab:red', 'tab:purple', 'tab:brown', 'tab:pink', 'tab:grey']
linestyles_loop = ['-.', '-.', '-.', '-.', '-.', '-.']
alpha_loop = [0.5, 0.5, 0.5, 0.5, 0.5, 0.5]
alpha_loop_rect = [0.2, 0.2, 0.2, 0.2, 0.2, 0.2]
# Display the count of clocks
clock_vertical_lines_spacing = 1
show_clocks_number = True
###############################################################################
# Define a hex to binary function
# global definition
# base = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, A, B, C, D, E, F]
base = [str(x) for x in range(10)] + [chr(x) for x in range(ord('A'), ord('A')+6)]
# dec2bin
def dec2bin(string_num):
num = int(string_num)
mid = []
while True:
if num == 0:
break
num, rem = divmod(num, 2)
mid.append(base[rem])
return ''.join([str(x) for x in mid[::-1]])
# dec2binary: better than dec2bin
def dec2binary(dec_num, width=None):
return binary_repr(int(dec_num), width=width)
# hex2dec
def hex2dec(string_num):
return str(int(string_num.upper(), 16))
# hex2bin
def hex2bin(string_num):
return dec2bin(hex2dec(string_num.upper()))
# hex2bin
def hex2binary(string_num, width=None):
return dec2binary(hex2dec(string_num.upper()), width=width)
parser = argparse.ArgumentParser()
parser.add_argument('-d', '--directory', required = True, help = "Working directory where the pattern is located")
parser.add_argument('-p', '--pattern', required = True, help = "Pattern name")
parser.add_argument('-a', '--alias', help = "Alias name")
parser.add_argument('-v', '--verbose', action='store_true')
args = parser.parse_args()
Folder = args.directory
File_pat = args.pattern
File_alias = args.alias
verbose = args.verbose
# Look at the alias file and generate the lookup table for pin names
# Create a 64 bit look up table
table = []
for i in range(64):
# for special bit
if i+1 == 59:
table.append([str(i+1), "external_trigger"])
elif i+1 == 63:
table.append([str(i+1), "adc_enable"])
elif i+1 == 62:
table.append([str(i+1), "dbit_enable"])
else:
table.append([str(i+1), ""])
# Loop all lines
try:
with open(Folder + "/" + File_alias + ".alias") as f:
lines = f.readlines()
f.close()
nlines = len(lines)
except:
nlines = 0
if nlines > 0:
for i in range(nlines):
# whether the line is bit definition
if lines[i][0:3] == "BIT":
# split words
words = lines[i].split()
bit_num = int(words[0][3:])
table[bit_num][0] = words[0][3:]
table[bit_num][1] = words[1]
else:
for i in range(64):
table[i][0] = i
table[i][1] = f'BIT#{i}'
if verbose:
print(table)
# Load the pattern and get all lines
# Loop all lines
if os.path.exists(Folder + "/" + File_pat + ".pat"):
with open(Folder + "/" + File_pat + ".pat") as f_pat:
lines_pat = f_pat.readlines()
elif os.path.exists(Folder + "/" + File_pat + ".pyat"):
with open(Folder + "/" + File_pat + ".pyat") as f_pat:
lines_pat = f_pat.readlines()
else:
print("No file found - Check it")
exit()
f_pat.close()
# number of lines for pattern file
nlines_pat = len(lines_pat)
# a counter
cnt = 0
if verbose:
print("The total number of lines of pattern:", nlines_pat)
# Loop all lines of pattern
waittime0 = None
waittime1 = None
waittime2 = None
waittime3 = None
waittime4 = None
waittime5 = None
nloop0 = None
nloop1 = None
nloop2 = None
nloop3 = None
nloop4 = None
nloop5 = None
for k in range(nlines_pat):
# content of line
words_line = lines_pat[k].split()
if words_line[0] == "patword":
# print words_line from b0 to b63
bits = hex2binary(words_line[-1], 64)[::-1]
if verbose:
print("The bits for line-", k+1, "is:", bits)
# convert string bits to decimal array
num_bits = array(list(map(str, bits)), dtype="uint16")
if cnt == 0:
mat_pat = num_bits
else:
# add bits to matrix
mat_pat = concatenate((mat_pat, num_bits), axis=0)
cnt = cnt + 1
# print("The matrix of pattern:", mat_pat.reshape(int(cnt), int(len(num_bits))))
# Look at the io: 0 for sending to ASIC, 1 for reading from ASIC
if words_line[0] == "patioctrl":
# print words_line
if verbose:
print(words_line[-1])
bits = hex2binary(words_line[-1], 64)[::-1]
if verbose:
print(bits)
# convert string bits to decimal array
out_bits = array(list(map(str, bits)), dtype="uint16")
if verbose:
print(words_line)
# Deal with waiting point
# ====== WAIT ======
if words_line[0] == "patwait" and words_line[1] == "0":
wait0 = int(hex2dec(words_line[2]))
if verbose:
print("wait 0 at:", wait0)
if words_line[0] == "patwaittime" and words_line[1] == "0":
waittime0 = int(words_line[2])
if verbose:
print("wait 0 for:", waittime0)
if words_line[0] == "patwait" and words_line[1] == "1":
wait1 = int(hex2dec(words_line[2]))
if verbose:
print("wait 1 at:", wait1)
if words_line[0] == "patwaittime" and words_line[1] == "1":
waittime1 = int(words_line[2])
if verbose:
print("wait 1 for:", waittime1)
if words_line[0] == "patwait" and words_line[1] == "2":
wait2 = int(hex2dec(words_line[2]))
if verbose:
print("wait 2 at:", wait2)
if words_line[0] == "patwaittime" and words_line[1] == "2":
waittime2 = int(words_line[2])
if verbose:
print("wait 2 for:", waittime2)
if words_line[0] == "patwait" and words_line[1] == "3":
wait3 = int(hex2dec(words_line[2]))
if verbose:
print("wait 0 at:", wait3)
if words_line[0] == "patwaittime" and words_line[1] == "3":
waittime3 = int(words_line[2])
if verbose:
print("wait 0 for:", waittime3)
if words_line[0] == "patwait" and words_line[1] == "4":
wait4 = int(hex2dec(words_line[2]))
if verbose:
print("wait 1 at:", wait4)
if words_line[0] == "patwaittime" and words_line[1] == "4":
waittime4 = int(words_line[2])
if verbose:
print("wait 1 for:", waittime4)
if words_line[0] == "patwait" and words_line[1] == "5":
wait5 = int(hex2dec(words_line[2]))
if verbose:
print("wait 2 at:", wait5)
if words_line[0] == "patwaittime" and words_line[1] == "5":
waittime5 = int(words_line[2])
if verbose:
print("wait 2 for:", waittime5)
# ====== LOOPS ======
if words_line[0] == "patloop" and words_line[1] == "0":
loop0_start = int(hex2dec(words_line[2]))
loop0_end = int(hex2dec(words_line[3]))
if verbose:
print("loop 0 start:", loop0_start, ", end:", loop0_end)
if words_line[0] == "patnloop" and words_line[1] == "0":
nloop0 = int(words_line[2])
if verbose:
print("loop 0 times:", nloop0)
if words_line[0] == "patloop" and words_line[1] == "1":
loop1_start = int(hex2dec(words_line[2]))
loop1_end = int(hex2dec(words_line[3]))
if verbose:
print("loop 1 start:", loop1_start, ", end:", loop1_end)
if words_line[0] == "patnloop" and words_line[1] == "1":
nloop1 = int(words_line[2])
if verbose:
print("loop 1 times:", nloop1)
if words_line[0] == "patloop" and words_line[1] == "2":
loop2_start = int(hex2dec(words_line[2]))
loop2_end = int(hex2dec(words_line[3]))
if verbose:
print("loop 2 start:", loop2_start, ", end:", loop2_end)
if words_line[0] == "patnloop" and words_line[1] == "2":
nloop2 = int(words_line[2])
if verbose:
print("loop 2 times:", nloop2)
if words_line[0] == "patloop" and words_line[1] == "3":
loop3_start = int(hex2dec(words_line[2]))
loop3_end = int(hex2dec(words_line[3]))
if verbose:
print("loop 3 start:", loop3_start, ", end:", loop3_end)
if words_line[0] == "patnloop" and words_line[1] == "3":
nloop3 = int(words_line[2])
if verbose:
print("loop 3 times:", nloop3)
if words_line[0] == "patloop" and words_line[1] == "4":
loop4_start = int(hex2dec(words_line[2]))
loop4_end = int(hex2dec(words_line[3]))
if verbose:
print("loop 4 start:", loop4_start, ", end:", loop4_end)
if words_line[0] == "patnloop" and words_line[1] == "4":
nloop4 = int(words_line[2])
if verbose:
print("loop 4 times:", nloop4)
if words_line[0] == "patloop" and words_line[1] == "5":
loop5_start = int(hex2dec(words_line[2]))
loop5_end = int(hex2dec(words_line[3]))
if verbose:
print("loop 5 start:", loop5_start, ", end:", loop5_end)
if words_line[0] == "patnloop" and words_line[1] == "5":
nloop5 = int(words_line[2])
if verbose:
print("loop 5 times:", nloop5)
# print(out_bits)
# internal counter
avail_index = []
avail_name = []
# Remove non-used bits
for i in range(64):
# if out_bits[0][i] == 1:
if out_bits[i] == 1:
avail_index.append(i)
avail_name.append(table[i][1])
if verbose:
print(avail_index)
print(avail_name)
# number of effective used bits
nbiteff = len(avail_name)
# subMat = mat_ext[:,index]
# print(mat_pat.shape)
subMat = mat_pat.reshape(int(cnt), int(len(num_bits)))[0:, avail_index]
# subMat = mat_pat[avail_index]
timing = linspace(0, subMat.shape[0]-1, subMat.shape[0])
rcParams['figure.figsize'] = 15, 5
# ============= PLOTTING =============
rcParams["font.weight"] = "bold"
rcParams["axes.labelweight"] = "bold"
fig2, axs2 = subplots(nbiteff, sharex='all')
subplots_adjust(wspace=0, hspace=0)
# axs2[nbiteff - 1].set(xlabel='Timing [clk]')
for idx, i in enumerate(range(nbiteff)):
axs2[idx].plot(subMat.T[i], "-", drawstyle="steps-post", linewidth=2.0, color=colors_plot[idx % 2])
x_additional = range(len(subMat.T[i]) - 1, len(subMat.T[i]) + 2)
additional_stuff = [subMat.T[i][-1]] * 3
axs2[idx].plot(x_additional, additional_stuff,
"--", drawstyle="steps-post", linewidth=2.0, color=colors_plot[idx % 2], alpha=0.5)
axs2[idx].yaxis.set_ticks([0.5], minor=False)
axs2[idx].xaxis.set_ticks(arange(0, len(subMat.T[i]) + 10, clock_vertical_lines_spacing))
axs2[idx].yaxis.set_ticklabels([avail_name[i]])
axs2[idx].get_yticklabels()[0].set_color(colors_plot[idx % 2])
axs2[idx].grid(1, 'both', 'both', alpha=0.5)
axs2[idx].yaxis.grid(which="both", color=colors_plot[idx % 2], alpha=0.2)
if idx != nbiteff - 1:
if not show_clocks_number:
axs2[idx].xaxis.set_ticklabels([])
axs2[idx].set(xlabel=' ', ylim=(-0.2, 1.2))
else:
axs2[idx].set(xlabel='Timing [clk]', ylim=(-0.2, 1.2))
# axs2[idx].set_xlim(left=0)
axs2[idx].set_xlim(left=0, right=len(subMat.T[i]) + 1)
axs2[idx].spines['top'].set_visible(False)
axs2[idx].spines['right'].set_alpha(0.2)
axs2[idx].spines['right'].set_visible(True)
axs2[idx].spines['bottom'].set_visible(False)
axs2[idx].spines['left'].set_visible(False)
# =====================================================================================================
# Plot the wait lines
# Wait 0
if waittime0 is not None:
if waittime0 == 0:
axs2[idx].plot([wait0, wait0], [-10, 10],
linestyle=linestyles_wait[0], color=colors_wait[0], alpha=alpha_wait[0], linewidth=2.0)
axs2[idx].plot([wait0 + 1, wait0 + 1], [-10, 10],
linestyle=linestyles_wait[0], color=colors_wait[0], linewidth=2.0, alpha=alpha_wait[0])
axs2[idx].add_patch(Rectangle((wait0, -10), 1, 20,
label="wait 0: skipped" if idx == 0 else "",
facecolor=colors_wait[0], alpha=alpha_wait_rect[0], hatch='\\\\'))
else:
axs2[idx].plot([wait0, wait0], [-10, 10],
linestyle=linestyles_wait[0], color=colors_wait[0],
label="wait 0: " + str(waittime0) + " clk" if idx == 0 else "",
linewidth=2.0, alpha=alpha_wait[0])
# Wait 1
if waittime1 is not None:
if waittime1 == 0:
axs2[idx].plot([wait1, wait1], [-10, 10],
linestyle=linestyles_wait[1], color=colors_wait[1], alpha=alpha_wait[1], linewidth=2.0)
axs2[idx].plot([wait1 + 1, wait1 + 1], [-10, 10],
linestyle=linestyles_wait[1], color=colors_wait[1], linewidth=2.0, alpha=alpha_wait[1])
axs2[idx].add_patch(Rectangle((wait1, -10), 1, 20,
label="wait 1: skipped" if idx == 0 else "",
facecolor=colors_wait[1], alpha=alpha_wait_rect[1], hatch='\\\\'))
else:
axs2[idx].plot([wait1, wait1], [-10, 10],
linestyle=linestyles_wait[1], color=colors_wait[1],
label="wait 1: " + str(waittime1) + " clk" if idx == 0 else "",
linewidth=2.0, alpha=alpha_wait[1])
# Wait 2
if waittime2 is not None:
if waittime2 == 0:
axs2[idx].plot([wait2, wait2], [-10, 10],
linestyle=linestyles_wait[2], color=colors_wait[2], alpha=alpha_wait[2], linewidth=2.0)
axs2[idx].plot([wait2 + 1, wait2 + 1], [-10, 10],
linestyle=linestyles_wait[2], color=colors_wait[2], linewidth=2.0, alpha=alpha_wait[2])
axs2[idx].add_patch(Rectangle((wait2, -10), 1, 20,
label="wait 2: skipped" if idx == 0 else "",
facecolor=colors_wait[2], alpha=alpha_wait_rect[2], hatch='\\\\'))
else:
axs2[idx].plot([wait2, wait2], [-10, 10],
linestyle=linestyles_wait[2], color=colors_wait[2],
label="wait 2: " + str(waittime2) + " clk" if idx == 0 else "",
linewidth=2.0, alpha=alpha_wait[2])
# Wait 3
if waittime3 is not None:
if waittime3 == 0:
axs2[idx].plot([wait3, wait3], [-10, 10],
linestyle=linestyles_wait[3], color=colors_wait[3], alpha=alpha_wait[3], linewidth=2.0)
axs2[idx].plot([wait3 + 1, wait3 + 1], [-10, 10],
linestyle=linestyles_wait[3], color=colors_wait[3], linewidth=2.0, alpha=alpha_wait[3])
axs2[idx].add_patch(Rectangle((wait3, -10), 1, 20,
label="wait 3: skipped" if idx == 0 else "",
facecolor=colors_wait[3], alpha=alpha_wait_rect[3], hatch='\\\\'))
else:
axs2[idx].plot([wait3, wait3], [-10, 10],
linestyle=linestyles_wait[3], color=colors_wait[3],
label="wait 3: " + str(waittime3) + " clk" if idx == 0 else "",
linewidth=2.0, alpha=alpha_wait[3])
# Wait 4
if waittime4 is not None:
if waittime4 == 0:
axs2[idx].plot([wait4, wait4], [-10, 10],
linestyle=linestyles_wait[4], color=colors_wait[4], alpha=alpha_wait[4], linewidth=2.0)
axs2[idx].plot([wait4 + 1, wait4 + 1], [-10, 10],
linestyle=linestyles_wait[4], color=colors_wait[4], linewidth=2.0, alpha=alpha_wait[4])
axs2[idx].add_patch(Rectangle((wait4, -10), 1, 20,
label="wait 4: skipped" if idx == 0 else "",
facecolor=colors_wait[4], alpha=alpha_wait_rect[4], hatch='\\\\'))
else:
axs2[idx].plot([wait4, wait4], [-10, 10],
linestyle=linestyles_wait[4], color=colors_wait[4],
label="wait 4: " + str(waittime4) + " clk" if idx == 0 else "",
linewidth=2.0, alpha=alpha_wait[4])
# Wait 5
if waittime5 is not None:
if waittime5 == 0:
axs2[idx].plot([wait5, wait5], [-10, 10],
linestyle=linestyles_wait[5], color=colors_wait[5], alpha=alpha_wait[5], linewidth=2.0)
axs2[idx].plot([wait5 + 1, wait5 + 1], [-10, 10],
linestyle=linestyles_wait[5], color=colors_wait[5], linewidth=2.0, alpha=alpha_wait[5])
axs2[idx].add_patch(Rectangle((wait5, -10), 1, 20,
label="wait 5: skipped" if idx == 0 else "",
facecolor=colors_wait[5], alpha=alpha_wait_rect[5], hatch='\\\\'))
else:
axs2[idx].plot([wait5, wait5], [-10, 10],
linestyle=linestyles_wait[5], color=colors_wait[5],
label="wait 5: " + str(waittime5) + " clk" if idx == 0 else "",
linewidth=2.0, alpha=alpha_wait[5])
# =====================================================================================================
# Plot the loop lines
# Loop 0
if nloop0 is not None:
if nloop0 == 0:
axs2[idx].plot([loop0_start, loop0_start], [-10, 10],
linestyle=linestyles_loop[0], color=colors_loop[0],
alpha=alpha_loop[0], linewidth=2.0)
axs2[idx].plot([loop0_end + 1, loop0_end + 1], [-10, 10],
linestyle=linestyles_loop[0], color=colors_loop[0], alpha=alpha_loop[0], linewidth=2.0)
axs2[idx].add_patch(Rectangle((loop0_start, -10), loop0_end + 1 - loop0_start, 20,
label="loop 0: skipped" if idx == 0 else "",
facecolor=colors_loop[0], alpha=alpha_loop_rect[0], hatch='//'))
else:
axs2[idx].plot([loop0_start, loop0_start], [-10, 10],
linestyle=linestyles_loop[0], color=colors_loop[0], alpha=alpha_loop[0],
label="loop 0: " + str(nloop0) + " times" if idx == 0 else "", linewidth=2.0)
axs2[idx].plot([loop0_end, loop0_end], [-10, 10],
linestyle=linestyles_loop[0], color=colors_loop[0], alpha=alpha_loop[0], linewidth=2.0)
# Loop 1
if nloop1 is not None:
if nloop1 == 0:
axs2[idx].plot([loop1_start, loop1_start], [-10, 10],
linestyle=linestyles_loop[1], color=colors_loop[1],
alpha=alpha_loop[1], linewidth=2.0)
axs2[idx].plot([loop1_end + 1, loop1_end + 1], [-10, 10],
linestyle=linestyles_loop[1], color=colors_loop[1], alpha=alpha_loop[1], linewidth=2.0)
axs2[idx].add_patch(Rectangle((loop1_start, -10), loop1_end + 1 - loop1_start, 20,
label="loop 1: skipped" if idx == 0 else "",
facecolor=colors_loop[1], alpha=alpha_loop_rect[1], hatch='//'))
else:
axs2[idx].plot([loop1_start, loop1_start], [-10, 10],
linestyle=linestyles_loop[1], color=colors_loop[1], alpha=alpha_loop[1],
label="loop 1: " + str(nloop1) + " times" if idx == 0 else "", linewidth=2.0)
axs2[idx].plot([loop1_end, loop1_end], [-10, 10],
linestyle=linestyles_loop[1], color=colors_loop[1], alpha=alpha_loop[1], linewidth=2.0)
# Loop 2
if nloop2 is not None:
if nloop2 == 0:
axs2[idx].plot([loop2_start, loop2_start], [-10, 10],
linestyle=linestyles_loop[2], color=colors_loop[2],
alpha=alpha_loop[2], linewidth=2.0)
axs2[idx].plot([loop2_end + 1, loop2_end + 1], [-10, 10],
linestyle=linestyles_loop[2], color=colors_loop[2], alpha=alpha_loop[2], linewidth=2.0)
axs2[idx].add_patch(Rectangle((loop2_start, -10), loop2_end + 1 - loop2_start, 20,
label="loop 2: skipped" if idx == 0 else "",
facecolor=colors_loop[2], alpha=alpha_loop_rect[2], hatch='//'))
else:
axs2[idx].plot([loop2_start, loop2_start], [-10, 10],
linestyle=linestyles_loop[2], color=colors_loop[2], alpha=alpha_loop[2],
label="loop 2: " + str(nloop2) + " times" if idx == 0 else "", linewidth=2.0)
axs2[idx].plot([loop2_end, loop2_end], [-10, 10],
linestyle=linestyles_loop[2], color=colors_loop[2], alpha=alpha_loop[2], linewidth=2.0)
# Loop 3
if nloop3 is not None:
if nloop3 == 0:
axs2[idx].plot([loop3_start, loop3_start], [-10, 10],
linestyle=linestyles_loop[3], color=colors_loop[3],
alpha=alpha_loop[3], linewidth=2.0)
axs2[idx].plot([loop3_end + 1, loop3_end + 1], [-10, 10],
linestyle=linestyles_loop[3], color=colors_loop[3], alpha=alpha_loop[3], linewidth=2.0)
axs2[idx].add_patch(Rectangle((loop3_start, -10), loop3_end + 1 - loop3_start, 20,
label="loop 3: skipped" if idx == 0 else "",
facecolor=colors_loop[3], alpha=alpha_loop_rect[3], hatch='//'))
else:
axs2[idx].plot([loop3_start, loop3_start], [-10, 10],
linestyle=linestyles_loop[3], color=colors_loop[3], alpha=alpha_loop[3],
label="loop 3: " + str(nloop3) + " times" if idx == 0 else "", linewidth=2.0)
axs2[idx].plot([loop3_end, loop3_end], [-10, 10],
linestyle=linestyles_loop[3], color=colors_loop[3], alpha=alpha_loop[3], linewidth=2.0)
# Loop 4
if nloop4 is not None:
if nloop4 == 0:
axs2[idx].plot([loop4_start, loop4_start], [-10, 10],
linestyle=linestyles_loop[4], color=colors_loop[4],
alpha=alpha_loop[4], linewidth=2.0)
axs2[idx].plot([loop4_end + 1, loop4_end + 1], [-10, 10],
linestyle=linestyles_loop[4], color=colors_loop[4], alpha=alpha_loop[4], linewidth=2.0)
axs2[idx].add_patch(Rectangle((loop4_start, -10), loop4_end + 1 - loop4_start, 20,
label="loop 4: skipped" if idx == 0 else "",
facecolor=colors_loop[4], alpha=alpha_loop_rect[4], hatch='//'))
else:
axs2[idx].plot([loop4_start, loop4_start], [-10, 10],
linestyle=linestyles_loop[4], color=colors_loop[4], alpha=alpha_loop[4],
label="loop 4: " + str(nloop4) + " times" if idx == 0 else "", linewidth=2.0)
axs2[idx].plot([loop4_end, loop4_end], [-10, 10],
linestyle=linestyles_loop[4], color=colors_loop[4], alpha=alpha_loop[4], linewidth=2.0)
# Loop 5
if nloop5 is not None:
if nloop5 == 0:
axs2[idx].plot([loop5_start, loop5_start], [-10, 10],
linestyle=linestyles_loop[5], color=colors_loop[5],
alpha=alpha_loop[5], linewidth=2.0)
axs2[idx].plot([loop5_end + 1, loop5_end + 1], [-10, 10],
linestyle=linestyles_loop[5], color=colors_loop[5], alpha=alpha_loop[5], linewidth=2.0)
axs2[idx].add_patch(Rectangle((loop5_start, -10), loop5_end + 1 - loop5_start, 20,
label="loop 5: skipped" if idx == 0 else "",
facecolor=colors_loop[5], alpha=alpha_loop_rect[5], hatch='//'))
else:
axs2[idx].plot([loop5_start, loop5_start], [-10, 10],
linestyle=linestyles_loop[5], color=colors_loop[5], alpha=alpha_loop[5],
label="loop 5: " + str(nloop5) + " times" if idx == 0 else "", linewidth=2.0)
axs2[idx].plot([loop5_end, loop5_end], [-10, 10],
linestyle=linestyles_loop[5], color=colors_loop[5], alpha=alpha_loop[5], linewidth=2.0)
n_cols = count_nonzero([waittime0 != 0, waittime1 != 0, waittime2 != 0, waittime3 != 0, waittime4 != 0, waittime5 != 0,
nloop0 != 0, nloop1 != 0, nloop2 != 0, nloop3 != 0, nloop4 != 0, nloop5 != 0])
if n_cols > 0:
fig2.legend(loc="upper center", ncol=n_cols)
# manager = get_current_fig_manager()
# manager.window.showMaximized()
figure = plt.gcf() # get current figure
figure.set_size_inches(20, 10)
# when saving, specify the DPI
# tight_layout()
plt.savefig(Folder+"/"+File_pat+".png", dpi=300)
# Remove the white space around the plot -- only works on Unix (ImageMagick command)
os.system(f'mogrify -trim {Folder}/{File_pat}.png')
show()
+37 -78
View File
@@ -58,39 +58,56 @@ DEFAULT_SIMULATOR_CONFIGS = [
SIMULATOR_IDS = [f"{det_type}_{num_interface}if_{num_mod}mod" for det_type, num_interface, num_mod in DEFAULT_SIMULATOR_CONFIGS]
'''
for more specific parameters
@pytest.mark.detectorintegration
@pytest.mark.parametrize(
"session_simulator",
[
("ctb", 1, 1),
("xilinx_ctb", 1, 1),
],
indirect=True,
)
def test_define_reg(session_simulator):
det_type, num_interfaces, num_mods, d = session_simulator
'''
@pytest.fixture(scope="session")
def session_simulator(request):
"""
Fixture to start the detector server once and clean up at the end.
Expects request.param = (det_type, num_interfaces, num_mods)
"""
det_type, num_interfaces, num_mods = request.param
fp = sys.stdout
try:
det_type, num_interfaces, num_mods = request.param
fp = sys.stdout
# set up: once per server
Log(LogLevel.INFOBLUE,
f'---- {det_type} | interfaces={num_interfaces} | modules={num_mods} ----', fp)
# set up: once per server
Log(LogLevel.INFOBLUE,
f'---- {det_type} | interfaces={num_interfaces} | modules={num_mods} ----', fp)
cleanup(fp)
startDetectorVirtualServer(det_type, num_mods, fp, True)
startReceiver(num_mods, fp, True)
cleanup(fp)
startDetectorVirtualServer(det_type, num_mods, fp, True, True)
startReceiver(num_mods, fp, True)
Log(LogLevel.INFOBLUE, f'Waiting for server to start up and connect', fp)
d = loadConfig(
name=det_type,
log_file_fp=fp,
num_mods=num_mods,
num_frames=1,
num_interfaces=num_interfaces,
)
Log(LogLevel.INFOBLUE, f'Waiting for server to start up and connect', fp)
d = loadConfig(
name=det_type,
log_file_fp=fp,
num_mods=num_mods,
num_frames=1,
num_interfaces=num_interfaces,
)
loadBasicSettings(name=det_type, d=d, fp=fp)
loadBasicSettings(name=det_type, d=d, fp=fp)
yield det_type, num_interfaces, num_mods, d
cleanup(fp)
yield det_type, num_interfaces, num_mods, d
cleanup(fp)
except Exception as e:
Log(LogLevel.ERROR, f'Tests Failed.', fp)
cleanup(fp)
def pytest_generate_tests(metafunc):
if "session_simulator" not in metafunc.fixturenames:
@@ -110,62 +127,4 @@ def pytest_generate_tests(metafunc):
indirect=True
)
'''
for more specific parameters
@pytest.mark.detectorintegration
@pytest.mark.parametrize(
"session_simulator",
[
("ctb", 1, 1),
("xilinx_ctb", 1, 1),
],
indirect=True,
)
def test_define_reg(session_simulator):
det_type, num_interfaces, num_mods, d = session_simulator
'''
#helper fixture for servers
@pytest.fixture(scope='module')
def setup_parameters(request): # only setup once per module if same parameters used for the scopes
try:
servers, nmods = request.param # comes from @pytest.mark.parametrize(..., indirect=True)
return servers, nmods
except AttributeError:
# fallback default if the test did not parametrize
return (['eiger', 'jungfrau', 'mythen3', 'gotthard2', 'ctb', 'moench', 'xilinx_ctb'], 2)
@pytest.fixture(scope='module')
def test_with_simulators(setup_parameters):
""" Fixture to automatically setup virtual detector servers for testing. """
fp = sys.stdout
servers, nmods = setup_parameters
print("servers:", servers)
print("nmods:", nmods)
try:
for server in servers:
for ninterfaces in range(1,2):
if ninterfaces == 2 and server != 'jungfrau' and server != 'moench':
continue
msg = f'Starting Python API Tests for {server}'
if server == 'jungfrau' or server == 'moench':
msg += f' with {ninterfaces} interfaces'
Log(LogLevel.INFOBLUE, msg, fp)
cleanup(fp)
startDetectorVirtualServer(server, nmods, fp)
startReceiver(nmods, fp)
d = loadConfig(name=server, log_file_fp=fp, num_mods=nmods, num_frames=1, num_interfaces=ninterfaces)
#loadBasicSettings(name=server, d=d, fp=fp)
yield # run test
cleanup(fp) # teardown
except Exception as e:
traceback.print_exc(file=fp)
Log(LogLevel.ERROR, f'Tests Failed.', fp)
cleanup(fp)
+59
View File
@@ -0,0 +1,59 @@
import pytest
import sys
from conftest import session_simulator
from slsdet import Detector
from slsdet._slsdet import slsDetectorDefs
detectorType = slsDetectorDefs.detectorType
@pytest.mark.detectorintegration
def test_rx_ROI(session_simulator):
""" Test rx_ROI property of Detector class. """
det_type, num_interfaces, num_mods, d = session_simulator
assert d is not None
if d.type == detectorType.CHIPTESTBOARD or d.type == detectorType.XILINX_CHIPTESTBOARD:
pytest.skip("Skipping ROI test for ctb/xilinx_ctb detector types.")
if(d.type == detectorType.MYTHEN3 or d.type == detectorType.GOTTHARD2):
d.rx_roi = (0, 10)
roi = d.rx_roi
assert roi == [(0, 10, -1, -1)]
#d.rx_roi = [[5,15, 0, 1]] # not allowed for mythen3
d.rx_roi = [0,10, -1, -1]
assert d.rx_roi == [(0,10,-1,-1)]
d.rx_clearroi()
else:
d.rx_roi = (0, 10, 10, 20)
roi = d.rx_roi
assert roi == [(0, 10, 10, 20)]
d.rx_roi = [5,15,15,25]
assert d.rx_roi == [(5,15,15,25)]
if d.nmod > 1 and (d.type != detectorType.JUNGFRAU) or (d.numinterfaces == 2 and d.type != detectorType.EIGER):
d.rx_roi = [[0,10,0,20], [5,20,410,420]]
roi = d.rx_roi
assert roi == [(0,10,0,20), (5,20,410,420)] #in same file for jungfrau
d.rx_clearroi()
roi = d.rx_roi
assert roi == [(-1,-1,-1,-1)]
+17
View File
@@ -12,6 +12,9 @@ from utils_for_test import (
)
from slsdet import Detector
from slsdet._slsdet import slsDetectorDefs
detectorType = slsDetectorDefs.detectorType
@pytest.mark.detectorintegration
@@ -903,3 +906,17 @@ def test_dac(session_simulator, request):
Log(LogLevel.INFOGREEN, f"{request.node.name} passed")
@pytest.mark.detectorintegration
@pytest.mark.parametrize("session_simulator",[("moench", 1, 2)],indirect=True)
def test_type(session_simulator):
d = Detector()
assert d.type == detectorType.MOENCH
@pytest.mark.detectorintegration
@pytest.mark.parametrize("session_simulator",[("moench", 1, 2), ("jungfrau", 1, 2)],indirect=True)
def test_numinterfaces(session_simulator):
d = Detector()
assert d.numinterfaces == 1
+12 -36
View File
@@ -17,46 +17,18 @@ sys.path.append(str(scripts_dir))
from slsdet import Detector, Ctb, freeSharedMemory
from utils_for_test import (
Log,
LogLevel,
cleanup,
startDetectorVirtualServer,
connectToVirtualServers,
SERVER_START_PORTNO
)
'''
scope = module =>Once per test file/module
to share expensive setup like startDetectorVirtualServer
'''
@pytest.fixture(scope="module")
def det_config():
return {
"name": "ctb",
"num_mods": 1
}
@pytest.fixture(scope="module", autouse=True)
def setup_simulator(det_config):
"""Fixture to start the detector server once and clean up at the end."""
fp = sys.stdout
cleanup(fp)
startDetectorVirtualServer(det_config["name"], det_config["num_mods"], fp)
Log(LogLevel.INFOBLUE, f'Waiting for server to start up and connect')
connectToVirtualServers(det_config["name"], det_config["num_mods"])
Log(LogLevel.INFOBLUE, f'Freeing shm before tests')
freeSharedMemory()
yield # tests run here
cleanup(fp)
from conftest import session_simulator
@pytest.mark.detectorintegration
def test_exptime_after_free_should_raise(setup_simulator):
@pytest.mark.parametrize("session_simulator",[("ctb", 1, 1)],indirect=True)
def test_exptime_after_free_should_raise(session_simulator):
Log(LogLevel.INFOBLUE, f'\nRunning test_exptime_after_free_should_raise')
@@ -78,7 +50,8 @@ def free_and_create_shm():
k.hostname = f"localhost:{SERVER_START_PORTNO}" # free and recreate shm, maps to local shm struct
@pytest.mark.detectorintegration
def test_exptime_after_not_passing_var_should_raise(setup_simulator):
@pytest.mark.parametrize("session_simulator",[("ctb", 1, 1)],indirect=True)
def test_exptime_after_not_passing_var_should_raise(session_simulator):
Log(LogLevel.INFOBLUE, f'\nRunning test_exptime_after_not_passing_var_should_raise')
@@ -102,7 +75,8 @@ def free_and_create_shm_passing_ctb_var(k):
k.hostname = f"localhost:{SERVER_START_PORTNO}" # free and recreate shm, maps to local shm struct
@pytest.mark.detectorintegration
def test_exptime_after_passing_ctb_var_should_raise(setup_simulator):
@pytest.mark.parametrize("session_simulator",[("ctb", 1, 1)],indirect=True)
def test_exptime_after_passing_ctb_var_should_raise(session_simulator):
Log(LogLevel.INFOBLUE, f'\nRunning test_exptime_after_passing_ctb_var_should_raise')
d = Ctb() # creates multi shm (assuming no shm exists)
@@ -125,7 +99,8 @@ def free_and_create_shm_returning_ctb():
return k
@pytest.mark.detectorintegration
def test_exptime_after_returning_ctb_should_raise(setup_simulator):
@pytest.mark.parametrize("session_simulator",[("ctb", 1, 1)],indirect=True)
def test_exptime_after_returning_ctb_should_raise(session_simulator):
Log(LogLevel.INFOBLUE, f'\nRunning test_exptime_after_returning_ctb_should_raise')
d = Ctb() # creates multi shm (assuming no shm exists)
@@ -148,7 +123,8 @@ def test_exptime_after_returning_ctb_should_raise(setup_simulator):
assert str(exc_info.value) == "Shared memory is invalid or freed. Close resources before access."
@pytest.mark.detectorintegration
def test_hostname_twice_acess_old_should_raise(setup_simulator):
@pytest.mark.parametrize("session_simulator",[("ctb", 1, 1)],indirect=True)
def test_hostname_twice_acess_old_should_raise(session_simulator):
Log(LogLevel.INFOBLUE, f'\nRunning test_hostname_twice_acess_old_should_raise')
d = Ctb() # creates multi shm (assuming no shm exists)
-51
View File
@@ -1,51 +0,0 @@
import pytest
import sys
from conftest import test_with_simulators
from slsdet import Detector
from utils_for_test import (
Log,
LogLevel,
)
@pytest.mark.detectorintegration
@pytest.mark.parametrize("setup_parameters", [(["moench"], 2)], indirect=True)
def test_rx_ROI_moench(test_with_simulators, setup_parameters):
""" Test setting and getting rx_ROI property of Detector class for moench. """
d = Detector()
d.rx_roi = (0, 10, 10, 20)
roi = d.rx_roi
assert roi == [(0, 10, 10, 20)]
d.rx_roi = [5,15,15,25]
assert d.rx_roi == [(5,15,15,25)]
d.rx_roi = [[0,10,0,20], [5,20,410,420]]
roi = d.rx_roi
assert roi == [(0,10,0,20), (5,20,410,420)]
d.rx_clearroi()
roi = d.rx_roi
assert roi == [(-1,-1,-1,-1)]
@pytest.mark.detectorintegration
@pytest.mark.parametrize("setup_parameters", [(["mythen3"], 1)], indirect=True)
def test_rx_ROI_mythen(test_with_simulators, setup_parameters):
""" Test setting and getting rx_ROI property of Detector class for mythen. """
d = Detector()
d.rx_roi = (0, 10)
roi = d.rx_roi
assert roi == [(0, 10, -1, -1)]
#d.rx_roi = [[5,15, 0, 1]] # not allowed for mythen3
d.rx_roi = [0,10, -1, -1]
assert d.rx_roi == [(0,10,-1,-1)]
+6 -3
View File
@@ -257,7 +257,7 @@ def connectToVirtualServers(name, num_mods, ctb_object=False):
counts_sec = 5
while (counts_sec != 0):
try:
d.virtual = [num_mods, SERVER_START_PORTNO]
d.virtual = [num_mods, SERVER_START_PORTNO] # sets the hostnames
break
except Exception as e:
# stop server still not up, wait a bit longer
@@ -288,13 +288,16 @@ def startReceiver(num_mods, fp, no_log_file = False, quiet_mode=False):
def loadConfig(name, rx_hostname = 'localhost', settingsdir = None, log_file_fp = None, num_mods = 1, num_frames = 1, num_interfaces = 1):
Log(LogLevel.INFO, 'Loading config', log_file_fp, True)
try:
d = connectToVirtualServers(name, num_mods)
if name == 'ctb' or name == 'xilinx_ctb':
d = connectToVirtualServers(name, num_mods, ctb_object=True)
else:
d = connectToVirtualServers(name, num_mods)
if name == 'jungfrau' or name == 'moench':
d.numinterfaces = num_interfaces
d.udp_dstport = DEFAULT_UDP_DST_PORTNO
if name == 'eiger' or num_interfaces == 2:
if d.numinterfaces == 2:
d.udp_dstport2 = DEFAULT_UDP_DST_PORTNO + 1
d.rx_hostname = rx_hostname