Merge branch 'master' into feature-add_calibscale

ctapipe/conftest.py

@@ -9,6 +9,7 @@
 from ctapipe.io import SimTelEventSource
 from ctapipe.utils import get_dataset_path
 from ctapipe.instrument import CameraGeometry
+from ctapipe.utils.filelock import FileLock
 
 
 # names of camera geometries available on the data server
@@ -135,3 +136,117 @@ def prod5_proton_simtel_path():
     return get_dataset_path(
         "proton_20deg_0deg_run4___cta-prod5-paranal_desert-2147m-Paranal-dark-100evts.simtel.zst"
     )
+
+
+@pytest.fixture(scope="session")
+def dl1_tmp_path(tmp_path_factory):
+    return tmp_path_factory.mktemp("dl1")
+
+
+@pytest.fixture(scope="session")
+def dl1_file(dl1_tmp_path):
+    """
+    DL1 file containing both images and parameters from a gamma simulation set.
+    """
+    from ctapipe.tools.stage1 import Stage1Tool
+    from ctapipe.core import run_tool
+
+    output = dl1_tmp_path / "images.dl1.h5"
+
+    # prevent running stage1 multiple times in case of parallel tests
+    with FileLock(output.with_suffix(output.suffix + ".lock")):
+        if output.is_file():
+            return output
+
+        infile = get_dataset_path("gamma_test_large.simtel.gz")
+
+        argv = [
+            f"--input={infile}",
+            f"--output={output}",
+            "--write-images",
+            "--max-events=20",
+            "--allowed-tels=[1,2,3]",
+        ]
+        assert run_tool(Stage1Tool(), argv=argv, cwd=dl1_tmp_path) == 0
+        return output
+
+
+@pytest.fixture(scope="session")
+def dl1_image_file(dl1_tmp_path,):
+    """
+    DL1 file containing only images (DL1A) from a gamma simulation set.
+    """
+    from ctapipe.tools.stage1 import Stage1Tool
+    from ctapipe.core import run_tool
+
+    output = dl1_tmp_path / "images.dl1.h5"
+
+    # prevent running stage1 multiple times in case of parallel tests
+    with FileLock(output.with_suffix(output.suffix + ".lock")):
+        if output.is_file():
+            return output
+
+        infile = get_dataset_path("gamma_test_large.simtel.gz")
+        argv = [
+            f"--input={infile}",
+            f"--output={output}",
+            "--write-images",
+            "--DataWriter.write_parameters=False",
+            "--max-events=20",
+            "--allowed-tels=[1,2,3]",
+        ]
+        assert run_tool(Stage1Tool(), argv=argv, cwd=dl1_tmp_path) == 0
+        return output
+
+
+@pytest.fixture(scope="session")
+def dl1_parameters_file(dl1_tmp_path):
+    """
+    DL1 File containing only parameters (DL1B) from a gamma simulation set.
+    """
+    from ctapipe.tools.stage1 import Stage1Tool
+    from ctapipe.core import run_tool
+
+    output = dl1_tmp_path / "parameters.dl1.h5"
+
+    # prevent running stage1 multiple times in case of parallel tests
+    with FileLock(output.with_suffix(output.suffix + ".lock")):
+        if output.is_file():
+            return output
+
+        infile = get_dataset_path("gamma_test_large.simtel.gz")
+        argv = [
+            f"--input={infile}",
+            f"--output={output}",
+            "--write-parameters",
+            "--max-events=20",
+            "--allowed-tels=[1,2,3]",
+        ]
+        assert run_tool(Stage1Tool(), argv=argv, cwd=dl1_tmp_path) == 0
+        return output
+
+
+@pytest.fixture(scope="session")
+def dl1_muon_file(dl1_tmp_path):
+    """
+    DL1 file containing only images from a muon simulation set.
+    """
+    from ctapipe.tools.stage1 import Stage1Tool
+    from ctapipe.core import run_tool
+
+    output = dl1_tmp_path / "muons.dl1.h5"
+
+    # prevent running stage1 multiple times in case of parallel tests
+    with FileLock(output.with_suffix(output.suffix + ".lock")):
+        if output.is_file():
+            return output
+
+        infile = get_dataset_path("lst_muons.simtel.zst")
+        argv = [
+            f"--input={infile}",
+            f"--output={output}",
+            "--write-images",
+            "--DataWriter.write_parameters=False",
+        ]
+        assert run_tool(Stage1Tool(), argv=argv, cwd=dl1_tmp_path) == 0
+        return output

ctapipe/io/hdf5tableio.py

@@ -377,6 +377,8 @@ def __init__(self, filename, **kwargs):
 
         super().__init__()
         self._tables = {}
+        self._cols_to_read = {}
+        self._missing_cols = {}
         kwargs.update(mode="r")
 
         if isinstance(filename, str) or isinstance(filename, PurePath):
@@ -447,7 +449,10 @@ def _map_table_to_containers(
         by comparing their names including an optional prefix."""
         tab = self._tables[table_name]
         self._cols_to_read[table_name] = []
+        self._missing_cols[table_name] = []
         for container, prefix in zip(containers, prefixes):
+            self._missing_cols[table_name].append([])
+
             for colname in tab.colnames:
                 if prefix and colname.startswith(prefix):
                     colname_without_prefix = colname[len(prefix) + 1 :]
@@ -473,6 +478,7 @@ def _map_table_to_containers(
                     colname_with_prefix = colname
 
                 if colname_with_prefix not in self._cols_to_read[table_name]:
+                    self._missing_cols[table_name][-1].append(colname)
                     self.log.warning(
                         f"Table {table_name} is missing column {colname_with_prefix} "
                         f"that is in container {container.__class__.__name__}. "
@@ -540,7 +546,9 @@ def read(self, table_name, containers, prefixes=False, ignore_columns=None):
                 row = tab[row_count]
             except IndexError:
                 return  # stop generator when done
-            for container, prefix in zip(containers, prefixes):
+
+            missing = self._missing_cols[table_name]
+            for container, prefix, missing_cols in zip(containers, prefixes, missing):
                 for fieldname in container.keys():
                     if prefix:
                         colname = f"{prefix}_{fieldname}"
@@ -551,6 +559,11 @@ def read(self, table_name, containers, prefixes=False, ignore_columns=None):
                     container[fieldname] = self._apply_col_transform(
                         table_name, colname, row[colname]
                     )
+
+                # set missing fields to None
+                for fieldname in missing_cols:
+                    container[fieldname] = None
+
             if return_iterable:
                 yield containers
             else:

ctapipe/io/tableio.py

@@ -210,7 +210,6 @@ class TableReader(Component, metaclass=ABCMeta):
 
     def __init__(self):
         super().__init__()
-        self._cols_to_read = defaultdict(dict)
         self._transforms = defaultdict(dict)
 
     def __enter__(self):

ctapipe/io/tests/test_dl1eventsource.py

@@ -3,7 +3,6 @@
 from ctapipe.io.dl1eventsource import DL1EventSource
 from ctapipe.io import EventSource
 import astropy.units as u
-import subprocess
 import numpy as np
 import tempfile
 import pytest
@@ -12,27 +11,8 @@
 
 
 @pytest.fixture(scope="module")
-def dl1_file():
-    simtel_path = get_dataset_path("gamma_test_large.simtel.gz")
-    command = f"ctapipe-stage1 --input {simtel_path} --output {d.name}/testfile.dl1.h5 --write-parameters --write-images --max-events 20 --allowed-tels=[1,2,3]"
-    subprocess.call(command.split(), stdout=subprocess.PIPE)
-    return f"{d.name}/testfile.dl1.h5"
-
-
-@pytest.fixture(scope="module")
-def dl1b_only_file():
-    simtel_path = get_dataset_path("gamma_test_large.simtel.gz")
-    command = f"ctapipe-stage1 --input {simtel_path} --output {d.name}/testfile_only_b.dl1.h5 --write-parameters --max-events 20 --allowed-tels=[1,2,3]"
-    subprocess.call(command.split(), stdout=subprocess.PIPE)
-    return f"{d.name}/testfile_only_b.dl1.h5"
-
-
-@pytest.fixture(scope="module")
-def dl1a_only_file():
-    simtel_path = get_dataset_path("gamma_test_large.simtel.gz")
-    command = f"ctapipe-stage1 --input {simtel_path} --output {d.name}/testfile_only_a.dl1.h5 --write-images --max-events 20 --allowed-tels=[1,2,3]"
-    subprocess.call(command.split(), stdout=subprocess.PIPE)
-    return f"{d.name}/testfile_only_a.dl1.h5"
+def dl1_dir(tmp_path_factory):
+    return tmp_path_factory.mktemp("dl1")
 
 
 def test_is_compatible(dl1_file):
@@ -92,15 +72,15 @@ def test_simulation_info(dl1_file):
                 assert event.simulation.tel[tel].true_parameters.hillas.x != np.nan
 
 
-def test_dl1_a_only_data(dl1a_only_file):
-    with DL1EventSource(input_url=dl1a_only_file) as source:
+def test_dl1_a_only_data(dl1_image_file):
+    with DL1EventSource(input_url=dl1_image_file) as source:
         for event in source:
             for tel in event.dl1.tel:
                 assert event.dl1.tel[tel].image.any()
 
 
-def test_dl1_b_only_data(dl1b_only_file):
-    with DL1EventSource(input_url=dl1b_only_file) as source:
+def test_dl1_b_only_data(dl1_parameters_file):
+    with DL1EventSource(input_url=dl1_parameters_file) as source:
         for event in source:
             for tel in event.dl1.tel:
                 assert event.dl1.tel[tel].parameters.hillas.x != np.nan

ctapipe/tools/tests/test_tools.py

@@ -3,8 +3,6 @@
 """
 import shlex
 import sys
-import subprocess
-import pytest
 
 import matplotlib as mpl
 
@@ -22,63 +20,6 @@
 LST_MUONS = get_dataset_path("lst_muons.simtel.zst")
 
 
-@pytest.fixture(scope="module")
-def dl1_tmp_path(tmp_path_factory):
-    return tmp_path_factory.mktemp("dl1")
-
-
-@pytest.fixture(scope="module")
-def dl1_image_file(dl1_tmp_path):
-    """
-    DL1 file containing only images (DL1A) from a gamma simulation set.
-    """
-    output = dl1_tmp_path / "images.dl1.h5"
-    command = [
-        "ctapipe-stage1",
-        f"--input={GAMMA_TEST_LARGE}",
-        f"--output={output}",
-        "--write-images",
-        "--max-events=20",
-        "--allowed-tels=[1,2,3]",
-    ]
-    subprocess.run(command, stdout=subprocess.PIPE, check=True)
-    return output
-
-
-@pytest.fixture(scope="module")
-def dl1_parameters_file(dl1_tmp_path):
-    """
-    DL1 File containing only parameters (DL1B) from a gamma simulation set.
-    """
-    output = dl1_tmp_path / "parameters.dl1.h5"
-    command = [
-        "ctapipe-stage1",
-        f"--input={GAMMA_TEST_LARGE}",
-        f"--output={output}",
-        "--write-parameters",
-        "--max-events=20",
-        "--allowed-tels=[1,2,3]",
-    ]
-    subprocess.run(command, stdout=subprocess.PIPE, check=True)
-    return output
-
-
-@pytest.fixture(scope="module")
-def dl1_muon_file(dl1_tmp_path):
-    """
-    DL1 file containing only images from a muon simulation set.
-    """
-    output = dl1_tmp_path / "muons.dl1.h5"
-    command = [
-        "ctapipe-stage1",
-        f"--input={LST_MUONS}",
-        f"--output={output}",
-        "--write-images",
-    ]
-    subprocess.run(command, stdout=subprocess.PIPE, check=True)
-    return output
-
-
 def test_stage_1_dl1(tmp_path, dl1_image_file, dl1_parameters_file):
     from ctapipe.tools.stage1 import Stage1Tool
 
@@ -208,7 +149,7 @@ def _generator(self):
     assert isinstance(tool.event_source, DummyEventSource)
 
 
-def test_muon_reconstruction(tmp_path, dl1_muon_file):
+def test_muon_reconstruction_simtel(tmp_path):
     from ctapipe.tools.muon_reconstruction import MuonAnalysis
 
     muon_simtel_output_file = tmp_path / "muon_reco_on_simtel.h5"
@@ -230,6 +171,10 @@ def test_muon_reconstruction(tmp_path, dl1_muon_file):
         assert len(table) > 20
         assert np.count_nonzero(np.isnan(table["muonring_radius"])) == 0
 
+
+def test_muon_reconstruction_dl1(tmp_path, dl1_muon_file):
+    from ctapipe.tools.muon_reconstruction import MuonAnalysis
+
     muon_dl1_output_file = tmp_path / "muon_reco_on_dl1a.h5"
     assert (
         run_tool(
@@ -330,16 +275,12 @@ def test_display_dl1(tmp_path, dl1_image_file, dl1_parameters_file):
         )
         == 0
     )
-    # test DL1B
-    assert (
-        run_tool(
-            DisplayDL1Calib(),
-            argv=shlex.split(
-                f"--input {dl1_parameters_file} --max_events=1 " "--telescope=11 "
-            ),
-        )
-        == 1
+    # test DL1B, should error since nothing to plot
+    ret = run_tool(
+        DisplayDL1Calib(),
+        argv=[f"--input={dl1_parameters_file}", "--max_events=1", "--telescope=11"],
     )
+    assert ret == 1
     assert run_tool(DisplayDL1Calib(), ["--help-all"]) == 0
 
 

ctapipe/utils/download.py

@@ -4,8 +4,7 @@
 import logging
 from tqdm.auto import tqdm
 from urllib.parse import urlparse
-import time
-from contextlib import contextmanager
+from .filelock import FileLock
 
 __all__ = ["download_file", "download_cached", "download_file_cached"]
 
@@ -77,30 +76,14 @@ def get_cache_path(url, cache_name="ctapipe", env_override="CTAPIPE_CACHE"):
     return path
 
 
-@contextmanager
-def file_lock(path):
-    # if the file already exists, we wait until it does not exist anymore
-    if path.is_file():
-        log.warning("Another download for this file is already running, waiting.")
-        while path.is_file():
-            time.sleep(0.1)
-
-    # create the lock_file file
-    path.open("w").close()
-    try:
-        yield
-    finally:
-        path.unlink()
-
-
 def download_cached(
     url, cache_name="ctapipe", auth=None, env_prefix="CTAPIPE_DATA_", progress=False
 ):
     path = get_cache_path(url, cache_name=cache_name)
     path.parent.mkdir(parents=True, exist_ok=True)
     lock_file = path.with_suffix(path.suffix + ".lock")
 
-    with file_lock(lock_file):
+    with FileLock(lock_file):
         # if we already dowloaded the file, just use it
         if path.is_file():
             log.debug(f"{url} is available in cache.")