Add Valdrada, the trigger simulation city

invisible_cities/cities/valdrada_test.py

@@ -0,0 +1,53 @@
+import os
+import tables as tb
+
+from .  valdrada           import valdrada
+from .. core.testing_utils import assert_tables_equality
+
+
+def test_valdrada_contains_all_tables(trigger_config):
+    conf, PATH_OUT = trigger_config
+    valdrada(**conf)
+    with tb.open_file(PATH_OUT) as h5out:
+        assert "Run"             in h5out.root
+        assert "Run/events"      in h5out.root
+        assert "Run/runInfo"     in h5out.root
+        assert "Trigger"         in h5out.root
+        assert "Trigger/events"  in h5out.root
+        assert "Trigger/trigger" in h5out.root
+        assert "Trigger/DST"     in h5out.root
+
+
+def test_valdrada_exact_result_multipeak(ICDATADIR, trigger_config):
+    true_out         = os.path.join(ICDATADIR, "test_trigger_fpga_multipeak.h5")
+    conf, PATH_OUT   = trigger_config
+    valdrada(**conf)
+
+    tables = ("Trigger/events", "Trigger/trigger", "Trigger/DST",
+              "Run/events"      , "Run/runInfo")
+
+    with tb.open_file(true_out)  as true_output_file:
+        with tb.open_file(PATH_OUT) as      output_file:
+            for table in tables:
+                assert hasattr(output_file.root, table)
+                got      = getattr(     output_file.root, table)
+                expected = getattr(true_output_file.root, table)
+                assert_tables_equality(got, expected)
+
+
+def test_valdrada_exact_result(ICDATADIR, trigger_config):
+    true_out          = os.path.join(ICDATADIR, "test_trigger_fpga.h5")
+    conf, PATH_OUT    = trigger_config
+    conf['trigger_config']['multipeak'] = None
+    valdrada(**conf)
+
+    tables = ("Trigger/events", "Trigger/trigger", "Trigger/DST",
+              "Run/events"      , "Run/runInfo")
+
+    with tb.open_file(true_out)  as true_output_file:
+        with tb.open_file(PATH_OUT) as      output_file:
+            for table in tables:
+                assert hasattr(output_file.root, table)
+                got      = getattr(     output_file.root, table)
+                expected = getattr(true_output_file.root, table)
+                assert_tables_equality(got, expected)

invisible_cities/conftest.py

@@ -50,6 +50,11 @@ def example_blr_wfs_filename(ICDATADIR):
     return os.path.join(ICDATADIR, "blr_examples.h5")
 
 
+@pytest.fixture(scope='session')
+def example_blr_fpga_wfs_filename(ICDATADIR):
+    return os.path.join(ICDATADIR, "blr_fpga_examples.h5")
+
+
 @pytest.fixture(scope  = 'session',
                 params = ['electrons_40keV_z250_MCRD.h5'])
 def electron_MCRD_file(request, ICDATADIR):
@@ -699,6 +704,44 @@ def deconvolution_config(ICDIR, ICDATADIR, PSFDIR, config_tmpdir):
 
     return conf, PATH_OUT
 
+<<<<<<< HEAD
+=======
+
+@pytest.fixture(scope='function') # Needs to be function as the config dict is modified when running
+def trigger_config(ICDIR, ICDATADIR, config_tmpdir):
+    PATH_IN     = os.path.join(ICDATADIR    ,    "blr_fpga_examples.h5")
+    PATH_OUT    = os.path.join(config_tmpdir,     "trigger_city_out.h5")
+    nevt_req    = 10
+    conf        = dict(files_in       = PATH_IN ,
+                       file_out       = PATH_OUT,
+                       event_range    = nevt_req,
+                       compression    = 'ZLIB4',
+                       print_mod      = 1000,
+                       run_number     = 8093,
+                       trigger_config = dict(coincidence_window = 64
+                                            ,discard_width      = 40
+                                            ,multipeak          = dict(q_min      = 100000
+                                                                      ,time_min   = 2  *units.mus
+                                                                      ,time_after = 800*units.mus)),
+                       channel_config = {0 : dict(q_min           = 5000
+                                                 ,q_max           = 50000
+                                                 ,time_min        = 2 *units.mus
+                                                 ,time_max        = 40*units.mus
+                                                 ,baseline_dev    = 10
+                                                 ,amp_max         = 1000
+                                                 ,pulse_valid_ext = 50*units.ns)
+                                        ,2 : dict(q_min           = 5000
+                                                 ,q_max           = 50000
+                                                 ,time_min        = 2 *units.mus
+                                                 ,time_max        = 40*units.mus
+                                                 ,baseline_dev    = 10
+                                                 ,amp_max         = 1000
+                                                 ,pulse_valid_ext = 50*units.ns)})
+
+    return conf, PATH_OUT
+
+
+>>>>>>> f8fa2343... fixup! Add related tests
 ## To make very slow tests only run with specific option
 def pytest_addoption(parser):
     parser.addoption(

invisible_cities/io/trigger_io_test.py

@@ -0,0 +1,56 @@
+import os
+
+import numpy  as np
+import tables as tb
+
+from numpy.testing import assert_allclose
+from hypothesis                import given
+from hypothesis.strategies     import integers
+from hypothesis.strategies     import lists
+from hypothesis.strategies     import booleans
+from hypothesis.strategies     import composite
+
+from ..core.testing_utils import assert_dataframes_equal
+from ..io.dst_io          import load_dst
+from . trigger_io         import trigger_dst_writer
+
+
+@composite
+def trigger_values(draw):
+    size  = draw(integers(min_value=1, max_value=10))
+    elements = []
+    for _ in range(6):
+        elements.append(draw(lists(integers(min_value=0, max_value=np.iinfo(np.uint32).max),
+                                   min_size=size, max_size=size)))
+    for _ in range(4):
+        elements.append(draw(lists(booleans(), min_size=size, max_size=size)))
+    for _ in range(5):
+        elements.append(draw(lists(integers(min_value=np.iinfo(np.int32).min, max_value=np.iinfo(np.int32).max),
+                                   min_size=size, max_size=size)))
+    return elements
+
+
+@given(triggers=trigger_values())
+def test_trigger_dst_writer(config_tmpdir, triggers):
+    output_file = os.path.join(config_tmpdir, "test_trigger_dst.h5")
+
+    trigger_save = [list(a) for a in zip(*triggers)]
+    with tb.open_file(output_file, 'w') as h5out:
+        write = trigger_dst_writer(h5out)
+        write(trigger_save)
+
+    trigger_dst = load_dst(output_file, group='Trigger', node='DST')
+
+    col_names = ["event"     , "pmt"       , "trigger_time" , "q"
+                ,"width"     , "height"    ,  "valid_q"     , "valid_w"
+                ,"valid_h"   , "valid_peak", "valid_all"    , "baseline"
+                ,"max_height", "n_coinc"   , "closest_ttime", "closest_pmt"]
+
+    assert np.all([tname == cname for tname, cname in zip(trigger_dst.columns.values, col_names)])
+
+    # Check values stored, the if is needed because the valid_all condition is done in the writer, not on the input trigger
+    for i, colname in enumerate(col_names):
+        if colname == 'valid_all':
+            assert_allclose(np.all(triggers[6:10], axis=0), trigger_dst.loc[:, colname])
+        elif i>10: assert_allclose(triggers[i-1]          , trigger_dst.loc[:, colname])
+        else     : assert_allclose(triggers[i  ]          , trigger_dst.loc[:, colname])