Merge branch 'L2SS-1582_construct-hdf5-from-json-statistics' into 'main'

Resolve L2SS-1582 "Migrate case insensitive dict from station control"

Closes L2SS-1582

See merge request !92

README.md

@@ -122,6 +122,7 @@ tox -e debug tests.requests.test_prometheus
 
 ## Release notes
 
+- 0.18.8 - Migrate case insensitive dict from station control
 - 0.18.7 - Add support for various ZeroMQ package receivers
 - 0.18.6 - Compatability with new black versions
 - 0.18.5 - Compatability with python 3.10 and higher

VERSION

-0.18.7
+0.18.8

lofar_station_client/common/__init__.py

+#  Copyright (C) 2024 ASTRON (Netherlands Institute for Radio Astronomy)
+#  SPDX-License-Identifier: Apache-2.0
+
+""" Common classes used in station """
+
+from ._case_insensitive_dict import CaseInsensitiveDict, ReversibleKeysView
+from ._case_insensitive_string import CaseInsensitiveString
+
+__all__ = ["CaseInsensitiveDict", "CaseInsensitiveString", "ReversibleKeysView"]

lofar_station_client/common/_case_insensitive_dict.py

+#  Copyright (C) 2024 ASTRON (Netherlands Institute for Radio Astronomy)
+#  SPDX-License-Identifier: Apache-2.0
+
+""" Provides a special dictionary with case-insensitive keys """
+
+import abc
+from collections import UserDict
+from typing import List
+from typing import Tuple
+from typing import Union
+
+from ._case_insensitive_string import CaseInsensitiveString
+
+
+def _case_insensitive_comprehend_keys(data: dict) -> List[CaseInsensitiveString]:
+    return [CaseInsensitiveString(key) for key in data]
+
+
+def _case_insensitive_comprehend_items(
+    data: dict,
+) -> List[Tuple[CaseInsensitiveString, any]]:
+    return [(CaseInsensitiveString(key), value) for key, value in data.items()]
+
+
+class ReversibleIterator:
+    """Reversible iterator using instance of self method
+
+    See real-python for yield iterator method:
+        https://realpython.com/python-reverse-list/#the-special-method-__reversed__
+    """
+
+    def __init__(self, data: List, start: int, stop: int, step: int):
+        self.data = data
+        self.current = start
+        self.stop = stop
+        self.step = step
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        if self.current == self.stop:
+            raise StopIteration
+
+        elem = self.data[self.current]
+        self.current += self.step
+        return elem
+
+    def __reversed__(self):
+        return ReversibleIterator(self.data, self.stop, self.current, -1)
+
+
+class AbstractReversibleView(abc.ABC):
+    """An abstract reversible view"""
+
+    def __init__(self, data: UserDict):
+        self.data = data
+        self.len = len(data)
+
+    def __repr__(self):
+        return f"{self.__class__.__name__}({self.data})"
+
+    @abc.abstractmethod
+    def __iter__(self):
+        pass
+
+    @abc.abstractmethod
+    def __reversed__(self):
+        pass
+
+
+class ReversibleItemsView(AbstractReversibleView):
+    """Reversible view on items"""
+
+    def __iter__(self):
+        return ReversibleIterator(
+            _case_insensitive_comprehend_items(self.data.data), 0, self.len, 1
+        )
+
+    def __reversed__(self):
+        return ReversibleIterator(
+            _case_insensitive_comprehend_items(self.data.data), self.len - 1, -1, -1
+        )
+
+
+class ReversibleKeysView(AbstractReversibleView):
+    """Reversible view on keys"""
+
+    def __iter__(self):
+        return ReversibleIterator(
+            _case_insensitive_comprehend_keys(self.data.data), 0, self.len, 1
+        )
+
+    def __reversed__(self):
+        return ReversibleIterator(
+            _case_insensitive_comprehend_keys(self.data.data), self.len - 1, -1, -1
+        )
+
+
+class ReversibleValuesView(AbstractReversibleView):
+    """Reversible view on values"""
+
+    def __iter__(self):
+        return ReversibleIterator(list(self.data.data.values()), 0, self.len, 1)
+
+    def __reversed__(self):
+        return ReversibleIterator(list(self.data.data.values()), self.len - 1, -1, -1)
+
+
+class CaseInsensitiveDict(UserDict):
+    """Special dictionary that ignores key casing if string
+
+    While UserDict is the least performant / flexible it ensures __set_item__ and
+    __get_item__ are used in all code paths reducing LoC severely.
+
+    Background reference:
+        https://realpython.com/inherit-python-dict/#creating-dictionary-like-classes-in-python
+
+    Alternative (should this stop working at some point):
+        https://github.com/DeveloperRSquared/case-insensitive-dict/blob/main/case_insensitive_dict/case_insensitive_dict.py
+    """
+
+    def __setitem__(self, key, value):
+        if isinstance(key, str):
+            key = CaseInsensitiveString(key)
+        super().__setitem__(key, value)
+
+    def __getitem__(self, key: Union[int, str]):
+        if isinstance(key, str):
+            key = CaseInsensitiveString(key)
+        return super().__getitem__(key)
+
+    def __iter__(self):
+        return ReversibleIterator(
+            _case_insensitive_comprehend_keys(self.data), 0, len(self.data), 1
+        )
+
+    def __contains__(self, key):
+        if isinstance(key, str):
+            key = CaseInsensitiveString(key)
+        return super().__contains__(key)
+
+    def keys(self) -> ReversibleKeysView:
+        return ReversibleKeysView(self)
+
+    def values(self) -> ReversibleValuesView:
+        return ReversibleValuesView(self)
+
+    def items(self) -> ReversibleItemsView:
+        return ReversibleItemsView(self)

lofar_station_client/common/_case_insensitive_string.py

+#  Copyright (C) 2024 ASTRON (Netherlands Institute for Radio Astronomy)
+#  SPDX-License-Identifier: Apache-2.0
+
+"""Special string that ignores casing in comparison"""
+
+
+class CaseInsensitiveString(str):
+    """Special string that ignores casing in comparison"""
+
+    def __eq__(self, other):
+        if isinstance(other, str):
+            return self.casefold() == other.casefold()
+
+        return self.casefold() == other
+
+    def __hash__(self):
+        return hash(self.__str__())
+
+    def __contains__(self, key):
+        if isinstance(key, str):
+            return key.casefold() in str(self)
+        return key in str(self)
+
+    def __str__(self) -> str:
+        return self.casefold().__str__()
+
+    def __repr__(self) -> str:
+        return self.casefold().__repr__()

lofar_station_client/statistics/statistics_data.py

-#  Copyright (C) 2023 ASTRON (Netherlands Institute for Radio Astronomy)
+#  Copyright (C) 2024 ASTRON (Netherlands Institute for Radio Astronomy)
 #  SPDX-License-Identifier: Apache-2.0
 
 # too-few-public-methods
@@ -193,26 +193,26 @@ class StatisticsFileHeader:
     antennafield_device: str = attribute(optional=True)
     """ Name of the antennafield device """
 
-    antenna_names: str = attribute(optional=True)
+    antenna_names: [str] = attribute(optional=True)
     """ Antenna names """
 
     antenna_type: str = attribute(optional=True)
     """ The type of antenna in this field (LBA or HBA). """
 
-    antenna_quality: str = attribute(optional=True)
+    antenna_quality: [str] = attribute(optional=True)
     """ The quality of each antenna, as a string. """
 
-    antenna_usage_mask: str = attribute(optional=True)
+    antenna_usage_mask: [bool] = attribute(optional=True)
     """ Whether each antenna would have been used. """
 
-    antenna_reference_itrf: str = attribute(optional=True)
+    antenna_reference_itrf: [str] = attribute(optional=True)
     """ Absolute reference position of each tile, in ITRF (XYZ) """
 
-    fpga_firmware_version: str = attribute(optional=True)
-    fpga_hardware_version: str = attribute(optional=True)
+    fpga_firmware_version: [str] = attribute(optional=True)
+    fpga_hardware_version: [str] = attribute(optional=True)
 
-    rcu_pcb_id: int = attribute(optional=True)
-    rcu_pcb_version: str = attribute(optional=True)
+    rcu_pcb_id: [int] = attribute(optional=True)
+    rcu_pcb_version: [str] = attribute(optional=True)
 
     def __eq__(self, other):
         for attr in [

tests/common/__init__.py

+#  Copyright (C) 2024 ASTRON (Netherlands Institute for Radio Astronomy)
+#  SPDX-License-Identifier: Apache-2.0

tests/common/test_case_insensitive_dict.py

+#  Copyright (C) 2024 ASTRON (Netherlands Institute for Radio Astronomy)
+#  SPDX-License-Identifier: Apache-2.0
+from enum import Enum
+
+from lofar_station_client.common import CaseInsensitiveDict
+from lofar_station_client.common import CaseInsensitiveString
+from lofar_station_client.common import ReversibleKeysView
+from tests import base
+
+
+class TestCaseInsensitiveDict(base.TestCase):
+    def test_set_get_item(self):
+        """Get and set an item with different casing"""
+
+        t_value = "VALUE"
+        t_key = "KEY"
+        t_dict = CaseInsensitiveDict()
+
+        t_dict[t_key] = t_value
+
+        self.assertEqual(t_value, t_dict[t_key.lower()])
+        self.assertEqual(t_value, t_dict.get(t_key.lower()))
+
+    def test_set_overwrite(self):
+        """Overwrite a previous element with different casing"""
+
+        t_value = "VALUE"
+        t_key = "KEY"
+        t_dict = CaseInsensitiveDict()
+
+        t_dict[t_key] = t_value
+        t_dict[t_key.lower()] = t_value.lower()
+
+        self.assertEqual(t_value.lower(), t_dict[t_key])
+        self.assertEqual(t_value.lower(), t_dict.get(t_key))
+
+    class ConstructTestEnum(Enum):
+        DICT = "dict"
+        ITER = "iter"
+        KWARGS = "kwargs"
+
+    def construct_base(self, test_type: ConstructTestEnum):
+        t_key1 = "KEY1"
+        t_key2 = "key2"
+        t_value1 = 123
+        t_value2 = 456
+        t_mapping = {t_key1: t_value1, t_key2: t_value2}
+
+        t_dict = CaseInsensitiveDict()
+        if test_type is self.ConstructTestEnum.DICT:
+            t_dict = CaseInsensitiveDict(t_mapping)
+        elif test_type is self.ConstructTestEnum.ITER:
+            t_dict = CaseInsensitiveDict(t_mapping.items())
+        elif test_type is self.ConstructTestEnum.KWARGS:
+            t_dict = CaseInsensitiveDict(KEY1=t_value1, key2=t_value2)
+
+        self.assertEqual(t_value1, t_dict[t_key1.lower()])
+        self.assertEqual(t_value2, t_dict.get(t_key2.upper()))
+
+    def test_construct_mapping(self):
+        self.construct_base(self.ConstructTestEnum.DICT)
+
+    def test_construct_iterable(self):
+        self.construct_base(self.ConstructTestEnum.ITER)
+
+    def test_construct_kwargs(self):
+        self.construct_base(self.ConstructTestEnum.KWARGS)
+
+    def test_setdefault(self):
+        t_key = "KEY"
+        t_value = "value"
+        t_dict = CaseInsensitiveDict()
+
+        t_dict.setdefault(t_key, t_value)
+
+        self.assertIn(t_key.lower(), t_dict.keys())
+
+        for key in t_dict.keys():
+            self.assertEqual(t_key, key)
+            self.assertIsInstance(key, CaseInsensitiveString)
+
+    def test_keys(self):
+        t_key = "KEY"
+        t_value = "value"
+        t_dict = CaseInsensitiveDict()
+
+        t_dict[t_key] = t_value
+
+        self.assertIn(t_key.lower(), t_dict.keys())
+        self.assertIsInstance(t_dict.keys(), ReversibleKeysView)
+
+        for key in t_dict.keys():
+            self.assertEqual(t_key, key)
+            self.assertIsInstance(key, CaseInsensitiveString)
+
+    def test_items(self):
+        t_key = "KEY"
+        t_value = "VALUE"
+        t_dict = CaseInsensitiveDict()
+
+        t_dict[t_key] = t_value
+
+        for key, value in t_dict.items():
+            self.assertEqual(t_key, key)
+            self.assertIsInstance(key, CaseInsensitiveString)
+            self.assertNotEqual(t_value.casefold(), value)
+
+    def test_values(self):
+        t_key = "KEY"
+        t_value = "VALUE"
+        t_dict = CaseInsensitiveDict()
+
+        t_dict[t_key] = t_value
+
+        for value in t_dict.values():
+            self.assertEqual(t_value, value)
+            self.assertIsInstance(value, str)
+
+    def test_in(self):
+        t_key = "KEY"
+        t_value = "VALUE"
+        t_dict = CaseInsensitiveDict()
+
+        t_dict[t_key] = t_value
+
+        self.assertTrue(t_key.lower() in t_dict)
+
+    def test_reverse(self):
+        t_key1 = "KEY1"
+        t_key2 = "KEY2"
+        t_value = "VALUE"
+        t_dict = CaseInsensitiveDict()
+
+        t_dict[t_key1] = t_value
+        t_dict[t_key2] = t_value
+
+        forward = []
+        for key, _ in t_dict.items():
+            forward.append(key)
+
+        backward = []
+        for key, _ in reversed(t_dict.items()):
+            backward.append(key)
+
+        self.assertEqual(forward[0], backward[1])
+        self.assertEqual(forward[1], backward[0])
+
+        backward = []
+        for key in reversed(t_dict.keys()):
+            backward.append(key)
+
+        self.assertEqual(forward[0], backward[1])
+        self.assertEqual(forward[1], backward[0])
+
+        forward = []
+        for item in t_dict.values():
+            forward.append(item)
+
+        backward = []
+        for value in reversed(t_dict.values()):
+            backward.append(value)
+
+        self.assertEqual(forward[0], backward[1])
+        self.assertEqual(forward[1], backward[0])

tests/common/test_case_insensitive_string.py

+#  Copyright (C) 2024 ASTRON (Netherlands Institute for Radio Astronomy)
+#  SPDX-License-Identifier: Apache-2.0
+from lofar_station_client.common import CaseInsensitiveString
+
+from tests import base
+
+
+class TestCaseInsensitiveString(base.TestCase):
+    def test_a_in_b(self):
+        """Get and set an item with different casing"""
+
+        self.assertTrue(CaseInsensitiveString("hba0") in CaseInsensitiveString("HBA0"))
+
+    def test_b_in_a(self):
+        """Get and set an item with different casing"""
+
+        self.assertTrue(CaseInsensitiveString("HBA0") in CaseInsensitiveString("hba0"))
+
+    def test_a_not_in_b(self):
+        """Get and set an item with different casing"""
+
+        self.assertFalse(CaseInsensitiveString("hba0") in CaseInsensitiveString("LBA0"))