diff --git a/lisainstrument/containers.py b/lisainstrument/containers.py
index e8c9679bc88adc76af82c7a5794f1863e8b180e1..f8fb9227db7f6ce0b3afdd49075ad47e1b29a170 100644
--- a/lisainstrument/containers.py
+++ b/lisainstrument/containers.py
@@ -137,6 +137,66 @@ class ForEachObject(abc.ABC):
return self.dict == other
return numpy.all([self[index] == other for index in self.indices()])
+ def __abs__(self):
+ return self.transformed(lambda index, value: abs(value))
+
+ def __neg__(self):
+ return self.transformed(lambda index, value: -value)
+
+ def __add__(self, other):
+ if numpy.isscalar(other):
+ return self.transformed(lambda _, value: value + other)
+ if isinstance(other, type(self)):
+ return self.transformed(lambda index, value: value + other[index])
+ raise TypeError(f"unsupported operand type for +: '{type(self)}' and '{type(other)}'")
+
+ def __radd__(self, other):
+ return self + other
+
+ def __sub__(self, other):
+ if numpy.isscalar(other):
+ return self.transformed(lambda _, value: value - other)
+ if isinstance(other, type(self)):
+ return self.transformed(lambda index, value: value - other[index])
+ raise TypeError(f"unsupported operand type for -: '{type(self)}' and '{type(other)}'")
+
+ def __rsub__(self, other):
+ return -(self - other)
+
+ def __mul__(self, other):
+ if numpy.isscalar(other):
+ return self.transformed(lambda _, value: value * other)
+ if isinstance(other, type(self)):
+ return self.transformed(lambda index, value: value * other[index])
+ raise TypeError(f"unsupported operand type for *: '{type(self)}' and '{type(other)}'")
+
+ def __rmul__(self, other):
+ return self * other
+
+ def __floordiv__(self, other):
+ if numpy.isscalar(other):
+ return self.transformed(lambda _, value: value // other)
+ if isinstance(other, type(self)):
+ return self.transformed(lambda index, value: value // other[index])
+ raise TypeError(f"unsupported operand type for //: '{type(self)}' and '{type(other)}'")
+
+ def __truediv__(self, other):
+ if numpy.isscalar(other):
+ return self.transformed(lambda _, value: value / other)
+ if isinstance(other, type(self)):
+ return self.transformed(lambda index, value: value / other[index])
+ raise TypeError(f"unsupported operand type for /: '{type(self)}' and '{type(other)}'")
+
+ def __rtruediv__(self, other):
+ if numpy.isscalar(other):
+ return self.transformed(lambda _, value: other / value)
+ raise TypeError(f"unsupported operand type for /: '{type(self)}' and '{type(other)}'")
+
+ def __pow__(self, other):
+ if numpy.isscalar(other):
+ return self.transformed(lambda _, value: value**other)
+ raise TypeError(f"unsupported operand type for **: '{type(self)}' and '{type(other)}'")
+
def __repr__(self):
return repr(self.dict)
@@ -149,14 +209,14 @@ class ForEachSC(ForEachObject):
return ['1', '2', '3']
@staticmethod
- def distant_left(sc):
+ def distant_left_sc(sc):
"""Return index of distant rleftspacecraft."""
if sc not in ForEachSC.indices():
raise ValueError(f"invalid spacecraft index '{sc}'")
return f'{int(sc) % 3 + 1}'
@staticmethod
- def distant_right(sc):
+ def distant_right_sc(sc):
"""Return index of distant right spacecraft."""
if sc not in ForEachSC.indices():
raise ValueError(f"invalid spacecraft index '{sc}'")
@@ -167,14 +227,43 @@ class ForEachSC(ForEachObject):
"""Return index of left MOSA."""
if sc not in ForEachSC.indices():
raise ValueError(f"invalid spacecraft index '{sc}'")
- return f'{sc}{ForEachSC.distant_left(sc)}'
+ return f'{sc}{ForEachSC.distant_left_sc(sc)}'
@staticmethod
def right_mosa(sc):
"""Return index of right MOSA."""
if sc not in ForEachSC.indices():
raise ValueError(f"invalid spacecraft index '{sc}'")
- return f'{sc}{ForEachSC.distant_right(sc)}'
+ return f'{sc}{ForEachSC.distant_right_sc(sc)}'
+
+ def for_each_mosa(self):
+ """Return a ForEachMOSA instance by sharing the spacecraft values on both MOSAs."""
+ return ForEachMOSA(lambda mosa: self[ForEachMOSA.sc(mosa)])
+
+ def __add__(self, other):
+ if isinstance(other, ForEachMOSA):
+ return self.for_each_mosa() + other
+ return super().__add__(other)
+
+ def __sub__(self, other):
+ if isinstance(other, ForEachMOSA):
+ return self.for_each_mosa() - other
+ return super().__sub__(other)
+
+ def __mul__(self, other):
+ if isinstance(other, ForEachMOSA):
+ return self.for_each_mosa() * other
+ return super().__mul__(other)
+
+ def __floordiv__(self, other):
+ if isinstance(other, ForEachMOSA):
+ return self.for_each_mosa() // other
+ return super().__floordiv__(other)
+
+ def __truediv__(self, other):
+ if isinstance(other, ForEachMOSA):
+ return self.for_each_mosa() / other
+ return super().__truediv__(other)
class ForEachMOSA(ForEachObject):
@@ -184,20 +273,13 @@ class ForEachMOSA(ForEachObject):
def indices(cls):
return ['12', '23', '31', '13', '32', '21']
- def __init__(self, values):
- """Initialize from a `ForEachSC` instance or rely on super's implementation."""
- if isinstance(values, ForEachSC):
- super().__init__(lambda mosa: values[ForEachMOSA.sc(mosa)])
- else:
- super().__init__(values)
-
@staticmethod
def sc(mosa):
"""Return index of spacecraft hosting MOSA."""
return f'{mosa[0]}'
@staticmethod
- def distant(mosa):
+ def distant_mosa(mosa):
"""Return index of distant MOSA.
In practive, we invert the indices to swap emitter and receiver.
@@ -207,7 +289,7 @@ class ForEachMOSA(ForEachObject):
return f'{mosa[1]}{mosa[0]}'
@staticmethod
- def adjacent(mosa):
+ def adjacent_mosa(mosa):
"""Return index of adjacent MOSA.
In practice, we replace the second index by the only unused spacecraft index.
@@ -218,3 +300,36 @@ class ForEachMOSA(ForEachObject):
if len(unused) != 1:
raise RuntimeError(f"cannot find adjacent MOSA for '{mosa}'")
return f'{mosa[0]}{unused[0]}'
+
+ def distant(self):
+ """Return a ForEachMOSA instance for distant MOSAs."""
+ return ForEachMOSA(lambda mosa: self[ForEachMOSA.distant_mosa(mosa)])
+
+ def adjacent(self):
+ """Return a ForEachMOSA instance for adjacent MOSAs."""
+ return ForEachMOSA(lambda mosa: self[ForEachMOSA.adjacent_mosa(mosa)])
+
+ def __add__(self, other):
+ if isinstance(other, ForEachSC):
+ return self + other.for_each_mosa()
+ return super().__add__(other)
+
+ def __sub__(self, other):
+ if isinstance(other, ForEachSC):
+ return self - other.for_each_mosa()
+ return super().__sub__(other)
+
+ def __mul__(self, other):
+ if isinstance(other, ForEachSC):
+ return self * other.for_each_mosa()
+ return super().__mul__(other)
+
+ def __floordiv__(self, other):
+ if isinstance(other, ForEachSC):
+ return self // other.for_each_mosa()
+ return super().__floordiv__(other)
+
+ def __truediv__(self, other):
+ if isinstance(other, ForEachSC):
+ return self / other.for_each_mosa()
+ return super().__truediv__(other)
diff --git a/lisainstrument/instrument.py b/lisainstrument/instrument.py
index 178649ec26783d9f7f943065ebcc7041fca8efc8..603db909f21c116840314ab339994e44335ec4a8 100644
--- a/lisainstrument/instrument.py
+++ b/lisainstrument/instrument.py
@@ -195,14 +195,15 @@ class Instrument:
elif callable(interpolation):
logger.info("Using user-provided interpolation function")
self.interpolation_order = None
- self.interpolate = lambda x, shift: x if numpy.isscalar(x) else interpolation(x, shift)
+ self.interpolate = lambda x, shift: x if numpy.isscalar(x) else \
+ interpolation(x, shift * self.physics_fs)
else:
method = str(interpolation[0])
if method == 'lagrange':
self.interpolation_order = int(interpolation[1])
logging.debug("Using Lagrange interpolation of order %s", self.interpolation_order)
self.interpolate = lambda x, shift: x if numpy.isscalar(x) else \
- dsp.timeshift(x, shift, self.interpolation_order)
+ dsp.timeshift(x, shift * self.physics_fs, self.interpolation_order)
else:
raise ValueError(f"invalid interpolation parameters '{interpolation}'")
@@ -413,105 +414,80 @@ class Instrument:
self.local_carrier_offsets = self.offsets_freqs
logger.debug("Computing carrier fluctuations for local beams")
+ self.local_carrier_fluctuations = self.laser_noises + self.glitch_lasers
if self.three_lasers:
- self.local_carrier_fluctuations = ForEachMOSA(lambda mosa:
- self.laser_noises[ForEachMOSA.sc(mosa)] + self.glitch_lasers[ForEachMOSA.sc(mosa)]
- )
- else:
- self.local_carrier_fluctuations = ForEachMOSA(lambda mosa:
- self.laser_noises[mosa] + self.glitch_lasers[mosa]
- )
+ self.local_carrier_fluctuations = self.local_carrier_fluctuations.for_each_mosa()
logger.debug("Computing upper sideband offsets for local beams")
- self.local_usb_offsets = ForEachMOSA(lambda mosa:
- self.offsets_freqs[mosa] + self.modulation_freqs[mosa] * (1 + self.clock_noise_offsets[mosa[0]])
- )
+ self.local_usb_offsets = \
+ self.offsets_freqs + self.modulation_freqs * (1 + self.clock_noise_offsets)
logger.debug("Computing upper sideband fluctuations for local beams")
- if self.three_lasers:
- self.local_usb_fluctuations = ForEachMOSA(lambda mosa:
- self.laser_noises[ForEachMOSA.sc(mosa)] + self.glitch_lasers[ForEachMOSA.sc(mosa)]
- + self.modulation_freqs[mosa] * (self.clock_noise_fluctuations[mosa[0]] + self.modulation_noises[mosa])
- )
- else:
- self.local_usb_fluctuations = ForEachMOSA(lambda mosa:
- self.laser_noises[mosa] + self.glitch_lasers[mosa] \
- + self.modulation_freqs[mosa] * (self.clock_noise_fluctuations[mosa[0]] + self.modulation_noises[mosa])
- )
+ self.local_usb_fluctuations = \
+ self.laser_noises + self.glitch_lasers \
+ + self.modulation_freqs * (self.clock_noise_fluctuations + self.modulation_noises)
logger.debug("Computing local timer deviations")
- self.local_timer_deviations = ForEachSC(lambda sc:
- self.clock_offsets[sc] + numpy.cumsum(
- numpy.broadcast_to(self.clock_noise_offsets[sc] + self.clock_noise_fluctuations[sc], \
- (self.physics_size,)) * self.dt)
- )
+ self.local_timer_deviations = \
+ self.clock_offsets + ForEachSC(lambda sc:
+ numpy.cumsum(numpy.broadcast_to(
+ self.clock_noise_offsets[sc] + self.clock_noise_fluctuations[sc],
+ self.physics_size)
+ * self.dt)
+ )
## Propagation to distant MOSA
logger.info("Propagating local beams to distant MOSAs")
logger.debug("Propagating carrier offsets to distant MOSAs")
- self.distant_carrier_offsets = ForEachMOSA(lambda mosa:
- (1 - self.d_pprs[mosa]) * \
- self.interpolate(self.local_carrier_offsets[ForEachMOSA.distant(mosa)],
- - self.pprs[mosa] * self.physics_fs) \
- - self.d_pprs[mosa] * self.central_freq
- )
+ self.distant_carrier_offsets = \
+ -self.d_pprs * self.central_freq \
+ + (1 - self.d_pprs) * self.local_carrier_offsets.distant() \
+ .transformed(lambda mosa, x: self.interpolate(x, -self.pprs[mosa]))
logger.debug("Propagating carrier fluctuations to distant MOSAs")
- self.distant_carrier_fluctuations = ForEachMOSA(lambda mosa:
- self.central_freq * self.gws[mosa] + (1 - self.d_pprs[mosa]) * \
- self.interpolate(self.local_carrier_fluctuations[ForEachMOSA.distant(mosa)],
- -self.pprs[mosa] * self.physics_fs)
- )
+ self.distant_carrier_fluctuations = \
+ self.central_freq * self.gws \
+ + (1 - self.d_pprs) * self.local_carrier_fluctuations.distant() \
+ .transformed(lambda mosa, x: self.interpolate(x, -self.pprs[mosa]))
logger.debug("Propagating upper sideband offsets to distant MOSAs")
- self.distant_usb_offsets = ForEachMOSA(lambda mosa:
- (1 - self.d_pprs[mosa]) * \
- self.interpolate(self.local_usb_offsets[ForEachMOSA.distant(mosa)],
- -self.pprs[mosa] * self.physics_fs) \
- - self.d_pprs[mosa] * self.central_freq
- )
+ self.distant_usb_offsets = \
+ -self.d_pprs * self.central_freq \
+ + (1 - self.d_pprs) * self.local_usb_offsets.distant() \
+ .transformed(lambda mosa, x: self.interpolate(x, -self.pprs[mosa]))
logger.debug("Propagating upper sideband fluctuations to distant MOSAs")
- self.distant_usb_fluctuations = ForEachMOSA(lambda mosa:
- self.central_freq * self.gws[mosa] + (1 - self.d_pprs[mosa]) * \
- self.interpolate(self.local_usb_fluctuations[ForEachMOSA.distant(mosa)],
- - self.pprs[mosa] * self.physics_fs)
- )
+ self.distant_usb_fluctuations = \
+ self.central_freq * self.gws \
+ + (1 - self.d_pprs) * self.local_usb_fluctuations.distant() \
+ .transformed(lambda mosa, x: self.interpolate(x, -self.pprs[mosa]))
logger.debug("Propagating timer deviations to distant MOSAs")
- self.distant_timer_deviations = ForEachMOSA(lambda mosa:
- self.interpolate(self.local_timer_deviations[mosa[1]],
- - self.pprs[mosa] * self.physics_fs) \
+ self.distant_timer_deviations = \
+ self.local_timer_deviations.for_each_mosa().distant() \
+ .transformed(lambda mosa, x: self.interpolate(x, -self.pprs[mosa])
- self.pprs[mosa]
)
- ## Propagation to adjacent MOSA
-
- logging.info("Propagating local beams to adjacent MOSAs")
+ logging.info("Propagating local beams to adjacent")
logger.debug("Propagating carrier offsets to adjacent MOSAs")
- self.adjacent_carrier_offsets = ForEachMOSA(lambda mosa:
- self.local_carrier_offsets[ForEachMOSA.adjacent(mosa)]
- )
+ self.adjacent_carrier_offsets = self.local_carrier_offsets.adjacent()
logger.debug("Propagating carrier fluctuations to adjacent MOSAs")
- self.adjacent_carrier_fluctuations = ForEachMOSA(lambda mosa:
- self.local_carrier_fluctuations[ForEachMOSA.adjacent(mosa)] \
- + self.central_freq * self.backlink_noises[mosa]
- )
+ self.adjacent_carrier_fluctuations = \
+ self.local_carrier_fluctuations.adjacent() \
+ + self.central_freq * self.backlink_noises
logger.debug("Propagating upper sideband offsets to adjacent MOSAs")
- self.adjacent_usb_offsets = ForEachMOSA(lambda mosa:
- self.local_usb_offsets[ForEachMOSA.adjacent(mosa)]
- )
+ self.adjacent_usb_offsets = self.local_usb_offsets.adjacent()
logger.debug("Propagating upper sideband fluctuations to adjacent MOSAs")
- self.adjacent_usb_fluctuations = ForEachMOSA(lambda mosa:
- self.local_usb_fluctuations[ForEachMOSA.adjacent(mosa)] \
- + self.central_freq * self.backlink_noises[mosa]
- )
+ self.adjacent_usb_fluctuations = \
+ self.local_usb_fluctuations.adjacent() \
+ + self.central_freq * self.backlink_noises
## Inter-spacecraft interferometer local beams
@@ -550,32 +526,27 @@ class Instrument:
logger.info("Computing inter-spacecraft beatnotes on TPS")
logger.debug("Computing inter-spacecraft carrier beatnote offsets on TPS")
- self.tps_isc_carrier_offsets = ForEachMOSA(lambda mosa:
- self.distant_isc_carrier_offsets[mosa] - self.local_isc_carrier_offsets[mosa]
- )
+ self.tps_isc_carrier_offsets = \
+ self.distant_isc_carrier_offsets - self.local_isc_carrier_offsets
logger.debug("Computing inter-spacecraft carrier beatnote fluctuations on TPS")
- self.tps_isc_carrier_fluctuations = ForEachMOSA(lambda mosa:
- self.distant_isc_carrier_fluctuations[mosa] - self.local_isc_carrier_fluctuations[mosa]
- )
+ self.tps_isc_carrier_fluctuations = \
+ self.distant_isc_carrier_fluctuations - self.local_isc_carrier_fluctuations
logging.debug("Computing inter-spacecraft upper sideband beatnote offsets on TPS")
- self.tps_isc_usb_offsets = ForEachMOSA(lambda mosa:
- self.distant_isc_usb_offsets[mosa] - self.local_isc_usb_offsets[mosa]
- )
+ self.tps_isc_usb_offsets = \
+ self.distant_isc_usb_offsets - self.local_isc_usb_offsets
logger.debug("Computing inter-spacecraft upper sideband beatnote fluctuations on TPS")
- self.tps_isc_usb_fluctuations = ForEachMOSA(lambda mosa:
- self.distant_isc_usb_fluctuations[mosa] - self.local_isc_usb_fluctuations[mosa]
- )
+ self.tps_isc_usb_fluctuations = \
+ self.distant_isc_usb_fluctuations - self.local_isc_usb_fluctuations
## Measured pseudo-ranging on TPS grid (high-frequency)
logger.info("Computing measured pseudo-ranges on TPS")
- self.tps_mprs = ForEachMOSA(lambda mosa:
- self.local_timer_deviations[ForEachMOSA.sc(mosa)] \
- - self.distant_timer_deviations[mosa] + self.ranging_noises[mosa]
- )
+ self.tps_mprs = \
+ self.local_timer_deviations \
+ - self.distant_timer_deviations + self.ranging_noises
## Test-mass interferometer local beams
@@ -585,19 +556,17 @@ class Instrument:
self.local_tm_carrier_offsets = self.local_carrier_offsets
logging.debug("Propagating local carrier fluctuations to test-mass interferometer")
- self.local_tm_carrier_fluctuations = ForEachMOSA(lambda mosa:
- self.local_carrier_fluctuations[mosa]
- + self.central_freq * (self.testmass_noises[mosa] + self.glitch_tms[mosa] / c)
- )
+ self.local_tm_carrier_fluctuations = \
+ self.local_carrier_fluctuations \
+ + self.central_freq * (self.testmass_noises + self.glitch_tms / c)
logger.debug("Propagating local upper sideband offsets to test-mass interferometer")
self.local_tm_usb_offsets = self.local_usb_offsets
logger.debug("Propagating local upper sideband fluctuations to test-mass interferometer")
- self.local_tm_usb_fluctuations = ForEachMOSA(lambda mosa:
- self.local_usb_fluctuations[mosa]
- + self.central_freq * (self.testmass_noises[mosa] + self.glitch_tms[mosa] / c)
- )
+ self.local_tm_usb_fluctuations = \
+ self.local_usb_fluctuations \
+ + self.central_freq * (self.testmass_noises + self.glitch_tms / c)
## Test-mass interferometer adjacent beams
@@ -620,24 +589,20 @@ class Instrument:
logger.info("Computing test-mass beatnotes on TPS")
logger.debug("Computing test-mass carrier beatnote offsets on TPS")
- self.tps_tm_carrier_offsets = ForEachMOSA(lambda mosa:
- self.adjacent_tm_carrier_offsets[mosa] - self.local_tm_carrier_offsets[mosa]
- )
+ self.tps_tm_carrier_offsets = \
+ self.adjacent_tm_carrier_offsets - self.local_tm_carrier_offsets
logger.debug("Computing test-mass carrier beatnote fluctuations on TPS")
- self.tps_tm_carrier_fluctuations = ForEachMOSA(lambda mosa:
- self.adjacent_tm_carrier_fluctuations[mosa] - self.local_tm_carrier_fluctuations[mosa]
- )
+ self.tps_tm_carrier_fluctuations = \
+ self.adjacent_tm_carrier_fluctuations - self.local_tm_carrier_fluctuations
logger.debug("Computing test-mass upper sideband beatnote offsets on TPS")
- self.tps_tm_usb_offsets = ForEachMOSA(lambda mosa:
- self.adjacent_tm_usb_offsets[mosa] - self.local_tm_usb_offsets[mosa]
- )
+ self.tps_tm_usb_offsets = \
+ self.adjacent_tm_usb_offsets - self.local_tm_usb_offsets
logger.debug("Computing test-mass upper sideband beatnote fluctuations on TPS")
- self.tps_tm_usb_fluctuations = ForEachMOSA(lambda mosa:
- self.adjacent_tm_usb_fluctuations[mosa] - self.local_tm_usb_fluctuations[mosa]
- )
+ self.tps_tm_usb_fluctuations = \
+ self.adjacent_tm_usb_fluctuations - self.local_tm_usb_fluctuations
## Reference interferometer local beams
@@ -676,116 +641,111 @@ class Instrument:
logger.info("Computing reference beatnotes on TPS")
logger.debug("Computing reference carrier beatnote offsets on TPS")
- self.tps_ref_carrier_offsets = ForEachMOSA(lambda mosa:
- self.adjacent_ref_carrier_offsets[mosa] - self.local_ref_carrier_offsets[mosa]
- )
+ self.tps_ref_carrier_offsets = \
+ self.adjacent_ref_carrier_offsets - self.local_ref_carrier_offsets
logger.debug("Computing reference carrier beatnote fluctuations on TPS")
- self.tps_ref_carrier_fluctuations = ForEachMOSA(lambda mosa:
- self.adjacent_ref_carrier_fluctuations[mosa] - self.local_ref_carrier_fluctuations[mosa]
- )
+ self.tps_ref_carrier_fluctuations = \
+ self.adjacent_ref_carrier_fluctuations - self.local_ref_carrier_fluctuations
logger.debug("Computing reference upper sideband beatnote offsets on TPS")
- self.tps_ref_usb_offsets = ForEachMOSA(lambda mosa:
- self.adjacent_ref_usb_offsets[mosa] - self.local_ref_usb_offsets[mosa]
- )
+ self.tps_ref_usb_offsets = \
+ self.adjacent_ref_usb_offsets - self.local_ref_usb_offsets
logger.debug("Computing reference upper sideband beatnote fluctuations on TPS")
- self.tps_ref_usb_fluctuations = ForEachMOSA(lambda mosa:
- self.adjacent_ref_usb_fluctuations[mosa] - self.local_ref_usb_fluctuations[mosa]
- )
+ self.tps_ref_usb_fluctuations = \
+ self.adjacent_ref_usb_fluctuations - self.local_ref_usb_fluctuations
## Sampling beatnotes and measured pseudo-ranges to THE grid
logger.info("Inverting timer deviations")
- self.inverse_timer_deviations = ForEachSC(lambda sc:
- self.invert_timer_deviations(self.local_timer_deviations[sc], sc)
- )
+ self.inverse_timer_deviations = self.local_timer_deviations \
+ .transformed(lambda sc, x: self.invert_timer_deviations(x, sc))
- self.timestamped = lambda mosa, x: self.interpolate(x,
- -self.inverse_timer_deviations[ForEachMOSA.sc(mosa)] * self.physics_fs)
+ self.timestamped = \
+ lambda mosa, x: self.interpolate(x, -self.inverse_timer_deviations.for_each_mosa()[mosa])
logger.info("Sampling inter-spacecraft beatnotes to THE grid")
logger.debug("Sampling inter-spacecraft carrier beatnote fluctuations to THE grid")
- self.the_isc_carrier_offsets = ForEachMOSA(lambda mosa:
- self.tps_isc_carrier_offsets[mosa] / (1 + self.clock_noise_offsets[mosa[0]])
+ self.the_isc_carrier_offsets = (
+ self.tps_isc_carrier_offsets / (1 + self.clock_noise_offsets)
).transformed(self.timestamped)
logger.debug("Sampling inter-spacecraft carrier beatnote fluctuations to THE grid")
- self.the_isc_carrier_fluctuations = ForEachMOSA(lambda mosa:
- self.tps_isc_carrier_fluctuations[mosa] / (1 + self.clock_noise_offsets[mosa[0]])
- - self.tps_isc_carrier_offsets[mosa] * self.clock_noise_fluctuations[mosa[0]]
- / (1 + self.clock_noise_offsets[mosa[0]])**2
+ self.the_isc_carrier_fluctuations = (
+ self.tps_isc_carrier_fluctuations / (1 + self.clock_noise_offsets)
+ - self.tps_isc_carrier_offsets * self.clock_noise_fluctuations
+ / (1 + self.clock_noise_offsets)**2
).transformed(self.timestamped)
logging.debug("Sampling inter-spacecraft upper sideband beatnote offsets to THE grid")
- self.the_isc_usb_offsets = ForEachMOSA(lambda mosa:
- self.tps_isc_usb_offsets[mosa] / (1 + self.clock_noise_offsets[mosa[0]])
+ self.the_isc_usb_offsets = (
+ self.tps_isc_usb_offsets / (1 + self.clock_noise_offsets)
).transformed(self.timestamped)
logger.debug("Sampling inter-spacecraft upper sideband beatnote fluctuations to THE grid")
- self.the_isc_usb_fluctuations = ForEachMOSA(lambda mosa:
- self.tps_isc_usb_fluctuations[mosa] / (1 + self.clock_noise_offsets[mosa[0]])
- - self.tps_isc_usb_offsets[mosa] * self.clock_noise_fluctuations[mosa[0]]
- / (1 + self.clock_noise_offsets[mosa[0]])**2
+ self.the_isc_usb_fluctuations = (
+ self.tps_isc_usb_fluctuations / (1 + self.clock_noise_offsets)
+ - self.tps_isc_usb_offsets * self.clock_noise_fluctuations
+ / (1 + self.clock_noise_offsets)**2
).transformed(self.timestamped)
logger.info("Sampling measured pseudo-ranges to THE grid")
- self.the_mprs = ForEachMOSA(lambda mosa:
- self.tps_mprs[mosa]
+ self.the_mprs = (
+ self.tps_mprs
).transformed(self.timestamped)
logger.info("Sampling test-mass beatnotes to THE grid")
logger.debug("Sampling test-mass carrier beatnote offsets to THE grid")
- self.the_tm_carrier_offsets = ForEachMOSA(lambda mosa:
- self.tps_tm_carrier_offsets[mosa] / (1 + self.clock_noise_offsets[mosa[0]])
+ self.the_tm_carrier_offsets = (
+ self.tps_tm_carrier_offsets / (1 + self.clock_noise_offsets)
).transformed(self.timestamped)
logger.debug("Sampling test-mass carrier beatnote fluctuations to THE grid")
- self.the_tm_carrier_fluctuations = ForEachMOSA(lambda mosa:
- self.tps_tm_carrier_fluctuations[mosa] / (1 + self.clock_noise_offsets[mosa[0]])
- - self.tps_tm_carrier_offsets[mosa] * self.clock_noise_fluctuations[mosa[0]]
- / (1 + self.clock_noise_offsets[mosa[0]])**2
+ self.the_tm_carrier_fluctuations = (
+ self.tps_tm_carrier_fluctuations / (1 + self.clock_noise_offsets)
+ - self.tps_tm_carrier_offsets * self.clock_noise_fluctuations
+ / (1 + self.clock_noise_offsets)**2
).transformed(self.timestamped)
logger.debug("Sampling test-mass upper sideband beatnote offsets to THE grid")
- self.the_tm_usb_offsets = ForEachMOSA(lambda mosa:
- self.tps_tm_usb_offsets[mosa] / (1 + self.clock_noise_offsets[mosa[0]])
+ self.the_tm_usb_offsets = (
+ self.tps_tm_usb_offsets / (1 + self.clock_noise_offsets)
).transformed(self.timestamped)
logging.debug("Sampling test-mass upper sideband beatnote fluctuations to THE grid")
- self.the_tm_usb_fluctuations = ForEachMOSA(lambda mosa:
- self.tps_tm_usb_fluctuations[mosa] / (1 + self.clock_noise_offsets[mosa[0]])
- - self.tps_tm_usb_offsets[mosa] * self.clock_noise_fluctuations[mosa[0]]
- / (1 + self.clock_noise_offsets[mosa[0]])**2
+ self.the_tm_usb_fluctuations = (
+ self.tps_tm_usb_fluctuations / (1 + self.clock_noise_offsets)
+ - self.tps_tm_usb_offsets * self.clock_noise_fluctuations
+ / (1 + self.clock_noise_offsets)**2
).transformed(self.timestamped)
logger.info("Sampling reference beatnotes to THE grid")
logger.debug("Sampling reference carrier beatnote offsets to THE grid")
- self.the_ref_carrier_offsets = ForEachMOSA(lambda mosa:
- self.tps_ref_carrier_offsets[mosa] / (1 + self.clock_noise_offsets[mosa[0]])
+ self.the_ref_carrier_offsets = (
+ self.tps_ref_carrier_offsets / (1 + self.clock_noise_offsets)
).transformed(self.timestamped)
logger.debug("Sampling reference carrier beatnote fluctuations to THE grid")
- self.the_ref_carrier_fluctuations = ForEachMOSA(lambda mosa:
- self.tps_ref_carrier_fluctuations[mosa] / (1 + self.clock_noise_offsets[mosa[0]])
- - self.tps_ref_carrier_offsets[mosa] * self.clock_noise_fluctuations[mosa[0]]
- / (1 + self.clock_noise_offsets[mosa[0]])**2
+ self.the_ref_carrier_fluctuations = (
+ self.tps_ref_carrier_fluctuations / (1 + self.clock_noise_offsets)
+ - self.tps_ref_carrier_offsets * self.clock_noise_fluctuations
+ / (1 + self.clock_noise_offsets)**2
).transformed(self.timestamped)
logger.debug("Sampling reference upper sideband beatnote offsets to THE grid")
- self.the_ref_usb_offsets = ForEachMOSA(lambda mosa:
- self.tps_ref_usb_offsets[mosa] / (1 + self.clock_noise_offsets[mosa[0]])
+ self.the_ref_usb_offsets = (
+ self.tps_ref_usb_offsets / (1 + self.clock_noise_offsets)
).transformed(self.timestamped)
logger.debug("Sampling reference upper sideband beatnote fluctuations to THE grid")
- self.the_ref_usb_fluctuations = ForEachMOSA(lambda mosa:
- self.tps_ref_usb_fluctuations[mosa] / (1 + self.clock_noise_offsets[mosa[0]])
- - self.tps_ref_usb_offsets[mosa] * self.clock_noise_fluctuations[mosa[0]]
- / (1 + self.clock_noise_offsets[mosa[0]])**2
+ self.the_ref_usb_fluctuations = (
+ self.tps_ref_usb_fluctuations / (1 + self.clock_noise_offsets)
+ - self.tps_ref_usb_offsets * self.clock_noise_fluctuations
+ / (1 + self.clock_noise_offsets)**2
).transformed(self.timestamped)
## Total frequencies
@@ -793,34 +753,28 @@ class Instrument:
logger.info("Computing total beatnote frequencies")
logger.debug("Computing total inter-spacecraft carrier beatnotes")
- self.the_isc_carriers = ForEachMOSA(lambda mosa:
- self.the_isc_carrier_offsets[mosa] + self.the_isc_carrier_fluctuations[mosa]
- )
+ self.the_isc_carriers = \
+ self.the_isc_carrier_offsets + self.the_isc_carrier_fluctuations
logger.debug("Computing total inter-spacecraft upper sideband beatnotes")
- self.the_isc_usbs = ForEachMOSA(lambda mosa:
- self.the_isc_usb_offsets[mosa] + self.the_isc_usb_fluctuations[mosa]
- )
+ self.the_isc_usbs = \
+ self.the_isc_usb_offsets + self.the_isc_usb_fluctuations
logger.debug("Computing total test-mass carrier beatnotes")
- self.the_tm_carriers = ForEachMOSA(lambda mosa:
- self.the_tm_carrier_offsets[mosa] + self.the_tm_carrier_fluctuations[mosa]
- )
+ self.the_tm_carriers = \
+ self.the_tm_carrier_offsets + self.the_tm_carrier_fluctuations
logger.debug("Computing total test-mass upper sideband beatnotes")
- self.the_tm_usbs = ForEachMOSA(lambda mosa:
- self.the_tm_usb_offsets[mosa] + self.the_tm_usb_fluctuations[mosa]
- )
+ self.the_tm_usbs = \
+ self.the_tm_usb_offsets + self.the_tm_usb_fluctuations
logger.debug("Computing total reference carrier beatnotes")
- self.the_ref_carriers = ForEachMOSA(lambda mosa:
- self.the_ref_carrier_offsets[mosa] + self.the_ref_carrier_fluctuations[mosa]
- )
+ self.the_ref_carriers = \
+ self.the_ref_carrier_offsets + self.the_ref_carrier_fluctuations
logger.debug("Computing total reference upper sideband beatnotes")
- self.the_ref_usbs = ForEachMOSA(lambda mosa:
- self.the_ref_usb_offsets[mosa] + self.the_ref_usb_fluctuations[mosa]
- )
+ self.the_ref_usbs = \
+ self.the_ref_usb_offsets + self.the_ref_usb_fluctuations
## Antialiasing filtering
@@ -912,15 +866,14 @@ class Instrument:
if self.clock_freqlindrifts == self.clock_freqquaddrifts == 0:
# Optimize to use a scalar if we only have a constant frequency offset
logger.debug("Generating clock noise offsets as constant frequency offsets")
- self.clock_noise_offsets = ForEachSC(lambda sc:
- self.clock_freqoffsets[sc]
- )
+ self.clock_noise_offsets = self.clock_freqoffsets
else:
logger.debug("Generating clock noise offsets")
t = self.physics_t
- self.clock_noise_offsets = ForEachSC(lambda sc:
- self.clock_freqoffsets[sc] + self.clock_freqlindrifts[sc] * t + self.clock_freqquaddrifts[sc] * t**2
- )
+ self.clock_noise_offsets = \
+ self.clock_freqoffsets \
+ + self.clock_freqlindrifts * t \
+ + self.clock_freqquaddrifts * t**2
logger.debug("Generating clock noise fluctuations")
self.clock_noise_fluctuations = ForEachSC(lambda sc:
@@ -939,7 +892,8 @@ class Instrument:
logger.info("Generating backlink noise")
self.backlink_noises = ForEachMOSA(lambda mosa:
- noises.backlink(self.physics_fs, self.physics_size, self.backlink_asds[mosa], self.backlink_fknees[mosa])
+ noises.backlink(self.physics_fs, self.physics_size,
+ self.backlink_asds[mosa], self.backlink_fknees[mosa])
)
## Test-mass acceleration noise
@@ -954,7 +908,8 @@ class Instrument:
logger.info("Generating ranging noise")
self.ranging_noises = ForEachMOSA(lambda mosa:
- self.ranging_biases[mosa] + noises.ranging(self.physics_fs, self.physics_size, self.ranging_asds[mosa])
+ self.ranging_biases[mosa] + noises.ranging(self.physics_fs,
+ self.physics_size, self.ranging_asds[mosa])
)
def invert_timer_deviations(self, timer_deviations, sc):
diff --git a/tests/test_containers.py b/tests/test_containers.py
index 094ced5a4a5ff424c9e9f15959d656263b332975..32abe7113158def261505539379e4f00e23c29bd 100644
--- a/tests/test_containers.py
+++ b/tests/test_containers.py
@@ -120,6 +120,141 @@ def test_equality():
assert object_1 == object_2
+def test_addition():
+ """Check that we can add two ForEachObject subclasses of the same type, or a scalar."""
+
+ object_1 = ForEachAB({'A': 1, 'B': 2})
+ object_2 = ForEachAB({'A': 3, 'B': -4})
+ object_3 = ForEachMOSA(0)
+
+ result = object_1 + object_2
+ assert result['A'] == 4
+ assert result['B'] == -2
+
+ result = object_1 + 10
+ assert result['A'] == 11
+ assert result['B'] == 12
+
+ result = -5 + object_2
+ assert result['A'] == -2
+ assert result['B'] == -9
+
+ with raises(TypeError):
+ result = object_1 + object_3
+ with raises(TypeError):
+ result = object_2 + object_3
+
+
+def test_subtraction():
+ """Check that we can subtract two ForEachObject subclasses of the same type, or a scalar."""
+
+ object_1 = ForEachAB({'A': 1, 'B': 2})
+ object_2 = ForEachAB({'A': 3, 'B': -4})
+ object_3 = ForEachMOSA(0)
+
+ result = object_1 - object_2
+ assert result['A'] == -2
+ assert result['B'] == 6
+
+ result = object_1 - 10
+ assert result['A'] == -9
+ assert result['B'] == -8
+
+ result = 10 - object_2
+ assert result['A'] == 7
+ assert result['B'] == 14
+
+ with raises(TypeError):
+ result = object_1 - object_3
+ with raises(TypeError):
+ result = object_2 - object_3
+
+
+def test_multiplication():
+ """Check that we can multiply two ForEachObject subclasses of the same type, or a scalar."""
+
+ object_1 = ForEachAB({'A': 1, 'B': 2})
+ object_2 = ForEachAB({'A': 3, 'B': -4})
+ object_3 = ForEachMOSA(0)
+
+ result = object_1 * object_2
+ assert result['A'] == 3
+ assert result['B'] == -8
+
+ result = object_1 * 2
+ assert result['A'] == 2
+ assert result['B'] == 4
+
+ result = 10 * object_2
+ assert result['A'] == 30
+ assert result['B'] == -40
+
+ with raises(TypeError):
+ result = object_1 * object_3
+ with raises(TypeError):
+ result = object_2 * object_3
+
+
+def test_floor_division():
+ """Check that we can apply floor division on two ForEachObject subclasses, or a scalar."""
+
+ object_1 = ForEachAB({'A': 1, 'B': 2})
+ object_2 = ForEachAB({'A': 3, 'B': -4})
+ object_3 = ForEachMOSA(0)
+
+ result = object_2 // object_1
+ assert result['A'] == 3
+ assert result['B'] == -2
+
+ result = object_2 // 2
+ assert result['A'] == 1
+ assert result['B'] == -2
+
+ with raises(TypeError):
+ result = object_1 // object_3
+ with raises(TypeError):
+ result = 20 // object_1
+
+
+def test_real_division():
+ """Check that we can divide two ForEachObject subclasses of the same type, or a scalar."""
+
+ object_1 = ForEachAB({'A': 1, 'B': 2})
+ object_2 = ForEachAB({'A': 3, 'B': -4})
+ object_3 = ForEachMOSA(0)
+
+ result = object_1 / object_2
+ assert result['A'] == 1/3
+ assert result['B'] == -1/2
+
+ result = object_1 / 5
+ assert result['A'] == 1/5
+ assert result['B'] == 2/5
+
+ result = 10 / object_2
+ assert result['A'] == 10/3
+ assert result['B'] == -10/4
+
+ with raises(TypeError):
+ result = object_1 / object_3
+ with raises(TypeError):
+ result = object_2 / object_3
+
+
+def test_power():
+ """Check that we take the power of a ForEachObject instance."""
+
+ object_1 = ForEachAB({'A': 1, 'B': -2})
+
+ result = object_1**1
+ assert result['A'] == 1
+ assert result['B'] == -2
+
+ result = object_1**2
+ assert result['A'] == 1
+ assert result['B'] == 4
+
+
def test_transformed():
"""Check that we transformation is correctly applied to ForEachObject instances."""
@@ -132,6 +267,25 @@ def test_transformed():
assert my_object['A'] == 'a'
assert my_object['B'] == 'b'
+
+def test_abs():
+ """Check that we can take the absolute value of ForEachObject instances."""
+
+ my_object = ForEachAB({'A': -1, 'B': 2})
+ my_object = abs(my_object)
+ assert my_object['A'] == 1
+ assert my_object['B'] == 2
+
+
+def test_neg():
+ """Check that we can take the negative value of ForEachObject instances."""
+
+ my_object = ForEachAB({'A': -1, 'B': 2})
+ my_object = -my_object
+ assert my_object['A'] == 1
+ assert my_object['B'] == -2
+
+
def test_write():
"""Check that we can write a ForEachObject instance."""
@@ -153,13 +307,13 @@ def test_sc_indices():
assert ForEachSC.indices() == ['1', '2', '3']
- assert ForEachSC.distant_left('1') == '2'
- assert ForEachSC.distant_left('2') == '3'
- assert ForEachSC.distant_left('3') == '1'
+ assert ForEachSC.distant_left_sc('1') == '2'
+ assert ForEachSC.distant_left_sc('2') == '3'
+ assert ForEachSC.distant_left_sc('3') == '1'
- assert ForEachSC.distant_right('1') == '3'
- assert ForEachSC.distant_right('2') == '1'
- assert ForEachSC.distant_right('3') == '2'
+ assert ForEachSC.distant_right_sc('1') == '3'
+ assert ForEachSC.distant_right_sc('2') == '1'
+ assert ForEachSC.distant_right_sc('3') == '2'
assert ForEachSC.left_mosa('1') == '12'
assert ForEachSC.left_mosa('2') == '23'
@@ -182,32 +336,95 @@ def test_mosa_indices():
assert ForEachMOSA.sc('32') == '3'
assert ForEachMOSA.sc('21') == '2'
- assert ForEachMOSA.distant('12') == '21'
- assert ForEachMOSA.distant('23') == '32'
- assert ForEachMOSA.distant('31') == '13'
- assert ForEachMOSA.distant('13') == '31'
- assert ForEachMOSA.distant('32') == '23'
- assert ForEachMOSA.distant('21') == '12'
+ assert ForEachMOSA.distant_mosa('12') == '21'
+ assert ForEachMOSA.distant_mosa('23') == '32'
+ assert ForEachMOSA.distant_mosa('31') == '13'
+ assert ForEachMOSA.distant_mosa('13') == '31'
+ assert ForEachMOSA.distant_mosa('32') == '23'
+ assert ForEachMOSA.distant_mosa('21') == '12'
- assert ForEachMOSA.adjacent('12') == '13'
- assert ForEachMOSA.adjacent('23') == '21'
- assert ForEachMOSA.adjacent('31') == '32'
- assert ForEachMOSA.adjacent('13') == '12'
- assert ForEachMOSA.adjacent('32') == '31'
- assert ForEachMOSA.adjacent('21') == '23'
+ assert ForEachMOSA.adjacent_mosa('12') == '13'
+ assert ForEachMOSA.adjacent_mosa('23') == '21'
+ assert ForEachMOSA.adjacent_mosa('31') == '32'
+ assert ForEachMOSA.adjacent_mosa('13') == '12'
+ assert ForEachMOSA.adjacent_mosa('32') == '31'
+ assert ForEachMOSA.adjacent_mosa('21') == '23'
-def test_init_mosa_with_sc():
- """Test one can initialize ForEachMOSA from ForEachSC."""
+
+def test_foreachsc_to_foreachmosa():
+ """Test that one can turn a ForEachSC instance to a ForEachMOSA instance."""
my_sc = ForEachSC(lambda sc: 10 * int(sc))
assert my_sc['1'] == 10
assert my_sc['2'] == 20
assert my_sc['3'] == 30
- my_mosa = ForEachMOSA(my_sc)
+ my_mosa = my_sc.for_each_mosa()
assert my_mosa['12'] == 10
assert my_mosa['13'] == 10
assert my_mosa['21'] == 20
assert my_mosa['23'] == 20
assert my_mosa['31'] == 30
assert my_mosa['32'] == 30
+
+
+def test_auto_foreachsc_to_foreachmosa():
+ """Test that ForEachSC turn automatically into ForEachMOSA during operations."""
+
+ my_sc = ForEachSC(int)
+ my_mosa = ForEachMOSA(int)
+
+ my_add_1 = my_sc + my_mosa
+ my_add_2 = my_mosa + my_sc
+ assert my_add_1 == my_sc.for_each_mosa() + my_mosa
+ assert my_add_1 == my_add_2
+
+ my_sub_1 = my_sc - my_mosa
+ my_sub_2 = my_mosa - my_sc
+ assert my_sub_1 == my_sc.for_each_mosa() - my_mosa
+ assert my_sub_1 == -my_sub_2
+
+ my_mult_1 = my_sc * my_mosa
+ my_mult_2 = my_mosa * my_sc
+ assert my_mult_1 == my_sc.for_each_mosa() * my_mosa
+ assert my_mult_1 == my_mult_2
+
+
+def test_foreachmosa_distant():
+ """Test that one can generate a ForEachMOSA instant for distant MOSAs."""
+
+ my_mosa = ForEachMOSA(int)
+ assert my_mosa['12'] == 12
+ assert my_mosa['13'] == 13
+ assert my_mosa['21'] == 21
+ assert my_mosa['23'] == 23
+ assert my_mosa['31'] == 31
+ assert my_mosa['32'] == 32
+
+ distant_mosa = my_mosa.distant()
+ assert distant_mosa['12'] == 21
+ assert distant_mosa['13'] == 31
+ assert distant_mosa['21'] == 12
+ assert distant_mosa['23'] == 32
+ assert distant_mosa['31'] == 13
+ assert distant_mosa['32'] == 23
+
+
+def test_foreachmosa_adjacent():
+ """Test that one can generate a ForEachMOSA instant for adjacent MOSAs."""
+
+ my_mosa = ForEachMOSA(int)
+ assert my_mosa['12'] == 12
+ assert my_mosa['13'] == 13
+ assert my_mosa['21'] == 21
+ assert my_mosa['23'] == 23
+ assert my_mosa['31'] == 31
+ assert my_mosa['32'] == 32
+
+ adjacent_mosa = my_mosa.adjacent()
+ assert adjacent_mosa['12'] == 13
+ assert adjacent_mosa['13'] == 12
+ assert adjacent_mosa['21'] == 23
+ assert adjacent_mosa['23'] == 21
+ assert adjacent_mosa['31'] == 32
+ assert adjacent_mosa['32'] == 31