Merge pull request #212 from cta-observatory/ignore_missing_end

Ignore missing TrackEnd when parsin TargetLog file by default

Author: maxnoe

src/ctapipe_io_lst/pointing.py

@@ -1,4 +1,5 @@
 from pathlib import Path
+import warnings
 
 import numpy as np
 from scipy.interpolate import interp1d
@@ -69,7 +70,7 @@ def __init__(self, subarray, config=None, parent=None, **kwargs):
         self.interp_alt = {}
 
     @staticmethod
-    def _read_target_log(path):
+    def read_target_log(path, ignore_missing_end=True):
         path = Path(path)
 
         def parse_start(tokens):
@@ -103,10 +104,19 @@ def parse_end(tokens):
 
                 tokens = line.strip().split(" ")
                 if tokens[1] == "TrackStart":
+                    start = parse_start(tokens)
+
                     if tracking:
-                        raise ValueError(f"Expected TrackingEnd, got {line}")
+                        msg = f"Expected TrackingEnd, got {line}"
+                        if ignore_missing_end:
+                            warnings.warn(msg)
+                            # let previous target end one second before new one
+                            targets[-1].update({"end_unix": start["start_unix"] - 1})
+                        else:
+                            raise ValueError(msg)
+
                     tracking = True
-                    targets.append(parse_start(tokens))
+                    targets.append(start)
 
                 elif tokens[1] == "TrackEnd":
                     if not tracking:
@@ -282,7 +292,7 @@ def get_target(self, tel_id, time):
             if path is None:
                 self.target_log[tel_id] = None
             else:
-                self.target_log[tel_id] = self._read_target_log(path)
+                self.target_log[tel_id] = self.read_target_log(path)
 
         targets = self.target_log[tel_id]
         if targets is None:

src/ctapipe_io_lst/tests/test_pointing.py

@@ -1,6 +1,7 @@
 import os
 from pathlib import Path
 import numpy as np
+import pytest
 from astropy.time import Time
 import astropy.units as u
 from ctapipe.core import Provenance
@@ -69,7 +70,7 @@ def test_load_position_and_bending_corrections():
 def test_read_target_log(tmp_path):
     from ctapipe_io_lst.pointing import PointingSource
 
-    targets = PointingSource._read_target_log(test_target_log)
+    targets = PointingSource.read_target_log(test_target_log)
     assert len(targets) == 7
     assert targets.colnames == ["start_unix", "ra", "dec", "name", "end_unix", "start", "end"]
 
@@ -83,7 +84,7 @@ def test_read_target_log(tmp_path):
     # test with empty file
     empty_log = (tmp_path / "Target_log.txt")
     empty_log.open("w").close()
-    targets = PointingSource._read_target_log(empty_log)
+    targets = PointingSource.read_target_log(empty_log)
     assert len(targets) == 0
     assert targets.colnames == ["start_unix", "ra", "dec", "name", "end_unix", "start", "end"]
     assert targets["ra"].unit == u.deg
@@ -99,13 +100,28 @@ def test_read_target_log(tmp_path):
                 tokens = tokens[:-1]
             f.write(" ".join(tokens) + "\n")
 
-    targets = PointingSource._read_target_log(log_no_names)
+    targets = PointingSource.read_target_log(log_no_names)
     assert len(targets) == 7
     assert targets.colnames == ["start_unix", "ra", "dec", "name", "end_unix", "start", "end"]
 
     np.testing.assert_array_equal(targets["name"], ["unknown"] * 7)
     np.testing.assert_array_equal(targets["ra"], [83.6296, 86.6333] * 3 + [79.1725])
 
+    # test with missing TrackEnd line
+    log_missing_end = tmp_path / "target_log_missing_end.txt"
+    log_lines.pop(1)
+    log_missing_end.write_text('\n'.join(log_lines))
+
+    with pytest.warns(match="Expected TrackingEnd"):
+        targets = PointingSource.read_target_log(log_missing_end)
+
+    assert len(targets) == 7
+    assert targets[0]['end_unix'] == targets[1]['start_unix'] - 1
+
+    with pytest.raises(ValueError, match="Expected TrackingEnd"):
+        targets = PointingSource.read_target_log(log_missing_end, ignore_missing_end=False)
+
+
 
 def test_targets():
     from ctapipe_io_lst import PointingSource, LSTEventSource