From 2ff4d5ade9663d12e526dc9d6cb20befd20e1b10 Mon Sep 17 00:00:00 2001 From: Nabil Freij Date: Wed, 8 Jul 2026 15:18:19 -0700 Subject: [PATCH] make channels time snapshots via Channel.at(obstime) --- sunkit_instruments/response/abstractions.py | 66 ++++++++++++------- .../response/tests/test_channel.py | 42 +++++++++++- sunkit_instruments/response/thermal.py | 19 +++--- 3 files changed, 93 insertions(+), 34 deletions(-) diff --git a/sunkit_instruments/response/abstractions.py b/sunkit_instruments/response/abstractions.py index 41a148d1..754b24f7 100644 --- a/sunkit_instruments/response/abstractions.py +++ b/sunkit_instruments/response/abstractions.py @@ -1,8 +1,11 @@ """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"] @@ -10,30 +13,52 @@ 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 @@ -41,27 +66,24 @@ def wavelength_response( ) @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 diff --git a/sunkit_instruments/response/tests/test_channel.py b/sunkit_instruments/response/tests/test_channel.py index 69a95f80..1a9e3eef 100644 --- a/sunkit_instruments/response/tests/test_channel.py +++ b/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)" diff --git a/sunkit_instruments/response/thermal.py b/sunkit_instruments/response/thermal.py index 7de91ee5..1d34965f 100644 --- a/sunkit_instruments/response/thermal.py +++ b/sunkit_instruments/response/thermal.py @@ -8,7 +8,7 @@ __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. @@ -16,8 +16,9 @@ def get_temperature_response(channel, spectra, obstime=None): 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'],