Make channels time snapshots via Channel.at(obstime)

sunkit_instruments/response/abstractions.py

@@ -1,67 +1,89 @@
 """This module defines abstractions for computing instrument response."""
 import abc
+import copy
 
 import astropy.units as u
 
+from sunpy.time import parse_time
+
 __all__ = ["AbstractChannel"]
 
 
 class AbstractChannel(abc.ABC):
     """
     An abstract base class for defining instrument channels.
 
+    A channel instance can be a snapshot of the instrument at one time:
+    time-dependent degradation enters exclusively through the ``at`` method.
+    Subclasses implement the time dependence in ``degradation``.
+
     For all methods and properties defined here, see the
     topic guide on instrument response for more information.
     """
 
+    _obstime = None
+
+    def at(self, obstime):
+        """
+        A copy of this channel evaluated at ``obstime``.
+
+        The single way time enters any calculation.
+        Bind the time first, then pass the returned channel wherever a channel is
+        accepted:
+
+        .. code-block:: python
+
+            get_temperature_response(channel.at("2020-01-01"), spectra)
+
+        Parameters
+        ----------
+        obstime: any format parsed by `sunpy.time.parse_time`
+        """
+        bound = copy.copy(self)
+        bound._obstime = parse_time(obstime) if obstime is not None else None
+        return bound
+
+    def __repr__(self):
+        obstime = "Non-degraded" if self._obstime is None else f"Degraded at {self._obstime.isot}"
+        return f"{type(self).__name__}({obstime})"
+
     @u.quantity_input
-    def wavelength_response(
-        self, obstime=None
-    ) -> u.cm**2 * u.DN * u.steradian / (u.photon * u.pixel):
+    def wavelength_response(self) -> u.cm**2 * u.DN * u.steradian / (u.photon * u.pixel):
         """
         Instrument response as a function of wavelength
 
         The wavelength response is the effective area with
         the conversion factors from photons to DN and steradians
-        to pixels.
-
-        Parameters
-        ----------
-        obstime: any format parsed by `~sunpy.time.parse_time`, optional
-            If specified, this is used to compute the time-dependent
-            instrument degradation.
+        to pixels. For time-dependent degradation, bind the time first
+        with `at`.
         """
-        area_eff = self.effective_area(obstime=obstime)
         return (
-            area_eff
+            self.effective_area()
             * self.energy_per_photon
             * self.pixel_solid_angle
             * self.camera_gain
             / self.energy_per_electron
         )
 
     @u.quantity_input
-    def effective_area(self, obstime=None) -> u.cm**2:
+    def effective_area(self) -> u.cm**2:
         """
         Effective area as a function of wavelength.
 
         The effective area is the geometrical collecting area
         weighted by the mirror reflectance, filter transmittance,
-        quantum efficiency, and instrument degradation.
-
-        Parameters
-        ----------
-        obstime: any format parsed by `sunpy.time.parse_time`, optional
-            If specified, this is used to compute the time-dependent
-            instrument degradation.
+        quantum efficiency, and instrument degradation evaluated at the
+        time bound with ``at``. An unbound channel never evaluates degradation.
         """
-        return (
+        area = (
             self.geometrical_area
             * self.mirror_reflectance
             * self.filter_transmittance
             * self.effective_quantum_efficiency
-            * self.degradation(obstime=obstime)
         )
+        if self._obstime is None:
+            return area
+        return area * self.degradation(obstime=self._obstime)
 
     @property
     @u.quantity_input

sunkit_instruments/response/tests/test_channel.py

@@ -87,6 +87,46 @@ def test_spectra_repr(fake_spectra):
 
 @pytest.mark.parametrize('obstime', [None, '2020-01-01'])
 def test_temperature_response(fake_channel, fake_spectra, obstime):
-    temp_response = fake_spectra.temperature_response(fake_channel, obstime=obstime)
+    channel = fake_channel if obstime is None else fake_channel.at(obstime)
+    temp_response = fake_spectra.temperature_response(channel)
     assert isinstance(temp_response, u.Quantity)
     assert temp_response.shape == fake_spectra.temperature.shape
+
+
+class TimeDependentChannel(TestChannel):
+    @u.quantity_input
+    def degradation(self, obstime=None) -> u.dimensionless_unscaled:
+        return 2.0 if obstime is not None else 1.0
+
+
+def test_at_binds_obstime():
+    channel = TimeDependentChannel()
+    bound = channel.at("2020-01-01")
+    wave_response = channel.wavelength_response()
+    assert u.allclose(bound.wavelength_response(), 2 * wave_response, atol=1e-20 * wave_response.unit)
+    area = channel.effective_area()
+    assert u.allclose(bound.effective_area(), 2 * area, atol=1e-20 * area.unit)
+    # Binding returns a copy; the original stays unbound
+    assert channel._obstime is None
+    assert u.allclose(bound.wavelength, channel.wavelength)
+
+
+class DegradationRequiresTimeChannel(TestChannel):
+    @u.quantity_input
+    def degradation(self, obstime=None) -> u.dimensionless_unscaled:
+        if obstime is None:
+            raise ValueError("obstime required")
+        return 0.5
+
+
+def test_unbound_channel_never_evaluates_degradation():
+    channel = DegradationRequiresTimeChannel()
+    pristine = channel.effective_area()
+    assert u.allclose(channel.at(None).effective_area(), pristine)
+    assert u.allclose(channel.at("2020-01-01").effective_area(), 0.5 * pristine)
+
+
+def test_repr_shows_binding(fake_channel):
+    assert repr(fake_channel) == "TestChannel(Non-degraded)"
+    assert repr(fake_channel.at("2020-01-01")) == "TestChannel(Degraded at 2020-01-01T00:00:00.000)"
+    assert repr(fake_channel.at("2020-01-01").at(None)) == "TestChannel(Non-degraded)"

sunkit_instruments/response/thermal.py

@@ -8,16 +8,17 @@
 __all__ = ["SourceSpectra", "get_temperature_response"]
 
 
-def get_temperature_response(channel, spectra, obstime=None):
+def get_temperature_response(channel, spectra):
     """
     Calculate the temperature response function for a given instrument channel
     and input spectra.
 
     Parameters
     ----------
     channel: `~sunkit_instruments.response.abstractions.AbstractChannel`
+        For time-dependent instrument degradation, bind the time first:
+        ``get_temperature_response(channel.at(obstime), spectra)``.
     spectra: `~sunkit_instruments.response.SourceSpectra`
-    obstime: any format parsed by `sunpy.time.parse_time` , optional
 
     Returns
     -------
@@ -28,7 +29,7 @@ def get_temperature_response(channel, spectra, obstime=None):
     --------
     sunkit_instruments.response.SourceSpectra.temperature_response
     """
-    return spectra.temperature, spectra.temperature_response(channel, obstime=obstime)
+    return spectra.temperature, spectra.temperature_response(channel)
 
 
 class SourceSpectra:
@@ -136,9 +137,7 @@ def data(self) -> u.photon * u.cm**3 / (u.s * u.Angstrom * u.steradian):
         return u.Quantity(self._da.data, self._da.attrs["unit"])
 
     @u.quantity_input
-    def temperature_response(
-        self, channel, obstime=None
-    ) -> u.cm**5 * u.DN / (u.pixel * u.s):
+    def temperature_response(self, channel) -> u.cm**5 * u.DN / (u.pixel * u.s):
         """
         Temperature response function for a given instrument channel.
 
@@ -151,12 +150,10 @@ def temperature_response(
         ----------
         channel: `~sunkit_instruments.response.abstractions.AbstractChannel`
             The relevant instrument channel object used to compute the wavelength
-            response function.
-        obstime: any format parsed by `sunpy.time.parse_time`, optional
-            A time of a particular observation. This is used to calculated any
-            time-dependent instrument degradation.
+            response function. For time-dependent instrument degradation,
+            bind the time first with ``channel.at(obstime)``.
         """
-        wave_response = channel.wavelength_response(obstime=obstime)
+        wave_response = channel.wavelength_response()
         da_response = xarray.DataArray(
             wave_response.to_value(wave_response.unit),
             dims=['wavelength'],