diff --git a/armarx_memory/aron/aron_dataclass.py b/armarx_memory/aron/aron_dataclass.py
index c370c2abfeb6ab2581eb9313517d56f058028076..60a3c6d1635b4bc4a07797eaab9f1db2a6837fb9 100644
--- a/armarx_memory/aron/aron_dataclass.py
+++ b/armarx_memory/aron/aron_dataclass.py
@@ -1,3 +1,5 @@
+import logging
+
import dataclasses as dc
import typing as ty
@@ -10,34 +12,31 @@ class AronDataclass:
ConversionOptions = ConversionOptions
def to_dict(self) -> ty.Dict[str, ty.Any]:
- from armarx_memory.aron.conversion.dataclass_from_to_pythonic import (
- dataclass_to_dict,
- )
+ from armarx_memory.aron.conversion import dataclass_from_to_pythonic
+ return dataclass_from_to_pythonic.dataclass_to_dict(self)
- return dataclass_to_dict(self)
-
- def to_aron_ice(self) -> "armarx.aron.data.dto.Dict":
- from armarx_memory.aron.conversion.dataclass_from_to_aron_ice import (
- dataclass_to_aron_ice,
- )
-
- return dataclass_to_aron_ice(self, options=self._get_conversion_options())
+ def to_aron_ice(
+ self,
+ logger: logging.Logger = None,
+ ) -> "armarx.aron.data.dto.Dict":
+ from armarx_memory.aron.conversion import dataclass_from_to_aron_ice
+ return dataclass_from_to_aron_ice.dataclass_to_aron_ice(self, logger=logger)
@classmethod
def from_dict(cls, data: ty.Dict[str, ty.Any]) -> "AronDataclass":
from armarx_memory.aron.conversion.dataclass_from_to_pythonic import (
dataclass_from_dict,
)
-
return dataclass_from_dict(cls, data)
@classmethod
- def from_aron_ice(cls, data: "armarx.aron.data.dto.Dict") -> "AronDataclass":
- from armarx_memory.aron.conversion.dataclass_from_to_aron_ice import (
- dataclass_from_aron_ice,
- )
-
- return dataclass_from_aron_ice(cls, data, options=cls._get_conversion_options())
+ def from_aron_ice(
+ cls,
+ data: "armarx.aron.data.dto.Dict",
+ logger: logging.Logger = None,
+ ) -> "AronDataclass":
+ from armarx_memory.aron.conversion import dataclass_from_to_aron_ice
+ return dataclass_from_to_aron_ice.dataclass_from_aron_ice(cls, aron=data, logger=logger)
@classmethod
def _get_conversion_options(cls) -> ty.Optional["ConversionOptions"]:
diff --git a/armarx_memory/aron/conversion/dataclass_from_to_aron_ice.py b/armarx_memory/aron/conversion/dataclass_from_to_aron_ice.py
index 139ca3570421c1a1037b4c9fb61803435ac8ce21..a5982e5b5ff0eca1043a61ab4e2beedf7b8cec9a 100644
--- a/armarx_memory/aron/conversion/dataclass_from_to_aron_ice.py
+++ b/armarx_memory/aron/conversion/dataclass_from_to_aron_ice.py
@@ -2,31 +2,39 @@ import typing as ty
import logging
-from .options import ConversionOptions
-
def dataclass_to_aron_ice(
obj,
- options: ty.Optional[ConversionOptions] = None,
logger: ty.Optional[logging.Logger] = None,
) -> "armarx.aron.data.dto.GenericData":
from .dataclass_from_to_pythonic import dataclass_to_dict
from .pythonic_from_to_aron_ice import pythonic_to_aron_ice
+ if logger is not None:
+ logger.debug(f"Convert object of ARON dataclass {type(obj)} to Pythonic types ...")
data = dataclass_to_dict(obj, logger=logger)
- aron = pythonic_to_aron_ice(data, options=options)
+
+ if logger is not None:
+ logger.debug("Convert Pythonic types to ARON Ice DTOs ...")
+ aron = pythonic_to_aron_ice(data)
+
return aron
def dataclass_from_aron_ice(
cls,
aron: "armarx.aron.data.dto.GenericData",
- options: ty.Optional[ConversionOptions] = None,
logger: ty.Optional[logging.Logger] = None,
):
from .dataclass_from_to_pythonic import dataclass_from_dict
from .pythonic_from_to_aron_ice import pythonic_from_aron_ice
- data = pythonic_from_aron_ice(aron, options=options)
+ if logger is not None:
+ logger.debug("Convert ARON Ice DTOs to Pythonic types ...")
+ data = pythonic_from_aron_ice(aron, logger=logger)
+
+ if logger is not None:
+ logger.debug(f"Convert Pythonic types to ARON dataclass {cls} ...")
obj = dataclass_from_dict(cls, data, logger=logger)
+
return obj
diff --git a/armarx_memory/aron/conversion/dataclass_from_to_pythonic.py b/armarx_memory/aron/conversion/dataclass_from_to_pythonic.py
index 06f154c8d2bf5cd294d728afe1a18948f227ea2c..88e9d364b984286d9709ed932448f089c7797b09 100644
--- a/armarx_memory/aron/conversion/dataclass_from_to_pythonic.py
+++ b/armarx_memory/aron/conversion/dataclass_from_to_pythonic.py
@@ -1,96 +1,301 @@
import logging
-import dataclasses as dc
import typing as ty
-from .options import ConversionOptions
+from armarx_memory.aron.conversion.options import ConversionOptions
-def dataclass_to_dict(
- obj,
- logger: ty.Optional[logging.Logger] = None,
-) -> ty.Dict[str, ty.Any]:
- """
- Deeply converts a dataclass to a dict.
+class DataclassFromToDict:
- :param obj: An object of a dataclass.
- :param logger: An optional logger.
- :return: A dict containing pythonic data types.
- """
- return dc.asdict(obj)
+ def __init__(self, logger: logging.Logger = None):
+ self.logger = logger
+ def _prefix(self, depth: int) -> str:
+ return " " * depth
-def dataclass_from_dict(
- cls,
- data: ty.Dict,
- logger: ty.Optional[logging.Logger] = None,
-):
- """
- Deeply converts a dictionary with pythonic data types
- to an instance of the given dataclass.
+ def dataclass_from_dict(
+ self,
+ cls,
+ data: ty.Dict,
+ depth: int = 0,
+ ):
+ """
+ Deeply converts a dictionary with pythonic data types
+ to an instance of the given dataclass.
- :param cls: The dataclass.
- :param data: The data.
- :param logger: A logger.
- :return: An instance of the dataclass.
- """
+ :param cls: The dataclass.
+ :param data: The data.
+ :param depth: The current recursion depth. Only used for logging.
+ :return: An instance of the dataclass.
+ """
+
+ pre = self._prefix(depth)
+
+ if self.logger is not None:
+ self.logger.debug(f"{pre}Construct value of type {cls.__name__} from a {type(data)} ...")
- if logger is not None:
- logger.info(f"\nConstruct class {cls.__name__} from a {type(data)} ...")
+ if cls == type(data):
+ if self.logger is not None:
+ self.logger.debug(f"{pre}> Type matches exactly. Return data {cls.__name__} as-is.")
+ # Nothing to do.
+ return data
+
+ from armarx_memory.aron.aron_dataclass import AronDataclass
+ if issubclass(cls, AronDataclass):
+ conversion_options = cls._get_conversion_options()
+ else:
+ conversion_options = None
- try:
- field_types = cls.__annotations__
- except AttributeError:
+ try:
+ field_types = cls.__annotations__
+ except AttributeError:
+ return self.non_dataclass_from_dict(cls=cls, data=data, depth=depth)
+
+ # Build kwargs for cls.
+ kwargs = dict()
+ for field_name, value in data.items():
+ if conversion_options is not None:
+ field_name = conversion_options.name_aron_to_python(field_name)
+
+ try:
+ field_type = field_types[field_name]
+ except KeyError as e:
+ raise KeyError(
+ f"Found no dataclass field '{field_name}' in ARON dataclass {cls} matching the data entry. "
+ "Available are: " + ", ".join(f"'{f}'" for f in field_types))
+
+ field_value = self.field_value_from_pythonic(field_name=field_name, field_type=field_type, value=value, depth=depth)
+
+ kwargs[field_name] = field_value
+
+ # Construct from kwargs and return.
+ return cls(**kwargs)
+
+ def non_dataclass_from_dict(self, cls, data, depth: int):
# Not a dataclass. Can just try to deliver kwargs. Or return data.
+
+ pre = self._prefix(depth)
+
+ method_name = "from_aron_ice"
+ if isinstance(cls.__dict__.get(method_name, None), classmethod):
+ if self.logger is not None:
+ self.logger.debug(f"{pre}Not a dataclass, but provides method '{method_name}()'.")
+ return cls.from_aron_ice(data)
+
try:
- return cls(**data)
+ result = cls(**data)
except TypeError:
+ if self.logger is not None:
+ self.logger.debug(f"{pre}Not a dataclass. Return data of type {type(data)} as-is..")
return data
+ else:
+ if self.logger is not None:
+ self.logger.debug(f"{pre}Not a dataclass. Construct {cls} from kwargs.")
+ return result
- # Build kwargs for cls.
- kwargs = dict()
- for field_name, value in data.items():
- field_type = field_types[field_name]
+
+ def field_value_from_pythonic(
+ self,
+ field_name: str,
+ field_type,
+ value,
+ depth: int,
+ ):
+ pre = self._prefix(depth)
try:
origin = field_type.__origin__
except AttributeError:
origin = None
- if logger is not None:
+ if self.logger is not None:
value_type = type(value)
- logger.debug(
- f"- Field '{field_name}' of type: {field_type}"
- f"\n- origin: {origin}"
- f"\n- data type: {value_type}"
+ self.logger.debug(
+ f"{pre}- Field '{field_name}':"
+ f"\n{pre} - type of annot.: {field_type}"
+ f"\n{pre} - origin: {origin}"
+ f"\n{pre} - type of value: {value_type}"
)
- if origin in (ty.List, list):
+ if field_type == type(None):
+ if self.logger is not None:
+ self.logger.debug(f"{pre}> Process NoneType")
+ if value is None:
+ return None
+
+ elif origin in (ty.List, list):
[vt] = field_type.__args__
- if logger is not None:
- logger.debug(f"> Process list of {vt} ")
- field_value = [dataclass_from_dict(vt, v) for v in value]
+ if self.logger is not None:
+ self.logger.debug(f"{pre}> Process list of {vt} ")
+ return [self.dataclass_from_dict(vt, v, depth=depth+1) for v in value]
elif origin in (ty.Dict, dict):
kt, vt = field_type.__args__
- if logger is not None:
- logger.debug(f"> Process dict {kt} -> {vt}")
+ if self.logger is not None:
+ self.logger.debug(f"{pre}> Process dict {kt} -> {vt}")
if value is not None:
- field_value = {
- kt(k): dataclass_from_dict(vt, v) for k, v in value.items()
+ return {
+ kt(k): self.dataclass_from_dict(vt, v, depth=depth+1) for k, v in value.items()
}
else:
- field_value = None
+ return None
+
+ elif origin == ty.Union:
+ if self.logger is not None:
+ self.logger.debug(f"{pre}> Process union.")
+
+ union_types = field_type.__args__
+ for union_type in union_types:
+ if self.logger is not None:
+ self.logger.debug(f"{pre} - Try option {union_type} ...")
+
+ result = self.dataclass_from_dict(union_type, value, depth=depth+1)
+ if isinstance(result, union_type):
+ return result
else:
- if logger is not None:
- logger.debug(f"> Process other: {field_type}")
+ if self.logger is not None:
+ self.logger.debug(f"{pre}> Process other type: {field_type}")
try:
- field_value = dataclass_from_dict(field_type, value)
+ return self.dataclass_from_dict(field_type, value, depth=depth+1)
except AttributeError:
# Cannot convert.
- field_value = value
+ if self.logger is not None:
+ self.logger.debug(f"{pre}> Not a dataclass. Take value {value} as-is..")
+ return value
+
+ def dataclass_to_dict(
+ self,
+ obj,
+ depth: ty.Optional[int] = 0,
+ ) -> ty.Dict[str, ty.Any]:
+ """
+ Deeply converts an ARON dataclass to a dict.
+
+ :param obj: An object of a dataclass.
+ :param logger: An optional logger.
+ :param depth:
+ :return: A dict containing pythonic data types.
+ """
+
+ pre = self._prefix(depth)
+
+ from armarx_memory.aron.aron_dataclass import AronDataclass
+ if isinstance(obj, AronDataclass):
+ conversion_options = obj._get_conversion_options()
+ else:
+ conversion_options = None
+
+ # Does not respect conversion_options.
+ # dc.asdict(obj)
+
+ if self.logger is not None:
+ self.logger.debug(f"{pre}Construct dictionary from object of class {obj.__class__.__name__} ...")
- kwargs[field_name] = field_value
+ field_types = obj.__annotations__
+
+ # Build kwargs for cls.
+ data = dict()
+ for field_name, field_type in field_types.items():
+ origin = field_type.__dict__.get("__origin__", None)
+
+ if ty.ClassVar in [field_type, origin]:
+ continue
+
+ try:
+ value = obj.__dict__[field_name]
+ except KeyError:
+ raise KeyError(f"Field '{field_name}' of type {field_type} (origin: {origin})"
+ f" not found in object of type {type(obj)}."
+ f" Available are: " + ", ".join(f"'{f}'" for f in obj.__dict__.keys()))
+
+ if conversion_options is not None:
+ aron_field_name = conversion_options.name_python_to_aron(field_name)
+ else:
+ aron_field_name = field_name
+
+ field_value = self.field_value_to_pythonic(name=aron_field_name, type_=field_type, value=value, depth=depth)
+
+ data[aron_field_name] = field_value
+
+ return data
+
+ def field_value_to_pythonic(
+ self,
+ name: str,
+ type_,
+ value,
+ depth: int,
+ ):
+ pre = self._prefix(depth)
+ origin = type_.__dict__.get("__origin__", None)
+
+ if self.logger is not None:
+ value_type = type(value)
+ self.logger.debug(
+ f"{pre}- Field '{name}':"
+ f"\n{pre} - type of annot.: {type_}"
+ f"\n{pre} - origin: {origin}"
+ f"\n{pre} - type of value: {value_type}"
+ )
+
+ if value is None:
+ if self.logger is not None:
+ self.logger.debug(f"{pre}> Process NoneType")
+
+ assert type_ == type(None), type_
+ return None
+
+ elif isinstance(value, list):
+ if self.logger is not None:
+ self.logger.debug(f"{pre}> Process list ")
+ return [self.dataclass_to_dict(v, depth=depth+1) for v in value]
+
+ elif isinstance(value, dict):
+ return {k: self.dataclass_to_dict(v, depth=depth+1) for k, v in value.items()}
+
+ else:
+ if self.logger is not None:
+ self.logger.debug(f"{pre}> Process other type: {type_}")
+ try:
+ return self.dataclass_to_dict(value, depth=depth + 1)
+ except AttributeError:
+ # Cannot convert.
+ if self.logger is not None:
+ self.logger.debug(f"{pre}> Not a dataclass. Return value {value} as-is..")
+ return value
+
+
+def dataclass_to_dict(
+ obj,
+ logger: ty.Optional[logging.Logger] = None,
+) -> ty.Dict[str, ty.Any]:
+ """
+ Deeply converts a dataclass to a dict.
+
+ :param obj: An object of an ARON dataclass.
+ :param logger: An optional logger.
+ :return: A dict containing pythonic data types.
+ """
+
+ converter = DataclassFromToDict(logger=logger)
+ return converter.dataclass_to_dict(obj=obj)
+
+
+def dataclass_from_dict(
+ cls,
+ data: ty.Dict,
+ logger: ty.Optional[logging.Logger] = None,
+):
+ """
+ Deeply converts a dictionary with pythonic data types
+ to an instance of the given ARON dataclass.
+
+ :param cls: The dataclass.
+ :param data: The data.
+ :param logger: A logger.
+ :return: An instance of the dataclass.
+ """
- return cls(**kwargs)
+ converter = DataclassFromToDict(logger=logger)
+ return converter.dataclass_from_dict(cls=cls, data=data)
diff --git a/armarx_memory/aron/conversion/options.py b/armarx_memory/aron/conversion/options.py
index 2f582238e12f8d1bea44600b70e67c15b72e2fbf..c49c48a140fd7636d33ef221f1e4ebb5621d8ed8 100644
--- a/armarx_memory/aron/conversion/options.py
+++ b/armarx_memory/aron/conversion/options.py
@@ -12,16 +12,14 @@ class ConversionOptions:
def name_python_to_aron(self, python_name: str) -> str:
aron_name = self.names_python_to_aron_dict.get(python_name, None)
+ if aron_name is not None:
+ return aron_name
- if aron_name is None:
- if self.names_snake_case_to_camel_case:
- from .name_conversion import snake_case_to_camel_case
-
- aron_name = snake_case_to_camel_case(python_name)
- else:
- aron_name = python_name
+ if self.names_snake_case_to_camel_case:
+ from .name_conversion import snake_case_to_camel_case
+ return snake_case_to_camel_case(python_name)
- return aron_name
+ return python_name
def name_aron_to_python(self, aron_name: str) -> str:
for python, aron in self.names_python_to_aron_dict.items():
@@ -30,9 +28,6 @@ class ConversionOptions:
if self.names_snake_case_to_camel_case:
from .name_conversion import camel_case_to_snake_case
+ return camel_case_to_snake_case(aron_name)
- python_name = camel_case_to_snake_case(aron_name)
- else:
- python_name = aron_name
-
- return python_name
+ return aron_name
diff --git a/armarx_memory/aron/conversion/pythonic_from_to_aron_ice.py b/armarx_memory/aron/conversion/pythonic_from_to_aron_ice.py
index fbb7b6e1137c928e4c189e8b8c35283964dc3cbc..5bc7f8add64e5352d930f0cf306876e71e83c3b4 100644
--- a/armarx_memory/aron/conversion/pythonic_from_to_aron_ice.py
+++ b/armarx_memory/aron/conversion/pythonic_from_to_aron_ice.py
@@ -1,15 +1,14 @@
import enum
+import logging
import numpy as np
import typing as ty
from armarx_memory.aron.aron_ice_types import AronIceTypes, dtypes_dict
-from armarx_memory.aron.conversion.options import ConversionOptions
def pythonic_to_aron_ice(
value: ty.Any,
- options: ty.Optional[ConversionOptions] = None,
) -> "armarx.aron.data.dto.GenericData":
"""
Deeply converts objects/values of pythonic types to their Aron Ice counterparts.
@@ -37,14 +36,7 @@ def pythonic_to_aron_ice(
return pythonic_to_aron_ice(value.value) # int
elif isinstance(value, dict):
- a = AronIceTypes.dict(
- {
- (
- options.name_python_to_aron(k) if options is not None else k
- ): pythonic_to_aron_ice(v)
- for k, v in value.items()
- }
- )
+ a = AronIceTypes.dict({k: pythonic_to_aron_ice(v) for k, v in value.items()})
return a
elif isinstance(value, AronIceTypes.Dict):
@@ -63,23 +55,19 @@ def pythonic_to_aron_ice(
def pythonic_from_aron_ice(
data: "armarx.aron.data.dto.GenericData",
- options: ty.Optional[ConversionOptions] = None,
+ logger: ty.Optional[logging.Logger] = None,
) -> ty.Any:
"""
Deeply converts an Aron data Ice object to its pythonic representation.
+ :param logger:
:param data: The Aron data Ice object.
:param options: Conversion options.
:return: The pythonic representation.
"""
def handle_dict(elements):
- return {
- (
- options.name_aron_to_python(k) if options is not None else k
- ): pythonic_from_aron_ice(v)
- for k, v in elements.items()
- }
+ return {k: pythonic_from_aron_ice(v) for k, v in elements.items()}
def handle_list(elements):
return list(map(pythonic_from_aron_ice, elements))
diff --git a/armarx_memory/core/MemoryID.py b/armarx_memory/core/MemoryID.py
index 7414c8a81c3d3bd40c1e21e6ca2f059b0e029aa7..5156be5281c4d873c1a492b713b0d1aab7843ad9 100644
--- a/armarx_memory/core/MemoryID.py
+++ b/armarx_memory/core/MemoryID.py
@@ -225,8 +225,9 @@ class MemoryID(ice_twin.IceTwin):
self.timestamp_usec = date_time_conv.from_ice(ice.timestamp)
self.instance_index = ice.instanceIndex
+
@classmethod
- def from_aron(cls, aron: "armarx.aron.data.dto.GenericData") -> "MemoryID":
+ def from_aron_ice(cls, aron: "armarx.aron.data.dto.GenericData") -> "MemoryID":
from armarx_memory.aron.conversion import from_aron
data = from_aron(aron)
@@ -239,7 +240,11 @@ class MemoryID(ice_twin.IceTwin):
self.instance_index = data["instanceIndex"]
return self
- def to_aron(self) -> "armarx.aron.data.dto.GenericData":
+ @classmethod
+ def from_aron(cls, aron: "armarx.aron.data.dto.GenericData") -> "MemoryID":
+ return cls.from_aron_ice(aron=aron)
+
+ def to_aron_ice(self) -> "armarx.aron.data.dto.GenericData":
import numpy as np
from armarx_memory.aron.conversion import to_aron
@@ -264,3 +269,7 @@ class MemoryID(ice_twin.IceTwin):
"instanceIndex": self.instance_index,
}
return to_aron(data)
+
+
+ def to_aron(self) -> "armarx.aron.data.dto.GenericData":
+ return self.to_aron_ice()