Skip to content

Commit

Permalink
I removed the Flatfield and pedestal interpolators
Browse files Browse the repository at this point in the history
  • Loading branch information
Christoph Toennis committed Aug 8, 2024
1 parent e9a2112 commit a5569c6
Show file tree
Hide file tree
Showing 3 changed files with 0 additions and 165 deletions.
4 changes: 0 additions & 4 deletions src/ctapipe/io/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,6 @@
from .interpolation import (
Interpolator,
PointingInterpolator,
FlatFieldInterpolator,
PedestalInterpolator,
)

__all__ = [
Expand All @@ -44,6 +42,4 @@
"get_hdf5_datalevels",
"Interpolator",
"PointingInterpolator",
"PedestalInterpolator",
"FlatFieldInterpolator",
]
116 changes: 0 additions & 116 deletions src/ctapipe/io/interpolation.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,8 +11,6 @@

from .astropy_helpers import read_table

dimensionless_unit = u.Unit()


class StepFunction:

Expand Down Expand Up @@ -231,117 +229,3 @@ def add_table(self, tel_id, input_table):
self._interpolators[tel_id] = {}
self._interpolators[tel_id]["az"] = interp1d(mjd, az, **self.interp_options)
self._interpolators[tel_id]["alt"] = interp1d(mjd, alt, **self.interp_options)


class FlatFieldInterpolator(Interpolator):
"""
Interpolator for flatfield data
"""

telescope_data_group = "dl1/calibration/gain" # TBD
required_columns = frozenset(["time", "gain"]) # TBD
expected_units = {"gain": dimensionless_unit}

def __call__(self, tel_id, time):
"""
Interpolate flatfield data for a given time and tel_id.
Parameters
----------
tel_id : int
telescope id
time : astropy.time.Time
time for which to interpolate the calibration data
Returns
-------
ffield : array [float]
interpolated flatfield data
"""

self._check_interpolators(tel_id)

ffield = self._interpolators[tel_id]["gain"](time)
return ffield

def add_table(self, tel_id, input_table):
"""
Add a table to this interpolator
Parameters
----------
tel_id : int
Telescope id
input_table : astropy.table.Table
Table of pointing values, expected columns
are ``time`` as ``Time`` column and "gain"
for the flatfield data
"""

self._check_tables(input_table)

input_table = input_table.copy()
input_table.sort("time")
time = input_table["time"]
gain = input_table["gain"]
self._interpolators[tel_id] = {}
self._interpolators[tel_id]["gain"] = StepFunction(
time, gain, **self.interp_options
)


class PedestalInterpolator(Interpolator):
"""
Interpolator for Pedestal data
"""

telescope_data_group = "dl1/calibration/pedestal" # TBD
required_columns = frozenset(["time", "pedestal"]) # TBD
expected_units = {"pedestal": dimensionless_unit}

def __call__(self, tel_id, time):
"""
Interpolate pedestal or gain for a given time and tel_id.
Parameters
----------
tel_id : int
telescope id
time : astropy.time.Time
time for which to interpolate the calibration data
Returns
-------
pedestal : array [float]
interpolated pedestal values
"""

self._check_interpolators(tel_id)

pedestal = self._interpolators[tel_id]["pedestal"](time)
return pedestal

def add_table(self, tel_id, input_table):
"""
Add a table to this interpolator
Parameters
----------
tel_id : int
Telescope id
input_table : astropy.table.Table
Table of pointing values, expected columns
are ``time`` as ``Time`` column and "pedestal"
for the pedestal data
"""

self._check_tables(input_table)

input_table = input_table.copy()
input_table.sort("time")
time = input_table["time"]
pedestal = input_table["pedestal"]
self._interpolators[tel_id] = {}
self._interpolators[tel_id]["pedestal"] = StepFunction(
time, pedestal, **self.interp_options
)
45 changes: 0 additions & 45 deletions src/ctapipe/io/tests/test_interpolator.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,6 @@
from astropy.time import Time

from ctapipe.io.interpolation import (
FlatFieldInterpolator,
PedestalInterpolator,
PointingInterpolator,
)

Expand Down Expand Up @@ -101,73 +99,30 @@ def test_bounds():
},
)

table_pedestal = Table(
{
"time": np.arange(0.0, 10.1, 2.0),
"pedestal": np.reshape(np.random.normal(4.0, 1.0, 1850 * 6), (6, 1850))
* u.Unit(),
},
)

table_flatfield = Table(
{
"time": np.arange(0.0, 10.1, 2.0),
"gain": np.reshape(np.random.normal(1.0, 1.0, 1850 * 6), (6, 1850))
* u.Unit(),
},
)

interpolator_pointing = PointingInterpolator()
interpolator_pedestal = PedestalInterpolator()
interpolator_flatfield = FlatFieldInterpolator()
interpolator_pointing.add_table(1, table_pointing)
interpolator_pedestal.add_table(1, table_pedestal)
interpolator_flatfield.add_table(1, table_flatfield)

error_message = "below the interpolation range"

with pytest.raises(ValueError, match=error_message):
interpolator_pointing(tel_id=1, time=t0 - 0.1 * u.s)

with pytest.raises(ValueError, match=error_message):
interpolator_pedestal(tel_id=1, time=-0.1)

with pytest.raises(ValueError, match=error_message):
interpolator_flatfield(tel_id=1, time=-0.1)

with pytest.raises(ValueError, match="above the interpolation range"):
interpolator_pointing(tel_id=1, time=t0 + 10.2 * u.s)

alt, az = interpolator_pointing(tel_id=1, time=t0 + 1 * u.s)
assert u.isclose(alt, 69 * u.deg)
assert u.isclose(az, 1 * u.deg)

pedestal = interpolator_pedestal(tel_id=1, time=1.0)
assert all(pedestal == table_pedestal["pedestal"][0])
flatfield = interpolator_flatfield(tel_id=1, time=1.0)
assert all(flatfield == table_flatfield["gain"][0])
with pytest.raises(KeyError):
interpolator_pointing(tel_id=2, time=t0 + 1 * u.s)
with pytest.raises(KeyError):
interpolator_pedestal(tel_id=2, time=1.0)
with pytest.raises(KeyError):
interpolator_flatfield(tel_id=2, time=1.0)

interpolator_pointing = PointingInterpolator(bounds_error=False)
interpolator_pedestal = PedestalInterpolator(bounds_error=False)
interpolator_flatfield = FlatFieldInterpolator(bounds_error=False)
interpolator_pointing.add_table(1, table_pointing)
interpolator_pedestal.add_table(1, table_pedestal)
interpolator_flatfield.add_table(1, table_flatfield)

for dt in (-0.1, 10.1) * u.s:
alt, az = interpolator_pointing(tel_id=1, time=t0 + dt)
assert np.isnan(alt.value)
assert np.isnan(az.value)

assert all(np.isnan(interpolator_pedestal(tel_id=1, time=-0.1)))
assert all(np.isnan(interpolator_flatfield(tel_id=1, time=-0.1)))

interpolator_pointing = PointingInterpolator(bounds_error=False, extrapolate=True)
interpolator_pointing.add_table(1, table_pointing)
alt, az = interpolator_pointing(tel_id=1, time=t0 - 1 * u.s)
Expand Down

0 comments on commit a5569c6

Please sign in to comment.