organize the metadata in the radiometric data record

ceos_alos2/dicttoolz.py

@@ -1,5 +1,6 @@
 import copy
 
+from tlz.dicttoolz import assoc as assoc_
 from tlz.dicttoolz import assoc_in, get_in, keyfilter
 from tlz.itertoolz import concat, groupby
 from tlz.itertoolz import identity as passthrough
@@ -26,6 +27,10 @@ def keysplit(predicate, d):
     return itemsplit(wrapper, d)
 
 
+def assoc(key, value, d):
+    return assoc_(d, key, value)
+
+
 def dissoc(keys, d):
     return keyfilter(lambda k: k not in keys, d)
 

ceos_alos2/sar_leader/radiometric_data.py

@@ -1,26 +1,104 @@
 from construct import Struct
+from tlz.dicttoolz import valmap
+from tlz.functoolz import curry, pipe
+from tlz.itertoolz import partition
 
 from ceos_alos2.common import record_preamble
-from ceos_alos2.datatypes import AsciiComplex, AsciiFloat, AsciiInteger, PaddedString
+from ceos_alos2.datatypes import (
+    AsciiComplex,
+    AsciiFloat,
+    AsciiInteger,
+    Metadata,
+    PaddedString,
+)
+from ceos_alos2.dicttoolz import apply_to_items, assoc, dissoc
+from ceos_alos2.transformers import as_group, remove_spares
 
 radiometric_data_record = Struct(
     "preamble" / record_preamble,
     "radiometric_data_records_sequence_number" / AsciiInteger(4),
     "number_of_radiometric_fields" / AsciiInteger(4),
-    "calibration_factor" / AsciiFloat(16),
-    "transmission_distortion_matrix"
-    / Struct(
-        "dt11" / AsciiComplex(32),
-        "dt12" / AsciiComplex(32),
-        "dt21" / AsciiComplex(32),
-        "dt22" / AsciiComplex(32),
+    "calibration_factor"
+    / Metadata(
+        AsciiFloat(16),
+        formula=(
+            "σ⁰=10*log_10<I^2 + Q^2> + CF - 32.0;" " σ⁰(level1.5/level3.1)=10*log_10<DN^2> + CF"
+        ),
+        I="level 1.1 real pixel value",
+        Q="level 1.1 imaginary pixel value",
+        DN="level 1.5/3.1 pixel value",
     ),
-    "reception_distortion_matrix"
-    / Struct(
-        "dr11" / AsciiComplex(32),
-        "dr12" / AsciiComplex(32),
-        "dr21" / AsciiComplex(32),
-        "dr22" / AsciiComplex(32),
+    "distortion_matrix"
+    / Metadata(
+        Struct(
+            "transmission"
+            / Struct(
+                "dt11" / AsciiComplex(32),
+                "dt12" / AsciiComplex(32),
+                "dt21" / AsciiComplex(32),
+                "dt22" / AsciiComplex(32),
+            ),
+            "reception"
+            / Struct(
+                "dr11" / AsciiComplex(32),
+                "dr12" / AsciiComplex(32),
+                "dr21" / AsciiComplex(32),
+                "dr22" / AsciiComplex(32),
+            ),
+        ),
+        formula="Z = A*1/r*exp(-4πr/λ) * RST + N",
+        Z="measurement matrix",
+        A="amplitude",
+        r="slant range",
+        S="true scattering matrix",
+        N="noise component",
+        R="reception distortion matrix",
+        T="transmission distortion matrix",
     ),
     "blanks" / PaddedString(9568),
 )
+
+
+def transform_matrices(mapping):
+    def transform_matrix(mapping):
+        values = mapping.values()
+        matrix = list(map(list, partition(2, values)))
+        dims = ["i", "j"]
+
+        return (dims, matrix, {})
+
+    if isinstance(mapping, tuple):
+        mapping, attrs = mapping
+    else:
+        attrs = {}
+
+    var_i = ("i", ["horizontal", "vertical"], {"long_name": "reception polarization"})
+    var_j = ("j", ["horizontal", "vertical"], {"long_name": "transmission polarization"})
+
+    matrices = pipe(
+        mapping,
+        curry(valmap, transform_matrix),
+        curry(assoc, "i", var_i),
+        curry(assoc, "j", var_j),
+    )
+
+    return matrices, attrs
+
+
+def transform_radiometric_data(mapping):
+    ignored = [
+        "preamble",
+        "radiometric_data_records_sequence_number",
+        "number_of_radiometric_fields",
+    ]
+    transformers = {
+        "distortion_matrix": transform_matrices,
+    }
+
+    return pipe(
+        mapping,
+        curry(dissoc, ignored),
+        curry(remove_spares),
+        curry(apply_to_items, transformers),
+        curry(as_group),
+    )

ceos_alos2/sar_leader/structure.py

@@ -18,8 +18,7 @@
     "map_projection" / map_projection_record[this.file_descriptor.map_projection.number_of_records],
     "platform_position" / platform_position_record,
     "attitude" / attitude_record,
-    "radiometric_data"
-    / radiometric_data_record[this.file_descriptor.radiometric_data.number_of_records],
+    "radiometric_data" / radiometric_data_record,
     "data_quality_summary" / data_quality_summary_record,
     "facility_related_data_1" / facility_related_data_record,
     "facility_related_data_2" / facility_related_data_record,

ceos_alos2/tests/test_dicttoolz.py

@@ -74,6 +74,20 @@ def test_dissoc(keys):
     assert actual == expected
 
 
+@pytest.mark.parametrize(
+    ["key", "value", "expected"],
+    (
+        pytest.param("b", "abc", {"a": 1, "b": "abc"}),
+        pytest.param("c", 2, {"a": 1, "c": 2}),
+    ),
+)
+def test_assoc(key, value, expected):
+    mapping = {"a": 1}
+    actual = dicttoolz.assoc(key, value, mapping)
+
+    assert actual == expected
+
+
 @pytest.mark.parametrize(
     ["funcs", "default", "expected"],
     (

ceos_alos2/tests/test_sar_leader.py

@@ -7,6 +7,7 @@
     dataset_summary,
     map_projection,
     platform_position,
+    radiometric_data,
 )
 from ceos_alos2.testing import assert_identical
 
@@ -794,3 +795,144 @@ def test_transform_attitude(self, raw_points, expected):
         actual = attitude.transform_attitude(mapping)
 
         assert_identical(actual, expected)
+
+
+class TestRadiometricData:
+    @pytest.mark.parametrize(
+        ["mapping", "expected"],
+        (
+            pytest.param(
+                {"a": {"a": 0, "b": 1, "c": 2, "d": 3}},
+                (
+                    {
+                        "a": (["i", "j"], [[0, 1], [2, 3]], {}),
+                        "i": (
+                            "i",
+                            ["horizontal", "vertical"],
+                            {"long_name": "reception polarization"},
+                        ),
+                        "j": (
+                            "j",
+                            ["horizontal", "vertical"],
+                            {"long_name": "transmission polarization"},
+                        ),
+                    },
+                    {},
+                ),
+            ),
+            pytest.param(
+                ({"a": {"a": 0, "b": 1, "c": 2, "d": 3}}, {"a": "def"}),
+                (
+                    {
+                        "a": (["i", "j"], [[0, 1], [2, 3]], {}),
+                        "i": (
+                            "i",
+                            ["horizontal", "vertical"],
+                            {"long_name": "reception polarization"},
+                        ),
+                        "j": (
+                            "j",
+                            ["horizontal", "vertical"],
+                            {"long_name": "transmission polarization"},
+                        ),
+                    },
+                    {"a": "def"},
+                ),
+            ),
+            pytest.param(
+                {
+                    "a": {"a": 1j, "b": 2j, "c": 3j, "d": 4j},
+                    "b": {"f": 0j, "e": 1j, "d": 2j, "c": 3j},
+                },
+                (
+                    {
+                        "a": (["i", "j"], [[1j, 2j], [3j, 4j]], {}),
+                        "b": (["i", "j"], [[0j, 1j], [2j, 3j]], {}),
+                        "i": (
+                            "i",
+                            ["horizontal", "vertical"],
+                            {"long_name": "reception polarization"},
+                        ),
+                        "j": (
+                            "j",
+                            ["horizontal", "vertical"],
+                            {"long_name": "transmission polarization"},
+                        ),
+                    },
+                    {},
+                ),
+            ),
+        ),
+    )
+    def test_transform_matrices(self, mapping, expected):
+        actual = radiometric_data.transform_matrices(mapping)
+
+        assert actual == expected
+
+    @pytest.mark.parametrize(
+        ["mapping", "expected"],
+        (
+            pytest.param(
+                {
+                    "preamble": "",
+                    "radiometric_data_records_sequence_number": 0,
+                    "number_of_radiometric_fields": 1,
+                    "blanks": "",
+                },
+                Group(path=None, url=None, data={}, attrs={}),
+                id="ignored",
+            ),
+            pytest.param(
+                {"calibration_factor": (-10.0, {"formula": "abc"})},
+                Group(
+                    path=None,
+                    url=None,
+                    data={"calibration_factor": Variable((), -10.0, {"formula": "abc"})},
+                    attrs={},
+                ),
+                id="calibration_factor",
+            ),
+            pytest.param(
+                {
+                    "distortion_matrix": (
+                        {
+                            "a": {"a": 1, "b": 2, "c": 3, "d": 4},
+                            "b": {"f": 0, "e": 1, "d": 2, "c": 3},
+                        },
+                        {"formula": "def"},
+                    )
+                },
+                Group(
+                    path=None,
+                    url=None,
+                    data={
+                        "distortion_matrix": Group(
+                            path=None,
+                            url=None,
+                            data={
+                                "a": Variable(["i", "j"], [[1, 2], [3, 4]], {}),
+                                "b": Variable(["i", "j"], [[0, 1], [2, 3]], {}),
+                                "i": Variable(
+                                    ["i"],
+                                    ["horizontal", "vertical"],
+                                    {"long_name": "reception polarization"},
+                                ),
+                                "j": Variable(
+                                    ["j"],
+                                    ["horizontal", "vertical"],
+                                    {"long_name": "transmission polarization"},
+                                ),
+                            },
+                            attrs={"formula": "def"},
+                        )
+                    },
+                    attrs={},
+                ),
+                id="distortion_matrix",
+            ),
+        ),
+    )
+    def test_transform_radiometric_data(self, mapping, expected):
+        actual = radiometric_data.transform_radiometric_data(mapping)
+
+        assert_identical(actual, expected)