.gitlab-ci.yml

@@ -11,8 +11,11 @@ pylint:
     - pip install -r requirements.txt
     - pylint **/*.py
 
-unittest:
+pytest:
   script:
     - pip install -r requirements.txt
     - pip install .
-    - python -m unittest -v
+    - python -m pytest -v --junitxml=report.xml
+  artifacts:
+    reports:
+      junit: report.xml

lisainstrument/containers.py

@@ -38,14 +38,6 @@ class ForEachObject(abc.ABC):
         """Return list of object indices."""
         raise NotImplementedError
 
-    def transform(self, transformation):
-        """Transform dictionary on-the-spot.
-
-        Args:
-            transformation: function (mosa, value -> new_value)
-        """
-        self.dict = {mosa: transformation(mosa, self[mosa]) for mosa in self.indices()}
-
     def transformed(self, transformation):
         """Return a new dictionary from transformed objects.
 
@@ -149,31 +141,31 @@ class ForEachSC(ForEachObject):
 
     @staticmethod
     def distant_left(sc):
-        """Return index of distant left spacecraft."""
+        """Return index of distant rleftspacecraft."""
         if sc not in ForEachSC.indices():
             raise ValueError(f"invalid spacecraft index '{sc}'")
-        return f'{(int(sc) - 2) % 3 + 1}'
+        return f'{int(sc) % 3 + 1}'
 
     @staticmethod
     def distant_right(sc):
         """Return index of distant right spacecraft."""
         if sc not in ForEachSC.indices():
             raise ValueError(f"invalid spacecraft index '{sc}'")
-        return f'{int(sc) % 3 + 1}'
+        return f'{(int(sc) - 2) % 3 + 1}'
 
     @staticmethod
     def left_mosa(sc):
         """Return index of left 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_left(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_left(sc)}'
+        return f'{sc}{ForEachSC.distant_right(sc)}'
 
 
 class ForEachMOSA(ForEachObject):

lisainstrument/instrument.py

@@ -164,38 +164,29 @@ class Instrument:
         self.init_aafilter(aafilter)
 
     def init_interpolation(self, interpolation):
-        """Initialize interpolation."""
+        """Initialize or design the interpolation function.
+
+        We support no interpolation, a custom interpolation function, or Lagrange interpolation.
+
+        Args:
+            parameters: see `interpolation` docstring in `__init__()`
+        """
         if interpolation is None:
             logging.info("Disabling interpolation")
+            self.interpolation_order = None
             self.interpolate = lambda x, _: x
         elif callable(interpolation):
             logging.info("Using user-provided interpolation function")
+            self.interpolation_order = None
             self.interpolate = lambda x, shift: x if numpy.isscalar(x) else interpolation(x, shift)
         else:
-            logging.info("Designing interpolation function")
-            self.interpolate = lambda x, shift: x if numpy.isscalar(x) \
-                else Instrument.design_interpolation(interpolation)(x, shift)
-
-    @staticmethod
-    def design_interpolation(parameters):
-        """Design the interpolation function.
-
-        We currently support Lagrange interpolation.
-
-        Args:
-            parameters: see `interpolation` docstring in `__init__()`
-
-        Returns:
-            A function which interpolates data.
-        """
-        method = parameters[0]
-        if method == 'lagrange':
-            logging.debug("Using Lagrange interpolations")
-            order = parameters[1]
-            logging.debug("Lagrange interpolation order is %s", order)
-            return lambda x, shift: dsp.timeshift(x, shift, order)
-
-        raise ValueError(f"invalid interpolation parameters '{parameters}'")
+            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)
+            raise ValueError(f"invalid interpolation parameters '{interpolation}'")
 
     def init_aafilter(self, aafilter):
         """Initialize antialiasing filter and downsampling."""
@@ -282,7 +273,7 @@ class Instrument:
             self.gw_file = gws
             logging.info("Interpolating gravitational-wave responses from GW file '%s'", self.gw_file)
             gwf = h5py.File(self.gw_file, 'r')
-            self.glitch_tms = ForEachMOSA(lambda mosa: scipy.interpolate.InterpolatedUnivariateSpline(
+            self.gws = ForEachMOSA(lambda mosa: scipy.interpolate.InterpolatedUnivariateSpline(
                 gwf['t'][:], gwf[f'l_{mosa}'][:], k=5, ext='raise')(self.physics_t)
             )
             gwf.close()
@@ -409,7 +400,7 @@ class Instrument:
 
         logging.info("Propagating carrier frequency fluctuations to distant MOSA")
         self.distant_carrier_fluctuations = ForEachMOSA(lambda mosa:
-            self.gws[mosa] + (1 - self.d_pprs[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)
         )
@@ -424,7 +415,7 @@ class Instrument:
 
         logging.info("Propagating upper sideband frequency fluctuations to distant MOSA")
         self.distant_usb_fluctuations = ForEachMOSA(lambda mosa:
-            self.gws[mosa] + (1 - self.d_pprs[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)
         )

tests/test_containers.py

+#! /usr/bin/env python3
+# -*- coding: utf-8 -*-
+# pylint: disable=missing-module-docstring
+
+import os
+import numpy as np
+from pytest import raises
+from h5py import File
+from lisainstrument.containers import ForEachObject, ForEachSC, ForEachMOSA
+
+
+class ForEachAB(ForEachObject):
+    """Concrete implementation of ForEachObject for tests."""
+    @classmethod
+    def indices(cls):
+        return ['A', 'B']
+
+
+def test_init_dictionary():
+    """Test that one can initialize a ForEachObject with a dictionary."""
+
+    my_object = ForEachAB({'A': 1, 'B': 2})
+    assert my_object['A'] == 1
+    assert my_object['B'] == 2
+
+    with raises(KeyError):
+        ForEachAB({'A': 1})
+
+
+def test_init_callable():
+    """Test that one can initialize a ForEachObject with a function."""
+
+    my_object = ForEachAB(lambda index: index.lower())
+    assert my_object['A'] == 'a'
+    assert my_object['B'] == 'b'
+
+
+def test_init_dataset():
+    """Test that one can initialize a ForEachObject with an HDF5 dataset."""
+
+    try:
+        with File('test-temp.h5', 'x') as h5file:
+            dtype = np.dtype({'names': ['A', 'B'], 'formats': [np.float64, np.float64]})
+            h5file.create_dataset('test-dataset', (1, ), dtype=dtype)
+            h5file['test-dataset']['A'] = 1
+            h5file['test-dataset']['B'] = 2
+
+        with File('test-temp.h5', 'r') as h5file:
+            my_object = ForEachAB(h5file['test-dataset'])
+            assert my_object['A'] == [1]
+            assert my_object['B'] == [2]
+    finally:
+        os.remove('test-temp.h5')
+
+    try:
+        with File('test-temp.h5', 'x') as h5file:
+            dtype = np.dtype({'names': ['A'], 'formats': [np.float64]})
+            h5file.create_dataset('test-dataset', (1, ), dtype=dtype)
+            h5file['test-dataset']['A'] = 1
+
+        with File('test-temp.h5', 'r') as h5file:
+            with raises(ValueError):
+                my_object = ForEachAB(h5file['test-dataset'])
+    finally:
+        os.remove('test-temp.h5')
+
+
+def test_init_scalar():
+    """Test that one can initialize a ForEachObject with a scalar."""
+
+    my_object = ForEachAB(42)
+    assert my_object['A'] == 42
+    assert my_object['B'] == 42
+
+
+def test_access_item():
+    """Test that one can get and set items."""
+
+    my_object = ForEachAB(0)
+
+    my_object['A'] = 1
+    assert my_object['A'] == 1
+    assert my_object['B'] == 0
+
+    my_object['B'] = 42
+    assert my_object['A'] == 1
+    assert my_object['B'] == 42
+
+
+def test_len():
+    """Test that one can retreive the length of values."""
+
+    my_object = ForEachAB(0)
+    assert len(my_object) == 1
+
+    my_object = ForEachAB({
+        'A': range(10),
+        'B': range(42),
+    })
+    assert len(my_object) == 42
+
+    my_object = ForEachAB({
+        'A': range(42),
+        'B': 1,
+    })
+    assert len(my_object) == 42
+
+
+def test_equality():
+    """Check that ForEachObject instances are equal when containing identical values."""
+
+    object_1 = ForEachAB({'A': 1, 'B': 2})
+    object_2 = ForEachAB({'A': 1, 'B': 2})
+    assert object_1 == object_2
+
+    object_2['A'] = 3
+    assert object_1 != object_2
+
+    object_1['A'] = 3
+    assert object_1 == object_2
+
+
+def test_transformed():
+    """Check that we transformation is correctly applied to ForEachObject instances."""
+
+    my_object = ForEachAB({'A': 1, 'B': 2})
+    my_object = my_object.transformed(lambda index, x: 2 * x)
+    assert my_object['A'] == 2
+    assert my_object['B'] == 4
+
+    my_object = my_object.transformed(lambda index, x: index.lower())
+    assert my_object['A'] == 'a'
+    assert my_object['B'] == 'b'
+
+def test_write():
+    """Check that we can write a ForEachObject instance."""
+
+    my_object = ForEachAB({'A': 1, 'B': 2})
+
+    try:
+        with File('test-temp.h5', 'x') as h5file:
+            my_object.write(h5file, 'test_dataset')
+
+        with File('test-temp.h5', 'r') as h5file:
+            assert h5file['test_dataset']['A'] == [1]
+            assert h5file['test_dataset']['B'] == [2]
+    finally:
+        os.remove('test-temp.h5')
+
+
+def test_sc_indices():
+    """Test spacecraft 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_right('1') == '3'
+    assert ForEachSC.distant_right('2') == '1'
+    assert ForEachSC.distant_right('3') == '2'
+
+    assert ForEachSC.left_mosa('1') == '12'
+    assert ForEachSC.left_mosa('2') == '23'
+    assert ForEachSC.left_mosa('3') == '31'
+
+    assert ForEachSC.right_mosa('1') == '13'
+    assert ForEachSC.right_mosa('2') == '21'
+    assert ForEachSC.right_mosa('3') == '32'
+
+
+def test_mosa_indices():
+    """Test MOSA indices."""
+
+    assert ForEachMOSA.indices() == ['12', '23', '31', '13', '32', '21']
+
+    assert ForEachMOSA.sc('12') == '1'
+    assert ForEachMOSA.sc('23') == '2'
+    assert ForEachMOSA.sc('31') == '3'
+    assert ForEachMOSA.sc('13') == '1'
+    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.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'