Integration/1.3.64

.github/workflows/pypi.yml

@@ -42,7 +42,7 @@ jobs:
 
   publish-to-pypi:
     name: Publish to PyPI
-    if: startsWith(github.ref, 'refs/tags/v')
+    if: startsWith(github.ref, 'refs/tags/v') || contains(github.ref, 'refs/heads/integration/')
     needs:
     - build
     runs-on: ubuntu-latest

sarpy/__about__.py

@@ -28,7 +28,7 @@
 
 from sarpy.__details__ import __classification__, _post_identifier
 
-__version__ = "1.3.63"
+__version__ = "1.3.64"
 
 __author__ = "National Geospatial-Intelligence Agency"
 __url__ = "https://github.com/ngageoint/sarpy"

sarpy/geometry/point_projection.py

@@ -176,9 +176,6 @@ def pfa_projection():
         polar_ang_poly_der = polar_ang_poly.derivative(der_order=1, return_poly=True)
         spatial_freq_sf_poly_der = spatial_freq_sf_poly.derivative(der_order=1, return_poly=True)
 
-        polar_ang_poly_der = polar_ang_poly.derivative(der_order=1, return_poly=True)
-        spatial_freq_sf_poly_der = spatial_freq_sf_poly.derivative(der_order=1, return_poly=True)
-
         # noinspection PyUnusedLocal, PyIncorrectDocstring
         def method_projection(instance, row_transform, col_transform, time_coa, arp_coa, varp_coa):
             """

sarpy/io/general/format_function.py

@@ -801,27 +801,29 @@ def _forward_functional_step(
                         data.shape[self.band_dimension + 1:]
 
         out = numpy.empty(out_shape, dtype='complex64')
+
+        # Build slices for even/odd indices along band_dimension
+        # Using slices is faster than numpy.take() with range()
+        even_slice = [slice(None)] * data.ndim
+        odd_slice = [slice(None)] * data.ndim
+        even_slice[self.band_dimension] = slice(0, band_dim_size, 2)
+        odd_slice[self.band_dimension] = slice(1, band_dim_size, 2)
+        even_data = data[tuple(even_slice)].reshape(out.shape)
+        odd_data = data[tuple(odd_slice)].reshape(out.shape)
+
         if self.order == 'IQ':
-            out.real = numpy.reshape(
-                data.take(indices=range(0, band_dim_size, 2), axis=self.band_dimension), out.shape)
-            out.imag = numpy.reshape(
-                data.take(indices=range(1, band_dim_size, 2), axis=self.band_dimension), out.shape)
+            out.real = even_data
+            out.imag = odd_data
         elif self.order == 'QI':
-            out.imag = numpy.reshape(
-                data.take(indices=range(0, band_dim_size, 2), axis=self.band_dimension), out.shape)
-            out.real = numpy.reshape(
-                data.take(indices=range(1, band_dim_size, 2), axis=self.band_dimension), out.shape)
+            out.imag = even_data
+            out.real = odd_data
         elif self.order in ['MP', 'PM']:
             if self.order == 'MP':
-                mag = numpy.reshape(
-                    data.take(indices=range(0, band_dim_size, 2), axis=self.band_dimension), out.shape)
-                theta = numpy.reshape(
-                    data.take(indices=range(1, band_dim_size, 2), axis=self.band_dimension), out.shape)
+                mag = even_data
+                theta = odd_data
             else:
-                mag = numpy.reshape(
-                    data.take(indices=range(1, band_dim_size, 2), axis=self.band_dimension), out.shape)
-                theta = numpy.reshape(
-                    data.take(indices=range(0, band_dim_size, 2), axis=self.band_dimension), out.shape)
+                mag = odd_data
+                theta = even_data
             self._forward_magnitude_theta(data, out, mag, theta, subscript)
         else:
             raise ValueError('Unhandled order value {}'.format(self.order))

tests/io/general/test_format_function.py

@@ -187,3 +187,105 @@ def test_forward(self):
                 func = SingleLUTFormatFunction(lut, base_data.shape, out_shape)
                 out_data = func(base_data, (slice(0, 51, 1), slice(0, 49, 1)))
                 self.assertTrue(numpy.array_equal(out_data, lut[base_data]), msg='LUT forward')
+
+
+class TestComplexFormatFunctionForwardFunctionalStep(unittest.TestCase):
+    def test_forward_functional_step_IQ(self):
+        # shape: (2, 3, 4), band_dimension=2, so 4 bands, should be 2 complex values per last axis
+        data = numpy.zeros((2, 3, 4), dtype='float32')
+        data[:, :, 0] = 1.0  # real part 1st complex
+        data[:, :, 1] = 2.0  # imag part 1st complex
+        data[:, :, 2] = 3.0  # real part 2nd complex
+        data[:, :, 3] = 4.0  # imag part 2nd complex
+
+        func = ComplexFormatFunction(
+            'float32', 'IQ', raw_shape=(2, 3, 4), formatted_shape=(2, 3, 2), band_dimension=2)
+        subscript = (slice(0, 2), slice(0, 3), slice(0, 4))
+        out = func._forward_functional_step(data, subscript)
+        self.assertEqual(out.shape, (2, 3, 2))
+        self.assertTrue(numpy.all(out.real == numpy.array([[[1, 3], [1, 3], [1, 3]],
+                                                           [[1, 3], [1, 3], [1, 3]]])))
+        self.assertTrue(numpy.all(out.imag == numpy.array([[[2, 4], [2, 4], [2, 4]],
+                                                           [[2, 4], [2, 4], [2, 4]]])))
+
+    def test_forward_functional_step_QI(self):
+        data = numpy.zeros((2, 3, 4), dtype='float32')
+        data[:, :, 0] = 10.0  # imag part 1st complex
+        data[:, :, 1] = 20.0  # real part 1st complex
+        data[:, :, 2] = 30.0  # imag part 2nd complex
+        data[:, :, 3] = 40.0  # real part 2nd complex
+
+        func = ComplexFormatFunction(
+            'float32', 'QI', raw_shape=(2, 3, 4), formatted_shape=(2, 3, 2), band_dimension=2)
+        subscript = (slice(0, 2), slice(0, 3), slice(0, 4))
+        out = func._forward_functional_step(data, subscript)
+        self.assertEqual(out.shape, (2, 3, 2))
+        self.assertTrue(numpy.all(out.real == numpy.array([[[20, 40], [20, 40], [20, 40]],
+                                                           [[20, 40], [20, 40], [20, 40]]])))
+        self.assertTrue(numpy.all(out.imag == numpy.array([[[10, 30], [10, 30], [10, 30]],
+                                                           [[10, 30], [10, 30], [10, 30]]])))
+
+    def test_forward_functional_step_MP(self):
+        # MP: magnitude, phase
+        mag = numpy.ones((2, 3, 2), dtype='float32') * 5
+        theta = numpy.ones((2, 3, 2), dtype='float32') * numpy.pi / 2
+        data = numpy.empty((2, 3, 4), dtype='float32')
+        data[:, :, 0] = mag[:, :, 0]
+        data[:, :, 1] = theta[:, :, 0]
+        data[:, :, 2] = mag[:, :, 1]
+        data[:, :, 3] = theta[:, :, 1]
+
+        func = ComplexFormatFunction(
+            'float32', 'MP', raw_shape=(2, 3, 4), formatted_shape=(2, 3, 2), band_dimension=2)
+        subscript = (slice(0, 2), slice(0, 3), slice(0, 4))
+        out = func._forward_functional_step(data, subscript)
+        self.assertEqual(out.shape, (2, 3, 2))
+        # For theta=pi/2, cos=0, sin=1, so real=0, imag=5
+        self.assertTrue(numpy.allclose(out.real, 0, atol=1e-6))
+        self.assertTrue(numpy.allclose(out.imag, 5, atol=1e-6))
+
+    def test_forward_functional_step_PM(self):
+        # PM: phase, magnitude
+        mag = numpy.ones((2, 3, 2), dtype='float32') * 7
+        theta = numpy.ones((2, 3, 2), dtype='float32') * numpy.pi
+        data = numpy.empty((2, 3, 4), dtype='float32')
+        data[:, :, 0] = theta[:, :, 0]
+        data[:, :, 1] = mag[:, :, 0]
+        data[:, :, 2] = theta[:, :, 1]
+        data[:, :, 3] = mag[:, :, 1]
+
+        func = ComplexFormatFunction(
+            'float32', 'PM', raw_shape=(2, 3, 4), formatted_shape=(2, 3, 2), band_dimension=2)
+        subscript = (slice(0, 2), slice(0, 3), slice(0, 4))
+        out = func._forward_functional_step(data, subscript)
+        self.assertEqual(out.shape, (2, 3, 2))
+        # For theta=pi, cos=-1, sin=0, so real=-7, imag=0
+        self.assertTrue(numpy.allclose(out.real, -7, atol=1e-6))
+        self.assertTrue(numpy.allclose(out.imag, 0, atol=1e-6))
+
+    def test_forward_functional_step_MP_uint8(self):
+        # Test quantized phase for uint8
+        mag = numpy.ones((2, 3, 2), dtype='uint8') * 10
+        theta = numpy.ones((2, 3, 2), dtype='uint8') * 64  # 64/256*2pi = pi/2
+        data = numpy.empty((2, 3, 4), dtype='uint8')
+        data[:, :, 0] = mag[:, :, 0]
+        data[:, :, 1] = theta[:, :, 0]
+        data[:, :, 2] = mag[:, :, 1]
+        data[:, :, 3] = theta[:, :, 1]
+
+        func = ComplexFormatFunction(
+            'uint8', 'MP', raw_shape=(2, 3, 4), formatted_shape=(2, 3, 2), band_dimension=2)
+        subscript = (slice(0, 2), slice(0, 3), slice(0, 4))
+        out = func._forward_functional_step(data, subscript)
+        self.assertEqual(out.shape, (2, 3, 2))
+        # For theta=64, bit_depth=8, theta=64/256*2pi=pi/2, so real=0, imag=10
+        self.assertTrue(numpy.allclose(out.real, 0, atol=1e-6))
+        self.assertTrue(numpy.allclose(out.imag, 10, atol=1e-6))
+
+    def test_forward_functional_step_bad_shape(self):
+        func = ComplexFormatFunction(
+            'float32', 'IQ', raw_shape=(2, 3, 4), formatted_shape=(2, 3, 2), band_dimension=2)
+        data = numpy.zeros((2, 3, 5), dtype='float32')  # wrong band size
+        subscript = (slice(0, 2), slice(0, 3), slice(0, 5))
+        with self.assertRaises(ValueError):
+            func._forward_functional_step(data, subscript)