From 4962d288a1d6e6b6054fc5440c39ac77e5df5a14 Mon Sep 17 00:00:00 2001
From: Steffen Hauf <haufs@max-exfl030.desy.de>
Date: Thu, 15 Nov 2018 16:43:55 +0100
Subject: [PATCH] PEP8 formatting
---
tests/correction_base.py | 354 ++++++++++++++++++++++++++++++---------
tests/test_agipd.py | 60 +++----
tests/test_lpd.py | 60 +++----
3 files changed, 337 insertions(+), 137 deletions(-)
diff --git a/tests/correction_base.py b/tests/correction_base.py
index 99e5d6cd4..569902363 100644
--- a/tests/correction_base.py
+++ b/tests/correction_base.py
@@ -1,35 +1,50 @@
-from abc import abstractmethod
import argparse
-from enum import Enum
-from functools import partial
import glob
import hashlib
-from multiprocessing import Pool
import os
import pickle
import re
import subprocess as sp
-import sys
-from time import sleep
import unittest
+from abc import abstractmethod
+from enum import Enum
+from functools import partial
+from multiprocessing import Pool
+from time import sleep
-from git import Repo
import h5py
import numpy as np
+from git import Repo
from scipy import stats
-
np.warnings.filterwarnings('ignore')
-
parser = argparse.ArgumentParser()
-parser.add_argument('--generate', action="store_true", default=False)
-parser.add_argument('--generate-wo-execution', action="store_true", default=False)
-parser.add_argument('--test-wo-execution', action="store_true", default=False)
-parser.add_argument('--skip-checksum', action="store_true", default=False)
-parser.add_argument('--skip-histogram', action="store_true", default=False)
-parser.add_argument('--skip-karabo-data', action="store_true", default=False)
-parser.add_argument('--artefact-dir', type=str, default="./artifacts/")
+parser.add_argument('--generate', action="store_true", default=False,
+ help="Set this flag to generte new artifacts from " +
+ "the test. These will be placed in the artifact " +
+ "directory, under the latest commit your git " +
+ "repository is on. This will launch the " +
+ "notebook under test first.")
+parser.add_argument('--generate-wo-execution', action="store_true",
+ default=False, help="Set this flag to generte new " +
+ "artifacts from the test. These will be placed in " +
+ "the artifact directory, under the latest commit " +
+ "your git repository is on. This will not launch " +
+ "the notebook being tested, but assumes it's " +
+ "output is already present. Use e.g. to debug tests.")
+parser.add_argument('--test-wo-execution', action="store_true", default=False,
+ help="Run tests, but do not execute the notebook being " +
+ "tested first. This is assumes its output is already " +
+ "present. Use e.g. to debug tests.")
+parser.add_argument('--skip-checksum', action="store_true", default=False,
+ help="Skip checksum tests (and artifact generation)")
+parser.add_argument('--skip-histogram', action="store_true", default=False,
+ help="Skip histogram tests (and artifact generation)")
+parser.add_argument('--skip-karabo-data', action="store_true", default=False,
+ help="Skip karabo_data tests (and artifact generation)")
+parser.add_argument('--artefact-dir', type=str, default="./artifacts/",
+ help="Set directory to place artifacts in.")
parser.add_argument('unittest_args', nargs='*')
args = parser.parse_args()
@@ -38,21 +53,29 @@ class Failures(Enum):
ARTIFACT_MISSING = "No artifact"
EMPTY_ARTIFACT = "Empty artifact"
-
+
def _do_generate():
+ """ Determine if artifacts should be generated
+ """
return args.generate or args.generate_wo_execution
-def _get_last_commit():
- r = Repo(os.path.dirname(os.path.realpath(__file__))+"/..")
+def get_last_commit():
+ """ Return the last commit from the git repo
+ """
+ r = Repo(os.path.dirname(os.path.realpath(__file__)) + "/..")
last_commit = None
for last_commit in r.iter_commits():
pass
return last_commit
-def _get_artifact_dir(cls):
- last_commit = _get_last_commit()
+def get_artifact_dir(cls):
+ """ Get the artifact director for the last commit
+
+ :param cls: Test class
+ """
+ last_commit = get_last_commit()
print("Last git commit is: {}".format(last_commit))
test_name = cls.__name__
art_base = os.path.realpath(args.artefact_dir)
@@ -61,8 +84,12 @@ def _get_artifact_dir(cls):
return path
-def _get_artifact_comp_dir(cls):
- r = Repo(os.path.dirname(os.path.realpath(__file__))+"/..")
+def get_artifact_comp_dir(cls):
+ """Get the artifact director for the last commit that generated any
+
+ :param cls: Test class
+ """
+ r = Repo(os.path.dirname(os.path.realpath(__file__)) + "/..")
all_commits = list(r.iter_commits())
test_name = cls.__name__
art_base = os.path.realpath(args.artefact_dir)
@@ -74,27 +101,70 @@ def _get_artifact_comp_dir(cls):
+ " in directory {}! Check if directory is correct"
+ " or create artifacts using the --generate flag.")
raise FileNotFoundError(msg.format(test_name, art_base))
-
-def _get_matching_h5_paths(f, template):
+
+def get_matching_h5_paths(f, template):
+ """ Return paths in h5 file that match a path template
+
+ :param f: and h5 file object
+ :param template: a path template, may contain regex expressions
+
+ Paths are evaluated on each hierarchy level vs. the template. Levels are
+ seprated by "/", e.g.
+
+ INSTRUMENT/(.+)LPD(.+)/DET/[0-9]+CH0:xtdf/image/mask
+
+ Will match any LPD instance on the second level, and any channels on the
+ fourth level.
+ """
matches = []
+
def check_key(pre, pc):
for k in f[pre].keys():
m = re.match(pc[0], k)
if m:
if len(pc[1:]) > 0:
- check_key(pre+m.group(0)+"/", pc[1:])
+ check_key(pre + m.group(0) + "/", pc[1:])
else:
- matches.append(pre+m.group(0))
+ matches.append(pre + m.group(0))
+
check_key("/", template.split("/"))
return matches
def parallel_hist_gen(hist_paths, artifact_dir, fname):
+ """ Function to generate histogram artifacts in parallel
+
+ :param hist_paths: paths to create histograms for. Should be a dict with
+ the following structure:
+
+ { path_template :
+ { htype_name1 : { "bins": 100, "range": (0, 1000),
+ "scl_fun": np.log2},
+ htype_name2 : { "bins": 100, "range": (0, 100000) },
+ }
+ }
+
+ here `path_template` is a template accepted by `get_matching_h5_paths`,
+ `htype_name1` is a name for the generated histogram, by which it
+ will be identified during testing, `bins` and `range` define the
+ histogram, and optionally, `scl_fun` is a scaling function to be
+ executed on the data before creating the histogram.
+
+ :param artifact_dir: the directory under which to place histograms.
+ For each input file a histogram file (a npz archive) containing
+ the named histograms defined by `hist_paths` is created.
+ :param fname: path to the h5 file containing the paths specified in
+ `hist_paths`.
+
+ For pickling to work in subprocess calls, this needs to be defined
+ on the module level.
+
+ """
all_hists = {}
with h5py.File(fname, "r") as f:
for path, hists in hist_paths.items():
- mpaths = _get_matching_h5_paths(f, path)
+ mpaths = get_matching_h5_paths(f, path)
for mpath in mpaths:
print("Creating histograms for path: {}".format(mpath))
d = f[mpath]
@@ -112,25 +182,50 @@ def parallel_hist_gen(hist_paths, artifact_dir, fname):
hpath = "{}/{}".format(artifact_dir, os.path.basename(hist_fname))
if len(all_hists):
np.savez(hpath, **all_hists)
-
-
+
+
def parallel_hist_eval(hist_paths, cls, fname):
+ """ Function to compare histogram artifacts in parallel
+
+ :param hist_paths: paths to create histograms for. Should be a dict with
+ the following structure:
+
+ { path_template :
+ { htype_name1 : { "bins": 100, "range": (0, 1000),
+ "scl_fun": np.log2},
+ htype_name2 : { "bins": 100, "range": (0, 100000) },
+ }
+ }
+
+ here `path_template` is a template accepted by `get_matching_h5_paths`,
+ `htype_name1` is a name for the generated histogram, by which it
+ will be identified during testing, `bins` and `range` define the
+ histogram, and optionally, `scl_fun` is a scaling function to be
+ executed on the data before creating the histogram.
+
+ :param cls: Test class
+ :param fname: path to the h5 file containing the paths specified in
+ `hist_paths`.
+
+ For pickling to work in subprocess calls, this needs to be defined
+ on the module level.
+ """
ret = []
with h5py.File(fname, "r") as f:
hist_fname = "{}.hist.npz".format(fname)
- test_art_dir = _get_artifact_comp_dir(cls)
+ test_art_dir = get_artifact_comp_dir(cls)
hpath = "{}/{}".format(test_art_dir, os.path.basename(hist_fname))
has_artifact = os.path.exists(hpath)
if not has_artifact:
return Failures.ARTIFACT_MISSING
-
+
try:
test_hists = np.load(hpath)
except OSError:
return Failures.EMPTY_ARTIFACT # likely empty data
-
+
for path, hists in hist_paths.items():
- mpaths = _get_matching_h5_paths(f, path)
+ mpaths = get_matching_h5_paths(f, path)
for mpath in mpaths:
print("Creating histograms for path: {}".format(mpath))
d = f[mpath]
@@ -142,7 +237,8 @@ def parallel_hist_eval(hist_paths, cls, fname):
h, _ = np.histogram(d, bins=bins, range=rnge)
else:
h, _ = np.histogram(fn(d), bins=bins, range=rnge)
- if mpath.startswith("/"): # would have been trucated on saving
+ if mpath.startswith("/"):
+ # would have been trucated on saving
mpath = mpath[1:]
th = test_hists["{}_{}".format(mpath, htype)]
ret.append((mpath, htype, th, h))
@@ -151,6 +247,9 @@ def parallel_hist_eval(hist_paths, cls, fname):
class CorrectionTestBase:
+ """
+ A base class for testing correction type notebooks.
+ """
detector = None
task = None
parms = {}
@@ -159,29 +258,38 @@ class CorrectionTestBase:
artifact_comp_dir = None
hist_paths = []
karabo_data_inspects = []
-
+
@classmethod
def setUpClass(cls):
- assert(cls.detector is not None)
- assert(cls.task is not None)
+ """
+ Sets up the test by executing the notebook under test
+
+ If artifact generation is requested an artifact directory is
+ created.
+
+ Note that this method will block until any slurm jobs scheduled
+ as part of the notebook execution have finished.
+ """
+ assert (cls.detector is not None)
+ assert (cls.task is not None)
cmd = ["xfel-calibrate", cls.detector, cls.task]
for k, v in cls.parms.items():
cmd += ["--{}".format(k), str(v)]
-
+
print("Executing {}".format(" ".join(cmd)))
-
+
if _do_generate():
print("Creating data paths for artifacts")
- cls.artifact_dir = _get_artifact_dir(cls)
+ cls.artifact_dir = get_artifact_dir(cls)
if not os.path.exists(cls.artifact_dir):
os.makedirs(cls.artifact_dir)
-
+
if args.generate_wo_execution or args.test_wo_execution:
return
-
+
out = sp.check_output(cmd)
joblist = None
-
+
for ln in out.decode().split("\n"):
if "Submitted the following SLURM jobs:" in ln:
txt, jobs = ln.split(":")
@@ -191,10 +299,16 @@ class CorrectionTestBase:
@abstractmethod
def _output_to_path(self):
+ """ Return the path a notebook under test places its results in.
+
+ Must be overwritten by concrete test classes.
+ """
pass
-
+
@classmethod
def _wait_on_jobs(cls, joblist):
+ """ Wait on SLURM jobs defined by `joblist` to finish.
+ """
print("Waiting on jobs to finish: {}".format(joblist))
while True:
found_jobs = set()
@@ -206,66 +320,114 @@ class CorrectionTestBase:
if len(found_jobs) == 0:
break
sleep(10)
-
+
def _create_md5(self, fname):
+ """ Create an MD5 checksum for the file at `fname`.
+ """
print("Creating MD5 checksum of: {}".format(fname))
hash_md5 = hashlib.md5()
with open(fname, "rb") as f:
for chunk in iter(lambda: f.read(4096), b""):
hash_md5.update(chunk)
return hash_md5.hexdigest()
-
+
@unittest.skipUnless(_do_generate() and not args.skip_checksum,
"Artifact generation is not requested")
def test_generate_checksums(self):
+ """ Generate MD5 checksums of output files from notebook
+ """
out_folder = self._output_to_path()
- files_to_check = glob.glob("{}/*{}".format(out_folder, self.rel_file_ext))
+ files_to_check = glob.glob(
+ "{}/*{}".format(out_folder, self.rel_file_ext))
for fname in files_to_check:
m5 = self._create_md5(fname)
m5fname = "{}.md5".format(fname)
- m5path = "{}/{}".format(self.artifact_dir, os.path.basename(m5fname))
+ m5path = "{}/{}".format(self.artifact_dir,
+ os.path.basename(m5fname))
with open(m5path, "w") as f:
f.write(m5)
-
+
@unittest.skipIf(args.skip_checksum,
"User requested to skip checksum test")
def test_checksums(self):
+ """ Compare MD5 checksums of notebook's output files with artifacts
+
+ This test will verify if output files are identical. Even for
+ small changes in the correction logic this test is likely to fail.
+ If this is the case, it is recommended to verify correctness using
+ the other tests, which inspect data, and the create new checksums
+ using the --generate option.
+
+ If no previous checksum exists for a given file the test for that
+ file will fail.
+ """
out_folder = self._output_to_path()
- files_to_check = glob.glob("{}/*{}".format(out_folder, self.rel_file_ext))
+ files_to_check = glob.glob(
+ "{}/*{}".format(out_folder, self.rel_file_ext))
for fname in files_to_check:
m5fname = "{}.md5".format(fname)
- test_art_dir = _get_artifact_comp_dir(self.__class__)
+ test_art_dir = get_artifact_comp_dir(self.__class__)
m5path = "{}/{}".format(test_art_dir, os.path.basename(m5fname))
-
+
with self.subTest(msg="Verifying against: {}".format(m5path)):
- self.assertTrue(os.path.exists(m5path), "No comparison MD5 found")
+ self.assertTrue(os.path.exists(m5path),
+ "No comparison MD5 found")
m5 = self._create_md5(fname)
with open(m5path, "r") as f:
m5test = f.read().strip()
self.assertEqual(m5.strip(), m5test)
-
+
@unittest.skipUnless(_do_generate() and not args.skip_histogram,
"Artifact generation is not requested")
def test_generate_histograms(self):
+ """ Generate histogram artifacts for the output of the notebook
+ """
out_folder = self._output_to_path()
- files_to_check = glob.glob("{}/*{}".format(out_folder, self.rel_file_ext))
+ files_to_check = glob.glob(
+ "{}/*{}".format(out_folder, self.rel_file_ext))
with Pool(8) as p:
- p.map(partial(parallel_hist_gen, self.hist_paths, self.artifact_dir),
- files_to_check)
+ pf = partial(parallel_hist_gen, self.hist_paths, self.artifact_dir)
+ p.map(pf, files_to_check)
self.assertTrue(True)
-
+
@unittest.skipIf(args.skip_histogram,
"User requested to skip histogram test")
def test_histograms(self):
+ """ Compare histograms of notebook output with previous artifacts
+
+ Comparison is performed in multiple tests:
+
+ * using np.allclose, which tests that histograms are equal within
+ numerical limits
+
+ Using statistical tests to check for p-value compatibility within
+ confidence levels of 0.9, 0.95 and 0.99
+
+ * via a Kolmogornov-Smirnoff test to verify that distributions
+ are of similar shape
+
+ * via a χ2 test
+
+ * via a Shapiro-Wilks test to verify normal distribution of the
+ the deviation of both histogram (which would be expected, if
+ the deviation is of statistical and not systematic nature).
+
+ If no previous histograms exist for a given file the test for that
+ file will fail.
+
+ Empty files are skipped.
+ """
out_folder = self._output_to_path()
- files_to_check = glob.glob("{}/*{}".format(out_folder, self.rel_file_ext))
+ files_to_check = glob.glob(
+ "{}/*{}".format(out_folder, self.rel_file_ext))
with Pool(8) as p:
- r = p.map(partial(parallel_hist_eval, self.hist_paths, self.__class__),
- files_to_check)
+ pf = partial(parallel_hist_eval, self.hist_paths, self.__class__)
+ r = p.map(pf, files_to_check)
for i, rvals in enumerate(r):
msg = "Verifying: {}".format(files_to_check[i])
with self.subTest(msg=msg):
- self.assertNotEqual(Failures.ARTIFACT_MISSING, "No comparison histograms found")
+ self.assertNotEqual(Failures.ARTIFACT_MISSING,
+ "No comparison histograms found")
if rvals is Failures.ARTIFACT_MISSING:
return
if rvals is Failures.EMPTY_ARTIFACT:
@@ -275,35 +437,46 @@ class CorrectionTestBase:
mpath, htype, th, h = rval
# straight-forward all equal
- msg = "Test all values equal for: {}/{}".format(mpath, htype)
+ msg = "Test all values equal for: {}/{}".format(mpath,
+ htype)
with self.subTest(msg=msg):
self.assertTrue(np.allclose(h, th))
-
+
confidence_levels = [0.9, 0.95, 0.99]
for cl in confidence_levels:
# run additional tests and report on them
- msg = "KS-Test for: {}/{}, confidence-level: {}".format(mpath, htype, cl)
+ msg = "KS-Test for: {}/{}, confidence-level: {}"
+ msg = msg.format(mpath, htype, cl)
with self.subTest(msg=msg):
D, p = stats.ks_2samp(h, th)
self.assertGreaterEqual(p, cl)
idx = (h != 0) & (th != 0)
- msg = "Chi2-Test for: {}/{}, confidence-level: {}".format(mpath, htype, cl)
+ msg = "Chi2-Test for: {}/{}, confidence-level: {}"
+ msg = msg.format(mpath, htype, cl)
with self.subTest(msg=msg):
chisq, p = stats.chisquare(h[idx], th[idx])
self.assertGreaterEqual(p, cl)
- msg = "AD-Test for: {}/{}, confidence-level: {}".format(mpath, htype, cl)
+ msg = ("Shapiro-Wilks-Test for: {}/{}, " +
+ "confidence-level: {}")
+ msg = msg.format(mpath, htype, cl)
with self.subTest(msg=msg):
- t, p = stats.shapiro((h[idx]-th[idx])/np.sqrt(th[idx]))
+ enorm = (h[idx] - th[idx]) / np.sqrt(th[idx])
+ t, p = stats.shapiro(enorm)
self.assertGreaterEqual(p, cl)
@unittest.skipUnless(_do_generate() and not args.skip_karabo_data,
"Artifact generation is not requested")
def test_generate_karabo_data(self):
+ """ Generate artifacts for the Karabo Data test of notebook output
+
+ Note that Karabo Data related imports are inline in this test so
+ that other tests may be executed without Karabo data installed.
+ """
out_folder = self._output_to_path()
kdata = "{}/karabo.data".format(self.artifact_dir)
- # we inlne the inport to be able to skip if not installed
+ # we inline the import to be able to skip if not installed
import karabo_data as kd
rd = kd.RunDirectory(out_folder)
d = {}
@@ -322,36 +495,55 @@ class CorrectionTestBase:
for key in train[source].keys():
if key in self.karabo_data_inspects:
d[identifier][source][key] = train[source][key]
-
+
_, first_train = next(rd.trains()) # also tests iteration
write_train_info(first_train, "first_train")
-
+
_, last_train = rd.train_from_id(rd.train_ids[-1])
write_train_info(last_train, "last_train")
-
+
with open(kdata, 'wb') as f:
- pickle.dump(d, f, pickle.HIGHEST_PROTOCOL)
-
+ pickle.dump(d, f, pickle.HIGHEST_PROTOCOL)
+
@unittest.skipIf(args.skip_karabo_data,
"User requested to skip karabo data test")
def test_karabo_data(self):
+ """ Test Karabo Data compatibility for notebook output
+
+ The following tests are performed:
+
+ * test that output files can be loaded as a `RunDirectory`
+ * verify that detector data sources are the same
+ * verify that detector info for all sources is the same
+ * verify that train ids are unchanged
+ * for the first and last train
+ - verify all sources and all keys are the same
+ - verify all data paths defined in `karabo_data_inspects` is
+ equal. This should be metadata, not data processed as part of
+ the notebook
+ these tests also test train iteration (`next(rd.trains())`) and
+ selection (`rd.train_from_id(rd.train_ids[-1])`) on the output.
+
+ Note that Karabo Data related imports are inline in this test so
+ that other tests may be executed without Karabo data installed.
+ """
out_folder = self._output_to_path()
- kdata = "{}/karabo.data".format(_get_artifact_comp_dir(self.__class__))
- # we inlne the inport to be able to skip if not installed
+ kdata = "{}/karabo.data".format(get_artifact_comp_dir(self.__class__))
+ # we inline the import to be able to skip if not installed
import karabo_data as kd
rd = kd.RunDirectory(out_folder)
-
+
# test against artifacts
with open(kdata, 'rb') as f:
d = pickle.load(f)
-
+
self.assertEqual(d["datasources"], rd.detector_sources)
for source, info in d["detector_infos"].items():
self.assertEqual(info, rd.detector_info(source))
self.assertEqual(d["trainids"], rd.train_ids)
-
+
def test_train_info(train, identifier):
-
+
self.assertEqual(sorted(d[identifier]["sources"]),
sorted(list(train.keys())))
for source in train.keys():
diff --git a/tests/test_agipd.py b/tests/test_agipd.py
index 51c4f7336..9d6219e1c 100644
--- a/tests/test_agipd.py
+++ b/tests/test_agipd.py
@@ -1,58 +1,62 @@
-import numpy as np
import sys
import unittest
-from correction_base import CorrectionTestBase, args
+import numpy as np
+from correction_base import CorrectionTestBase, args
class TestAGIPDCorrection(CorrectionTestBase, unittest.TestCase):
-
detector = "AGIPD"
task = "CORRECT"
parms = {"in-folder": "/gpfs/exfel/exp/SPB/201831/p900039/raw/",
"run": 412,
"out-folder": "/gpfs/exfel/data/scratch/haufs/test/",
- "calfile": "/gpfs/exfel/data/scratch/haufs/agipd_on_demand/agipd_store.h5",
+ "calfile": "/gpfs/exfel/data/scratch/haufs/agipd_on_demand"
+ "/agipd_store.h5",
"sequences": 0}
- hist_paths = {"INSTRUMENT/(.+)AGIPD(.+)/DET/[0-9]+CH0:xtdf/image/data":
- {
- "broad": {"bins": 10000, "range": (-1e4, 1e6)},
- "detail": {"bins": 10000, "range": (-100, 9900)}
- },
- "INSTRUMENT/(.+)AGIPD(.+)/DET/[0-9]+CH0:xtdf/image/mask":
- {
- "detail": {"bins": 32, "range": (0, 32),
- "scl_fun": np.log2}
- },
- "INSTRUMENT/(.+)AGIPD(.+)/DET/[0-9]+CH0:xtdf/image/gain":
- {
- "detail": {"bins": 3, "range": (0, 3)}
- }
- }
-
- karabo_data_inspects = ['trailer.checksum', 'header.trainId', 'header.reserved',
+ hist_paths = {
+ "INSTRUMENT/(.+)AGIPD(.+)/DET/[0-9]+CH0:xtdf/image/data":
+ {
+ "broad": {"bins": 10000, "range": (-1e4, 1e6)},
+ "detail": {"bins": 10000, "range": (-100, 9900)}
+ },
+ "INSTRUMENT/(.+)AGIPD(.+)/DET/[0-9]+CH0:xtdf/image/mask":
+ {
+ "detail": {"bins": 32, "range": (0, 32),
+ "scl_fun": np.log2}
+ },
+ "INSTRUMENT/(.+)AGIPD(.+)/DET/[0-9]+CH0:xtdf/image/gain":
+ {
+ "detail": {"bins": 3, "range": (0, 3)}
+ }
+ }
+
+ karabo_data_inspects = ['trailer.checksum', 'header.trainId',
+ 'header.reserved',
'image.cellId', 'header.dataId', 'header.linkId',
'detector.data', 'metadata', 'image.length',
- 'header.magicNumberBegin', 'header.majorTrainFormatVersion',
- 'trailer.magicNumberEnd', 'image.status', 'image.trainId',
+ 'header.magicNumberBegin',
+ 'header.majorTrainFormatVersion',
+ 'trailer.magicNumberEnd', 'image.status',
+ 'image.trainId',
'image.pulseId', 'header.minorTrainFormatVersion',
- 'header.pulseCount', 'trailer.status', 'detector.trainId',
+ 'header.pulseCount', 'trailer.status',
+ 'detector.trainId',
'trailer.trainId']
-
+
def _output_to_path(self):
opath = self.parms["out-folder"]
run = int(self.parms["run"])
return "{}/r{:04d}".format(opath, run)
-
if __name__ == '__main__':
sys.argv[1:] = args.unittest_args
-
+
loader = unittest.TestLoader()
# make sure generators get run first
ln = lambda f: "generate" not in f
lncmp = lambda a, b: (ln(a) > ln(b)) - (ln(a) < ln(b))
loader.sortTestMethodsUsing = lncmp
- unittest.main(testLoader=loader, verbosity=3)
\ No newline at end of file
+ unittest.main(testLoader=loader, verbosity=3)
diff --git a/tests/test_lpd.py b/tests/test_lpd.py
index 3740602a8..5607cd8cd 100644
--- a/tests/test_lpd.py
+++ b/tests/test_lpd.py
@@ -1,60 +1,64 @@
-import numpy as np
import sys
import unittest
-from correction_base import CorrectionTestBase, args
+import numpy as np
+from correction_base import CorrectionTestBase, args
np.warnings.filterwarnings('ignore')
class TestLPDCorrection(CorrectionTestBase, unittest.TestCase):
-
detector = "LPD"
task = "CORRECT"
parms = {"in-folder": "/gpfs/exfel/exp/FXE/201831/p900038/raw/",
"run": 154,
"out-folder": "/gpfs/exfel/data/scratch/haufs/test/",
- "calfile": "/gpfs/exfel/exp/FXE/201831/p900038/usr/calibration0818/cal_constants2.h5",
+ "calfile": "/gpfs/exfel/exp/FXE/201831/p900038/usr"
+ "/calibration0818/cal_constants2.h5",
"sequences": 0}
- hist_paths = {"INSTRUMENT/(.+)LPD(.+)/DET/[0-9]+CH0:xtdf/image/data":
- {
- "broad": {"bins": 10000, "range": (-1e4, 1e6)},
- "detail": {"bins": 10000, "range": (-100, 9900)}
- },
- "INSTRUMENT/(.+)LPD(.+)/DET/[0-9]+CH0:xtdf/image/mask":
- {
- "detail": {"bins": 32, "range": (0, 32),
- "scl_fun": np.log2}
- },
- "INSTRUMENT/(.+)LPD(.+)/DET/[0-9]+CH0:xtdf/image/gain":
- {
- "detail": {"bins": 3, "range": (0, 3)}
- }
- }
-
- karabo_data_inspects = ['trailer.checksum', 'header.trainId', 'header.reserved',
+ hist_paths = {
+ "INSTRUMENT/(.+)LPD(.+)/DET/[0-9]+CH0:xtdf/image/data":
+ {
+ "broad": {"bins": 10000, "range": (-1e4, 1e6)},
+ "detail": {"bins": 10000, "range": (-100, 9900)}
+ },
+ "INSTRUMENT/(.+)LPD(.+)/DET/[0-9]+CH0:xtdf/image/mask":
+ {
+ "detail": {"bins": 32, "range": (0, 32),
+ "scl_fun": np.log2}
+ },
+ "INSTRUMENT/(.+)LPD(.+)/DET/[0-9]+CH0:xtdf/image/gain":
+ {
+ "detail": {"bins": 3, "range": (0, 3)}
+ }
+ }
+
+ karabo_data_inspects = ['trailer.checksum', 'header.trainId',
+ 'header.reserved',
'image.cellId', 'header.dataId', 'header.linkId',
'detector.data', 'metadata', 'image.length',
- 'header.magicNumberBegin', 'header.majorTrainFormatVersion',
- 'trailer.magicNumberEnd', 'image.status', 'image.trainId',
+ 'header.magicNumberBegin',
+ 'header.majorTrainFormatVersion',
+ 'trailer.magicNumberEnd', 'image.status',
+ 'image.trainId',
'image.pulseId', 'header.minorTrainFormatVersion',
- 'header.pulseCount', 'trailer.status', 'detector.trainId',
+ 'header.pulseCount', 'trailer.status',
+ 'detector.trainId',
'trailer.trainId']
-
+
def _output_to_path(self):
opath = self.parms["out-folder"]
run = int(self.parms["run"])
return "{}/r{:04d}".format(opath, run)
-
if __name__ == '__main__':
sys.argv[1:] = args.unittest_args
-
+
loader = unittest.TestLoader()
# make sure generators get run first
ln = lambda f: "generate" not in f
lncmp = lambda a, b: (ln(a) > ln(b)) - (ln(a) < ln(b))
loader.sortTestMethodsUsing = lncmp
- unittest.main(testLoader=loader, verbosity=3)
\ No newline at end of file
+ unittest.main(testLoader=loader, verbosity=3)
--
GitLab