diff --git a/armarx_core/ice_conversion/ice_converter.py b/armarx_core/ice_conversion/ice_converter.py
index fa26827dfaa8914f985a71041d181862e49b08c1..f5de6f6a8d7260f6951528e77d681e9698b34812 100644
--- a/armarx_core/ice_conversion/ice_converter.py
+++ b/armarx_core/ice_conversion/ice_converter.py
@@ -83,7 +83,7 @@ class IceConverter(abc.ABC):
def to_ice(self, bo, *args, **kwargs):
"""Convert the Python Business Object(s) to Ice Data Transfer Object(s)."""
- for t, handler in self._handlers_from_ice.items():
+ for t, handler in self._handlers_to_ice.items():
if isinstance(bo, t):
return handler(bo, *args, **kwargs)
return self._to_ice(bo, *args, **kwargs)
diff --git a/armarx_core/time/duration.py b/armarx_core/time/duration.py
index 36efbdc0620cde9001f090ec7c07b92c60f97f7b..b8912d9342e83184d13a39c90a900334b6a820f9 100644
--- a/armarx_core/time/duration.py
+++ b/armarx_core/time/duration.py
@@ -25,6 +25,9 @@ class Duration(AronDataclass):
def __repr__(self) -> str:
return f"<Duration {self.microSeconds} µs>"
+
+ def toMilliSeconds(self) -> int:
+ return int(self.microSeconds / 1000)
def __str__(self) -> str:
unit = "µs"
diff --git a/armarx_memory/aron/aron_ice_types/__init__.py b/armarx_memory/aron/aron_ice_types/__init__.py
index f1830451cb420b2a38b65294addd443e5052ecfc..e5d93adfd7a9fb603e2be6c0c0b68d34fd122736 100644
--- a/armarx_memory/aron/aron_ice_types/__init__.py
+++ b/armarx_memory/aron/aron_ice_types/__init__.py
@@ -1,2 +1,3 @@
from .import_aron_slice import import_aron_slice
-from .aron_ice_types import AronIceTypes
+from .aron_ice_types import AronDataIceTypes
+from .aron_type_ice_types import AronTypeIceTypes
diff --git a/armarx_memory/aron/aron_ice_types/aron_ice_types.py b/armarx_memory/aron/aron_ice_types/aron_ice_types.py
index ea37c04cfa9108a1cd2400d00e7025a47570b94e..fa39e67bf8cac6b81ba357a077ad5235cb1182e6 100644
--- a/armarx_memory/aron/aron_ice_types/aron_ice_types.py
+++ b/armarx_memory/aron/aron_ice_types/aron_ice_types.py
@@ -4,7 +4,7 @@ armarx_aron = import_aron_slice()
data_dto = armarx_aron.data.dto
-class AronIceTypes:
+class AronDataIceTypes:
ARON_VERSION = armarx_aron.Version()
Version = armarx_aron.Version
diff --git a/armarx_memory/aron/aron_ice_types/aron_type_ice_types.py b/armarx_memory/aron/aron_ice_types/aron_type_ice_types.py
new file mode 100644
index 0000000000000000000000000000000000000000..22949e0c044471d66dd40ff20a6aa76059ff98c3
--- /dev/null
+++ b/armarx_memory/aron/aron_ice_types/aron_type_ice_types.py
@@ -0,0 +1,420 @@
+from typing import Literal, List as typing_list, Dict as typing_dict, Optional
+from .import_aron_slice import import_aron_slice
+
+
+armarx_aron = import_aron_slice()
+type_dto = armarx_aron.type.dto
+
+AronIceTypeMaybe = Literal["NONE", "OPTIONAL", "RAW_PTR", "SHARED_PTR", "UNIQUE_PTR"]
+
+AronQuaternionElementTyp = Literal["FLOAT64", "FLOAT32"]
+AronQuaternionDefaultValue = Literal["ONES", "ZEROS", "DEFAULT"] | str
+
+AronIceMatrixDefaultValue = Literal["IDENTITY", "ONES", "ZEROS", "DEFAULT"] | str
+AronIceMatrixElementType = Literal[
+ "UINT8", "UINT16", "UINT32", "INT8", "INT16", "INT32", "INT64", "FLOAT32", "FLOAT64"
+]
+
+AronIceNDArrayElementType = Literal[
+ "UINT8", "INT8", "UINT16", "INT16", "UINT32", "INT32", "FLOAT32", "FLOAT64"
+]
+AronIceNDArrayDefaultValue = Literal["ONES", "ZEROS", "DEFAULT"] | str
+
+AronIceEnumIntDefaultValue = Literal["DEFAULT"] | str
+
+AronIcePointCloudVoxelType = Literal[
+ "POINT_XYZ",
+ "POINT_XYZI",
+ "POINT_XYZL",
+ "POINT_XYZRGB",
+ "POINT_XYZRGBL",
+ "POINT_XYZRGBA",
+ "POINT_XYZHSV",
+]
+AronIcePointCloudDefaultValue = Literal["IDENTITY", "ONES", "ZEROS", "DEFAULT"] | str
+
+AronIceImagePixelType = Literal["RGB24", "DEPTH32"]
+AronIceImageDefaultValue = Literal["IDENTITY", "ONES", "ZEROS", "DEFAULT"] | str
+
+
+class AronTypeIceTypes:
+ _maybe = {
+ "NONE": armarx_aron.type.Maybe.NONE,
+ "OPTIONAL": armarx_aron.type.Maybe.OPTIONAL,
+ "RAW_PTR": armarx_aron.type.Maybe.RAW_PTR,
+ "SHARED_PTR": armarx_aron.type.Maybe.SHARED_PTR,
+ "UNIQUE_PTR": armarx_aron.type.Maybe.UNIQUE_PTR,
+ }
+ _quaternion_element_type = {
+ "FLOAT64": armarx_aron.type.quaternion.ElementType.FLOAT64,
+ "FLOAT32": armarx_aron.type.quaternion.ElementType.FLOAT32,
+ }
+ _quaternion_default_value = {
+ "ONES": armarx_aron.type.quaternion.default_value.ONES,
+ "ZEROS": armarx_aron.type.quaternion.default_value.ZEROS,
+ "DEFAULT": armarx_aron.type.quaternion.default_value.DEFAULT,
+ }
+ _matrix_element_type = {
+ "UINT8": armarx_aron.type.matrix.ElementType.UINT8,
+ "UINT16": armarx_aron.type.matrix.ElementType.UINT16,
+ "UINT32": armarx_aron.type.matrix.ElementType.UINT32,
+ "INT8": armarx_aron.type.matrix.ElementType.INT8,
+ "INT16": armarx_aron.type.matrix.ElementType.INT16,
+ "INT32": armarx_aron.type.matrix.ElementType.INT32,
+ "INT64": armarx_aron.type.matrix.ElementType.INT64,
+ "FLOAT32": armarx_aron.type.matrix.ElementType.FLOAT32,
+ "FLOAT64": armarx_aron.type.matrix.ElementType.FLOAT64,
+ }
+ _matrix_default_value = {
+ "IDENTITY": armarx_aron.type.matrix.default_value.IDENTITY,
+ "ONES": armarx_aron.type.matrix.default_value.ONES,
+ "ZEROS": armarx_aron.type.matrix.default_value.ZEROS,
+ "DEFAULT": armarx_aron.type.matrix.default_value.DEFAULT,
+ }
+ _ndarray_element_type = {
+ "UINT8": armarx_aron.type.ndarray.ElementType.UINT8,
+ "INT8": armarx_aron.type.ndarray.ElementType.INT8,
+ "UINT16": armarx_aron.type.ndarray.ElementType.UINT16,
+ "INT16": armarx_aron.type.ndarray.ElementType.INT16,
+ "UINT32": armarx_aron.type.ndarray.ElementType.UINT32,
+ "INT32": armarx_aron.type.ndarray.ElementType.INT32,
+ "FLOAT32": armarx_aron.type.ndarray.ElementType.FLOAT32,
+ "FLOAT64": armarx_aron.type.ndarray.ElementType.FLOAT64,
+ }
+ _enum_int_default_value = {
+ "DEFAULT": armarx_aron.type.aron_enum.default_value.DEFAULT
+ }
+ _point_cloud_voxel_type = {
+ "POINT_XYZ": armarx_aron.type.pointcloud.VoxelType.POINT_XYZ,
+ "POINT_XYZI": armarx_aron.type.pointcloud.VoxelType.POINT_XYZI,
+ "POINT_XYZL": armarx_aron.type.pointcloud.VoxelType.POINT_XYZL,
+ "POINT_XYZRGB": armarx_aron.type.pointcloud.VoxelType.POINT_XYZRGB,
+ "POINT_XYZRGBL": armarx_aron.type.pointcloud.VoxelType.POINT_XYZRGBL,
+ "POINT_XYZRGBA": armarx_aron.type.pointcloud.VoxelType.POINT_XYZRGBA,
+ "POINT_XYZHSV": armarx_aron.type.pointcloud.VoxelType.POINT_XYZHSV,
+ }
+ _point_cloud_default_value = {
+ "IDENTITY": armarx_aron.type.pointcloud.default_value.IDENTITY,
+ "ONES": armarx_aron.type.pointcloud.default_value.ONES,
+ "ZEROS": armarx_aron.type.pointcloud.default_value.ZEROS,
+ "DEFAULT": armarx_aron.type.pointcloud.default_value.DEFAULT,
+ }
+ _image_pixel_type = {
+ "RGB24": armarx_aron.type.image.PixelType.RGB24,
+ "DEPTH32": armarx_aron.type.image.PixelType.DEPTH32,
+ }
+ _image_default_value = {
+ "IDENTITY": armarx_aron.type.image.default_value.IDENTITY,
+ "ONES": armarx_aron.type.image.default_value.ONES,
+ "ZEROS": armarx_aron.type.image.default_value.ZEROS,
+ "DEFAULT": armarx_aron.type.image.default_value.DEFAULT,
+ }
+
+ ARON_VERSION = armarx_aron.Version()
+
+ Version = armarx_aron.Version
+
+ Bool = type_dto.AronBool
+ Int = type_dto.AronInt
+ Long = type_dto.AronLong
+ Float = type_dto.AronFloat
+ Double = type_dto.AronDouble
+ String = type_dto.AronString
+
+ List = type_dto.List
+ Tuple = type_dto.Tuple
+ Pair = type_dto.Pair
+ Quaternion = type_dto.Quaternion
+ Dict = type_dto.Dict
+
+ Object = type_dto.AronObject
+
+ Matrix = type_dto.Matrix
+ NDArray = type_dto.NDArray
+
+ GenericType = type_dto.GenericType
+ IntEnum = type_dto.IntEnum
+ PointCloud = type_dto.PointCloud
+ Image = type_dto.Image
+
+ @classmethod
+ def string(
+ cls, default_value: Optional[str] = None, maybe: AronIceTypeMaybe = "NONE"
+ ) -> String:
+ if default_value is None:
+ ret = cls.String(
+ VERSION=cls.ARON_VERSION,
+ maybe=cls._maybe[maybe],
+ )
+ else:
+ ret = cls.String(
+ VERSION=cls.ARON_VERSION,
+ defaultValue=default_value,
+ maybe=cls._maybe[maybe],
+ )
+ return ret
+
+ @classmethod
+ def bool(cls, default_value: Optional[bool] = None, maybe: AronIceTypeMaybe = "NONE") -> Bool:
+ if default_value is None:
+ ret = cls.Bool(
+ VERSION=cls.ARON_VERSION,
+ maybe=cls._maybe[maybe],
+ )
+ else:
+ ret = cls.Bool(
+ VERSION=cls.ARON_VERSION,
+ defaultValue=default_value,
+ maybe=cls._maybe[maybe],
+ )
+ return ret
+
+ @classmethod
+ def long(cls, default_value: Optional[int] = None, maybe: AronIceTypeMaybe = "NONE") -> Long:
+ if default_value is None:
+ ret = cls.Long(
+ VERSION=cls.ARON_VERSION,
+ defaultValue=default_value,
+ maybe=cls._maybe[maybe],
+ )
+ else:
+ ret = cls.Long(
+ VERSION=cls.ARON_VERSION,
+ maybe=cls._maybe[maybe],
+ )
+ return ret
+
+ @classmethod
+ def int(cls, default_value: Optional[int] = None, maybe: AronIceTypeMaybe = "NONE") -> Int:
+ if default_value is None:
+ ret = cls.Int(
+ VERSION=cls.ARON_VERSION,
+ maybe=cls._maybe[maybe],
+ )
+ else:
+ ret = cls.Int(
+ VERSION=cls.ARON_VERSION,
+ defaultValue=default_value,
+ maybe=cls._maybe[maybe],
+ )
+ return ret
+
+ @classmethod
+ def double(
+ cls, default_value: Optional[float] = None, maybe: AronIceTypeMaybe = "NONE"
+ ) -> Double:
+ if default_value is None:
+ ret = cls.Double(
+ VERSION=cls.ARON_VERSION,
+ maybe=cls._maybe[maybe],
+ )
+ else:
+ ret = cls.Double(
+ VERSION=cls.ARON_VERSION,
+ defaultValue=default_value,
+ maybe=cls._maybe[maybe],
+ )
+ return ret
+
+ @classmethod
+ def float(
+ cls, default_value: Optional[float] = None, maybe: AronIceTypeMaybe = "NONE"
+ ) -> Float:
+ if default_value is None:
+ ret = cls.Float(
+ VERSION=cls.ARON_VERSION,
+ maybe=cls._maybe[maybe],
+ )
+ else:
+ ret = cls.Float(
+ VERSION=cls.ARON_VERSION,
+ defaultValue=default_value,
+ maybe=cls._maybe[maybe],
+ )
+ return ret
+
+ @classmethod
+ def list(cls, accepted_type, maybe: AronIceTypeMaybe = "NONE") -> List:
+ return cls.List(
+ VERSION=cls.ARON_VERSION,
+ maybe=cls._maybe[maybe],
+ acceptedType=accepted_type,
+ )
+
+ @classmethod
+ def tuple(cls, element_types, maybe: AronIceTypeMaybe = "NONE") -> Tuple:
+ return cls.Tuple(
+ VERSION=cls.ARON_VERSION,
+ maybe=cls._maybe[maybe],
+ elementTypes=element_types,
+ )
+
+ @classmethod
+ def pair(
+ cls, accepted_type1, accepted_type2, maybe: AronIceTypeMaybe = "NONE"
+ ) -> Pair:
+ return cls.Pair(
+ VERSION=cls.ARON_VERSION,
+ maybe=cls._maybe[maybe],
+ acceptedType1=accepted_type1,
+ acceptedType2=accepted_type2,
+ )
+
+ @classmethod
+ def quaternion(
+ cls,
+ element_type: AronQuaternionElementTyp = "FLOAT64",
+ maybe: AronIceTypeMaybe = "NONE",
+ default_value: AronQuaternionDefaultValue = "DEFAULT",
+ ) -> Quaternion:
+ default = (
+ cls._quaternion_default_value[default_value]
+ if default_value in cls._quaternion_default_value
+ else default_value
+ )
+ return cls.Quaternion(
+ VERSION=cls.ARON_VERSION,
+ maybe=cls._maybe[maybe],
+ elementType=cls._quaternion_element_type[element_type],
+ defaultValue=default,
+ )
+
+ @classmethod
+ def dict(cls, accepted_type, maybe: AronIceTypeMaybe = "NONE") -> Dict:
+ return cls.Dict(
+ VERSION=cls.ARON_VERSION,
+ maybe=cls._maybe[maybe],
+ acceptedType=accepted_type,
+ )
+
+ @classmethod
+ def matrix(
+ cls,
+ rows,
+ cols,
+ element_type: AronIceMatrixElementType = "FLOAT32",
+ maybe: AronIceTypeMaybe = "NONE",
+ default_value: AronIceMatrixDefaultValue = "DEFAULT",
+ ) -> Matrix:
+ default = (
+ cls._matrix_default_value[default_value]
+ if default_value in cls._matrix_default_value
+ else default_value
+ )
+ element = (
+ cls._matrix_element_type[element_type]
+ if element_type in cls._matrix_element_type
+ else element_type
+ )
+ return cls.Matrix(
+ VERSION=cls.ARON_VERSION,
+ elementType=element,
+ maybe=cls._maybe[maybe],
+ rows=rows,
+ cols=cols,
+ defaultValue=default,
+ )
+
+ @classmethod
+ def nd_array(
+ cls,
+ ndimensions: int,
+ element_type: AronIceNDArrayElementType = "FLOAT32",
+ maybe: AronIceTypeMaybe = "NONE",
+ default_value: AronIceNDArrayDefaultValue = "DEFAULT",
+ ) -> NDArray:
+ default = (
+ cls._matrix_default_value[default_value]
+ if default_value in cls._matrix_default_value
+ else default_value
+ )
+ return cls.NDArray(
+ VERSION=cls.ARON_VERSION,
+ ndimensions=ndimensions,
+ elementType=cls._ndarray_element_type[element_type],
+ maybe=cls._maybe[maybe],
+ defaultValue=default,
+ )
+
+ @classmethod
+ def generic_type(cls, maybe: AronIceTypeMaybe = "NONE") -> GenericType:
+ return cls.GenericType(VERSION=cls.ARON_VERSION, maybe=cls._maybe[maybe])
+
+ @classmethod
+ def int_enum(
+ cls,
+ name: str,
+ accepted_values={},
+ default_value: AronIceEnumIntDefaultValue = "DEFAULT",
+ maybe: AronIceTypeMaybe = "NONE",
+ ) -> IntEnum:
+ default = (
+ cls._enum_int_default_value[default_value]
+ if default_value in cls._enum_int_default_value
+ else default_value
+ )
+ return cls.IntEnum(
+ VERSION=cls.ARON_VERSION,
+ maybe=cls._maybe[maybe],
+ acceptedValues=accepted_values,
+ enumName=name,
+ defaultValue=default,
+ )
+
+ @classmethod
+ def point_cloud(
+ cls,
+ maybe: AronIceTypeMaybe = "NONE",
+ voxel_type: AronIcePointCloudVoxelType = "POINT_XYZ",
+ default_value: AronIcePointCloudDefaultValue = "DEFAULT",
+ ) -> PointCloud:
+ default = (
+ cls._point_cloud_default_value[default_value]
+ if default_value in cls._point_cloud_default_value
+ else default_value
+ )
+ return cls.PointCloud(
+ VERSION=cls.ARON_VERSION,
+ maybe=cls._maybe[maybe],
+ voxelType=cls._point_cloud_voxel_type[voxel_type],
+ defaultValue=default,
+ )
+
+ @classmethod
+ def image(
+ cls,
+ maybe: AronIceTypeMaybe = "NONE",
+ pixel_type: AronIceImagePixelType = "RGB24",
+ default_value: AronIceImageDefaultValue = "DEFAULT",
+ ) -> Image:
+ default = (
+ cls._image_default_value[default_value]
+ if default_value in cls._image_default_value
+ else default_value
+ )
+ return cls.Image(
+ VERSION=cls.ARON_VERSION,
+ maybe=cls._maybe[maybe],
+ pixelType=cls._image_pixel_type[pixel_type],
+ defaultValue=default,
+ )
+
+ @classmethod
+ def object(
+ cls,
+ object_name: str,
+ templates: typing_list[str] = [],
+ template_instantiations: typing_list[str] = [],
+ element_types: typing_dict[str, GenericType] = {},
+ maybe: AronIceTypeMaybe = "NONE",
+ parent: Object = None,
+ ) -> Object:
+ return cls.Object(
+ VERSION=cls.ARON_VERSION,
+ maybe=cls._maybe[maybe],
+ objectName=object_name,
+ templates=templates,
+ parent=parent,
+ templateInstantiations=template_instantiations,
+ elementTypes=element_types,
+ )
diff --git a/armarx_memory/aron/conversion/ndarray/common.py b/armarx_memory/aron/conversion/ndarray/common.py
index a16d26d5d0c6bdb62c40242db0273c7cc57f0873..80192d038f20f7763e5c10746e46f20cdafd0d1d 100644
--- a/armarx_memory/aron/conversion/ndarray/common.py
+++ b/armarx_memory/aron/conversion/ndarray/common.py
@@ -2,7 +2,7 @@ import numpy as np
import typing as ty
import armarx
-from armarx_memory.aron.aron_ice_types import AronIceTypes
+from armarx_memory.aron.aron_ice_types import AronDataIceTypes
from armarx_memory.aron.conversion.ndarray.point_cloud import PointCloudConversions
@@ -42,7 +42,7 @@ def is_point_cloud(array: np.ndarray):
def ndarray_to_aron(
value: np.ndarray,
-) -> AronIceTypes.NDArray:
+) -> AronDataIceTypes.NDArray:
"""
Convert a numpy array to an ARON Ice NDArray.
:param value: The numpy array.
@@ -54,7 +54,7 @@ def ndarray_to_aron(
# General case.
shape = (*value.shape, 1) # value.itemsize)
- return AronIceTypes.NDArray(
+ return AronDataIceTypes.NDArray(
shape=shape,
type=dtypes_dict_to_aron[str(value.dtype)],
data=value.tobytes(),
@@ -62,7 +62,7 @@ def ndarray_to_aron(
def ndarray_from_aron(
- data: AronIceTypes.NDArray,
+ data: AronDataIceTypes.NDArray,
) -> np.ndarray:
"""
Convert an ARON Ice NDArray to a suitable numpy array.
diff --git a/armarx_memory/aron/conversion/ndarray/point_cloud.py b/armarx_memory/aron/conversion/ndarray/point_cloud.py
index 2fc41eb1e64673ff85aa89d82e0c39c26ec5169e..509316b5c26eff11d0367fc6191be0f69401ed14 100644
--- a/armarx_memory/aron/conversion/ndarray/point_cloud.py
+++ b/armarx_memory/aron/conversion/ndarray/point_cloud.py
@@ -1,7 +1,7 @@
import numpy as np
import typing as ty
-from armarx_memory.aron.aron_ice_types import AronIceTypes
+from armarx_memory.aron.aron_ice_types import AronDataIceTypes
class PointCloudConversions:
@@ -156,7 +156,7 @@ class PointCloudConversions:
pcl_point_cloud = cls.point_cloud_with_paddings(py_point_cloud, pcl_dtype)
shape = (*pcl_point_cloud.shape, pcl_point_cloud.itemsize)
- return AronIceTypes.NDArray(
+ return AronDataIceTypes.NDArray(
shape=shape,
type=type_string,
data=pcl_point_cloud.tobytes(),
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 6bdfe76e7dc96612f94522c5e697845f963cf521..f82bbc2aa923f6ed571d971ff3463729db8188c2 100644
--- a/armarx_memory/aron/conversion/pythonic_from_to_aron_ice.py
+++ b/armarx_memory/aron/conversion/pythonic_from_to_aron_ice.py
@@ -5,7 +5,7 @@ import numpy as np
import typing as ty
import armarx
-from armarx_memory.aron.aron_ice_types import AronIceTypes
+from armarx_memory.aron.aron_ice_types import AronDataIceTypes
def pythonic_to_aron_ice(
@@ -23,11 +23,11 @@ def pythonic_to_aron_ice(
if value is None:
return None
if isinstance(value, str):
- return AronIceTypes.string(value)
+ return AronDataIceTypes.string(value)
elif isinstance(value, bool):
- return AronIceTypes.bool(value)
+ return AronDataIceTypes.bool(value)
elif isinstance(value, np.int64):
- return AronIceTypes.long(int(value))
+ return AronDataIceTypes.long(int(value))
elif isinstance(value, int) or isinstance(value, np.int32):
assert(isinstance(value, int)), value
assert 'invalid value - expected int' not in str(AronIceTypes.int(int(value))), \
@@ -35,11 +35,11 @@ def pythonic_to_aron_ice(
'but assigned a plain int value, or converted the value somewhere in the meantime?'
return AronIceTypes.int(int(value))
elif isinstance(value, np.float64):
- return AronIceTypes.double(float(value))
+ return AronDataIceTypes.double(float(value))
elif isinstance(value, float) or isinstance(value, np.float32):
- return AronIceTypes.float(float(value))
+ return AronDataIceTypes.float(float(value))
elif isinstance(value, list):
- return AronIceTypes.list(list(map(pythonic_to_aron_ice, value)))
+ return AronDataIceTypes.list(list(map(pythonic_to_aron_ice, value)))
elif isinstance(value, enum.IntEnum):
assert(isinstance(value.value, int)), f'Value of IntEnum is not an int: {value.value}'
return pythonic_to_aron_ice(value.value) # int
@@ -52,10 +52,10 @@ def pythonic_to_aron_ice(
return pythonic_to_aron_ice(value.value)
elif isinstance(value, dict):
- a = AronIceTypes.dict({k: pythonic_to_aron_ice(v) for k, v in value.items()})
+ a = AronDataIceTypes.dict({k: pythonic_to_aron_ice(v) for k, v in value.items()})
return a
- elif isinstance(value, AronIceTypes.Dict):
+ elif isinstance(value, AronDataIceTypes.Dict):
return value
elif isinstance(value, np.ndarray):
@@ -98,10 +98,10 @@ def pythonic_from_aron_ice(
elif isinstance(data, (float, int, str)):
return data
- if isinstance(data, AronIceTypes.NDArray):
+ if isinstance(data, AronDataIceTypes.NDArray):
return ndarray_from_aron(data)
- if isinstance(data, AronIceTypes.Long):
+ if isinstance(data, AronDataIceTypes.Long):
return np.int64(data.value)
if isinstance(data, np.int32):
diff --git a/armarx_memory/client/Commit.py b/armarx_memory/client/Commit.py
index f908ad14f86ec8e9a287221aec30281cdc2d2cc9..0ef112d45921ef9bc3968acaf43618b1d07ef75b 100644
--- a/armarx_memory/client/Commit.py
+++ b/armarx_memory/client/Commit.py
@@ -1,6 +1,6 @@
from typing import List, Optional
-from armarx_memory.aron.aron_ice_types import AronIceTypes
+from armarx_memory.aron.aron_ice_types import AronDataIceTypes
from armarx_memory.core import MemoryID, time_usec, DateTimeIceConverter
from armarx_memory.ice_conv import ice_twin
@@ -18,7 +18,7 @@ class EntityUpdate(ice_twin.IceTwin):
def __init__(
self,
entity_id: MemoryID = None,
- instances_data: List[AronIceTypes.Data] = None,
+ instances_data: List[AronDataIceTypes.Data] = None,
referenced_time_usec: Optional[int] = None,
confidence: Optional[float] = None,
time_sent_usec: Optional[int] = None,
diff --git a/armarx_memory/client/Reader.py b/armarx_memory/client/Reader.py
index f75b3c38db8e4bc89f976c257845f7734945e277..5f9eb89936ceb88ce70e0cfe8241f81620970e79 100644
--- a/armarx_memory/client/Reader.py
+++ b/armarx_memory/client/Reader.py
@@ -9,7 +9,7 @@ from armarx import armem
from armarx.armem import data as dto # The ice type namespace.
from armarx_memory.core import MemoryID
-from armarx_memory.aron.aron_ice_types import AronIceTypes
+from armarx_memory.aron.aron_ice_types import AronDataIceTypes
class Reader:
diff --git a/armarx_skills/manager/dto.py b/armarx_skills/manager/dto.py
index ae21f3f1171b88c1c39a895b63f090e4b17ffaf0..05097253c1fbba8cdce4c8b9c02dc01ae4212042 100644
--- a/armarx_skills/manager/dto.py
+++ b/armarx_skills/manager/dto.py
@@ -1,13 +1,43 @@
-import typing as ty
+from typing import Dict, List
from armarx_skills.slice import load_slice
-
load_slice()
-from armarx.skills.manager.dto import \
- ProviderID, SkillID, SkillDescription, ProviderInfo, \
- SkillExecutionID, SkillStatusUpdate, SkillExecutionRequest
+from armarx_memory.aron.aron_ice_types import AronDataIceTypes
+
+from armarx.skills.manager.dto import (
+ ProviderID,
+ SkillID,
+ FluxioException,
+ SkillDescription,
+ ProviderInfo,
+ SkillExecutionID,
+ SkillStatusUpdate,
+ SkillExecutionRequest,
+ FluxioSkill,
+ FluxioIdentificator,
+ FluxioParameter,
+ FluxioNode,
+ FluxioEdge,
+ FluxioProfile,
+ FluxioProvider,
+ FluxioValue,
+ FluxioSkillStatusUpdate,
+ FluxioTypeIdentificator,
+)
+
+from armarx_memory.aron.aron_ice_types import AronDataIceTypes
-SkillDescriptionMap = ty.Dict[SkillID, SkillDescription]
-SkillStatusUpdateMap = ty.Dict[SkillExecutionID, SkillStatusUpdate]
+SkillDescriptionMap = Dict[SkillID, SkillDescription]
+SkillStatusUpdateMap = Dict[SkillExecutionID, SkillStatusUpdate]
+FluxioTypesMap = Dict[str, AronDataIceTypes]
+FluxioSkillList = List[FluxioSkill]
+FluxioParameterList = List[FluxioParameter]
+FluxioValueList = List[FluxioValue]
+FluxioNodeList = List[FluxioNode]
+FluxioEdgeList = List[FluxioEdge]
+FluxioProviderList = List[FluxioProvider]
+FluxioProfileList = List[FluxioProfile]
+FluxioSkillStatusUpdateList = List[FluxioSkillStatusUpdate]
+FluxioIdentificatorList = List[FluxioIdentificator]
diff --git a/armarx_skills/manager/fluxio_control_node.py b/armarx_skills/manager/fluxio_control_node.py
new file mode 100644
index 0000000000000000000000000000000000000000..debab2a5bb170d890cb591a5c24178bf1c2c8035
--- /dev/null
+++ b/armarx_skills/manager/fluxio_control_node.py
@@ -0,0 +1,62 @@
+import dataclasses as dc
+from typing import List
+from enum import Enum
+
+from armarx_skills.manager import dto
+from armarx_memory.ice_conv.ice_converter import IceConverter
+from armarx_skills.manager.fluxio_node import Node, NodeTypes
+from armarx_skills.manager.fluxio_parameter import Parameter, FluxioParameterConv
+
+
+class ControlType(str, Enum):
+ SPLITTER = "SPLITTER"
+ AND_MERGER = "AND_MERGER"
+
+
+@dc.dataclass
+class ControlNode(Node):
+
+ controlType: ControlType
+ parameters: List[Parameter]
+
+ def __init__(
+ self,
+ nodeType: NodeTypes,
+ nodeId: str,
+ xPos: float,
+ yPos: float,
+ controlType: ControlType,
+ parameters: List[Parameter],
+ ) -> None:
+ super().__init__(nodeType, nodeId, xPos, yPos)
+ self.controlType = controlType
+ self.parameters = parameters
+
+
+class FluxioControlNodeConv(IceConverter):
+ def __init__(self):
+ super().__init__()
+
+ @classmethod
+ def _import_dto(cls):
+ return dto.FluxioNode
+
+ def _from_ice(self, dt: dto.FluxioNode) -> ControlNode:
+ return ControlNode(
+ nodeType=NodeTypes(dt.nodeType),
+ nodeId=dt.nodeId,
+ xPos=dt.xPos,
+ yPos=dt.yPos,
+ controlType=ControlType(dt.controlType),
+ parameters=[FluxioParameterConv().from_ice(p) for p in dt.parameters],
+ )
+
+ def _to_ice(self, bo: ControlNode) -> dto.FluxioNode:
+ return dto.FluxioNode(
+ nodeType=bo.nodeType,
+ nodeId=bo.nodeId,
+ xPos=bo.xPos,
+ yPos=bo.yPos,
+ controlType=bo.controlType,
+ parameters=[FluxioParameterConv().to_ice(p) for p in bo.parameters],
+ )
diff --git a/armarx_skills/manager/fluxio_data.py b/armarx_skills/manager/fluxio_data.py
new file mode 100644
index 0000000000000000000000000000000000000000..50c980aa887fe5ff295ed630b25987c0e5cf332d
--- /dev/null
+++ b/armarx_skills/manager/fluxio_data.py
@@ -0,0 +1,85 @@
+import dataclasses as dc
+from enum import Enum
+import numpy as np
+from numpy import ndarray
+
+from Ice import Object
+
+
+class FluxioDataIdentifier(str, Enum):
+ STRING = "STRING"
+ BOOL = "BOOL"
+ INT = "INT"
+ LONG = "LONG"
+ FLOAT = "FLOAT"
+ DOUBLE = "DOUBLE"
+ EVENT = "EVENT"
+ LIST = "LIST"
+ DICT = "DICT"
+ NDARRAY = "NDARRAY"
+
+
+@dc.dataclass
+class FluxioBaseData: # EVENT
+ data_identifier: FluxioDataIdentifier
+
+ def __init__(self, data_identifier: FluxioDataIdentifier):
+ self.data_identifier = data_identifier
+
+
+@dc.dataclass
+class FluxioSimpleData(
+ FluxioBaseData
+): # BOOL=bool, INT=int, LONG=int, FLOAT=float, DOUBLE=flout, STRING=str
+ value: str | bool | int | float
+
+ def __init__(
+ self, data_identifier: FluxioDataIdentifier, value: str | bool | int | float
+ ):
+ super().__init__(data_identifier)
+ self.value = value
+
+
+@dc.dataclass
+class FluxioListData(FluxioBaseData):
+ elements: list[FluxioBaseData]
+
+ def __init__(
+ self, data_identifier: FluxioDataIdentifier, elements: list[FluxioBaseData]
+ ):
+ super().__init__(data_identifier)
+ self.elements = elements
+
+
+@dc.dataclass
+class FluxioDictEntry:
+ key: str
+ value: FluxioBaseData
+
+
+@dc.dataclass
+class FluxioDictData(FluxioBaseData):
+ elements: list[FluxioDictEntry]
+
+ def __init__(self, data_identifier: FluxioDataIdentifier):
+ super().__init__(data_identifier)
+ self.elements = []
+
+ def add_attribute(self, key: str, value: FluxioBaseData):
+ self.elements.append(FluxioDictEntry(key=key, value=value))
+
+
+@dc.dataclass
+class FluxioNDArrayData(FluxioBaseData):
+ shape: tuple[int, ...]
+ base: ndarray
+
+ def __init__(
+ self,
+ data_identifier: FluxioDataIdentifier,
+ shape: tuple[int, ...],
+ base: ndarray,
+ ):
+ super().__init__(data_identifier)
+ self.shape = shape
+ self.base = base
diff --git a/armarx_skills/manager/fluxio_data_aron_converter.py b/armarx_skills/manager/fluxio_data_aron_converter.py
new file mode 100644
index 0000000000000000000000000000000000000000..03f19c2514690f25014932f100b007bcb32d25a6
--- /dev/null
+++ b/armarx_skills/manager/fluxio_data_aron_converter.py
@@ -0,0 +1,85 @@
+import numpy as np
+from armarx_memory.aron.aron_ice_types import AronDataIceTypes
+from armarx_memory.aron.conversion.ndarray.common import (
+ ndarray_from_aron,
+ ndarray_to_aron,
+)
+from armarx_skills.manager.fluxio_data import (
+ FluxioBaseData,
+ FluxioDataIdentifier,
+ FluxioSimpleData,
+ FluxioDictData,
+ FluxioNDArrayData,
+ FluxioListData,
+)
+
+
+# :)
+class FluxioDataAronConverter:
+ @classmethod
+ def aron_data_to_fluxio_data(
+ cls, aron_data: AronDataIceTypes.Data
+ ) -> FluxioBaseData | FluxioSimpleData | FluxioDictData | FluxioNDArrayData | FluxioListData:
+ ret = None
+ if isinstance(aron_data, AronDataIceTypes.String):
+ ret = FluxioSimpleData(FluxioDataIdentifier.STRING, aron_data.value)
+ elif isinstance(aron_data, AronDataIceTypes.Bool):
+ ret = FluxioSimpleData(FluxioDataIdentifier.BOOL, aron_data.value)
+ elif isinstance(aron_data, AronDataIceTypes.Int):
+ ret = FluxioSimpleData(FluxioDataIdentifier.INT, aron_data.value)
+ elif isinstance(aron_data, AronDataIceTypes.Long):
+ ret = FluxioSimpleData(FluxioDataIdentifier.LONG, aron_data.value)
+ elif isinstance(aron_data, AronDataIceTypes.Float):
+ ret = FluxioSimpleData(FluxioDataIdentifier.FLOAT, aron_data.value)
+ elif isinstance(aron_data, AronDataIceTypes.Double):
+ ret = FluxioSimpleData(FluxioDataIdentifier.DOUBLE, aron_data.value)
+ elif isinstance(aron_data, AronDataIceTypes.List):
+ elements = [cls.aron_data_to_fluxio_data(e) for e in aron_data.elements]
+ ret = FluxioListData(FluxioDataIdentifier.LIST, elements)
+ elif isinstance(aron_data, AronDataIceTypes.Dict):
+ ret = FluxioDictData(FluxioDataIdentifier.DICT)
+ for k, v in aron_data.elements.items():
+ ret.add_attribute(k, cls.aron_data_to_fluxio_data(v))
+ elif isinstance(aron_data, AronDataIceTypes.NDArray):
+ numpy_nparray = ndarray_from_aron(aron_data)
+ assert numpy_nparray.base is not None
+ ret = FluxioNDArrayData(
+ FluxioDataIdentifier.NDARRAY, numpy_nparray.shape, numpy_nparray.base
+ )
+ return ret
+
+ @classmethod
+ def fluxio_data_to_aron(
+ cls,
+ fluxio_data: FluxioBaseData,
+ ) -> any: # just take the any and stop complaining
+ ret = None
+ identifier = fluxio_data.data_identifier
+ if identifier == FluxioDataIdentifier.STRING:
+ ret = AronDataIceTypes.String(value=fluxio_data.value)
+ elif identifier == FluxioDataIdentifier.BOOL:
+ ret = AronDataIceTypes.Bool(value=fluxio_data.value)
+ elif identifier == FluxioDataIdentifier.INT:
+ ret = AronDataIceTypes.Int(value=fluxio_data.value)
+ elif identifier == FluxioDataIdentifier.LONG:
+ ret = AronDataIceTypes.Long(value=fluxio_data.value)
+ elif identifier == FluxioDataIdentifier.FLOAT:
+ ret = AronDataIceTypes.Float(value=fluxio_data.value)
+ elif identifier == FluxioDataIdentifier.DOUBLE:
+ ret = AronDataIceTypes.Double(value=fluxio_data.value)
+ elif identifier == FluxioDataIdentifier.LIST:
+ elements = [cls.fluxio_data_to_aron(e) for e in fluxio_data.elements]
+ ret = AronDataIceTypes.List(elements=elements)
+ elif identifier == FluxioDataIdentifier.DICT:
+ elements = {
+ e.key: cls.fluxio_data_to_aron(e.value) for e in fluxio_data.elements
+ }
+ ret = AronDataIceTypes.Dict(elements=elements)
+ elif identifier == FluxioDataIdentifier.NDARRAY:
+ numpy_ndarray = np.ndarray(
+ shape=fluxio_data.shape,
+ dtype=fluxio_data.base.dtype,
+ buffer=fluxio_data.base,
+ )
+ ret = ndarray_to_aron(numpy_ndarray)
+ return ret
diff --git a/armarx_skills/manager/fluxio_edge.py b/armarx_skills/manager/fluxio_edge.py
new file mode 100644
index 0000000000000000000000000000000000000000..7896eb8d74c616e396dac5d2f5bbb60217577b7d
--- /dev/null
+++ b/armarx_skills/manager/fluxio_edge.py
@@ -0,0 +1,115 @@
+import dataclasses as dc
+
+from armarx_skills.manager import dto
+from armarx_skills.manager.fluxio_identificator import (
+ Identificator,
+ FluxioIdentificatorConv,
+)
+from armarx_memory.ice_conv.ice_converter import IceConverter
+
+
+@dc.dataclass
+class Edge:
+ """
+ A class representing an edge for a Fluxio composite skill.
+
+ Attributes:
+ - fromNodeId: Identificator - the ID of the node where the edge starts
+ - fromParameterId: Identificator - the ID of the parameter at the starting node
+ - toNodeId: Identificator - the ID of the node where the edge ends
+ - toParameterId: Identificator - the ID of the parameter at the ending node
+ """
+
+ fromNodeId: Identificator
+ fromParameterId: Identificator
+ toNodeId: Identificator
+ toParameterId: Identificator
+
+
+class FluxioEdgeConv(IceConverter):
+ """
+ A class representing a Fluxio Edge Converter.
+
+ This class inherits from IceConverter and is used to convert data from Fluxio Edge devices.
+
+ Attributes:
+ - None
+
+ Methods:
+ - __init__(): Initializes the FluxioEdgeConv object.
+ - _import_dto(): Imports the DTO class for the converter.
+ - _from_ice(): Converts the DTO to the Edge object.
+ - _to_ice(): Converts the Edge object to the DTO.
+
+ """
+
+ def __init__(self):
+ """
+ Initializes the FluxioEdgeConv object.
+
+ Parameters:
+ - None
+
+ Return:
+ - None
+ """
+ super().__init__()
+
+ @classmethod
+ def _import_dto(cls):
+ """
+ This function imports and returns the DTO (Data Transfer Object) class for FluxioEdge.
+
+ Parameters:
+ cls (class): The class for which the DTO is being imported.
+
+ Return:
+ class: The DTO class for FluxioEdge.
+
+ Raises:
+ None
+ """
+
+ return dto.FluxioEdge
+
+ def _from_ice(self, dt: dto.FluxioEdge) -> Edge:
+ """
+ Converts a dto.FluxioEdge object to an Edge object.
+
+ Parameters:
+ dt (dto.FluxioEdge): The dto.FluxioEdge object to convert.
+
+ Return:
+ Edge: The converted Edge object.
+
+ Raises:
+ None
+ """
+ return Edge(
+ fromNodeId=FluxioIdentificatorConv().from_ice(dt.fromNodeId),
+ fromParameterId=FluxioIdentificatorConv().from_ice(dt.fromParameterId),
+ toNodeId=FluxioIdentificatorConv().from_ice(dt.toNodeId),
+ toParameterId=FluxioIdentificatorConv().from_ice(dt.toParameterId),
+ )
+
+ def _to_ice(self, bo: Edge) -> dto.FluxioEdge:
+ """
+ Converts a business object Edge to a data transfer object FluxioEdge.
+
+ Parameters:
+ bo (Edge): The business object Edge to be converted.
+
+ Return:
+ dto.FluxioEdge: The converted data transfer object FluxioEdge.
+
+ Raises:
+ None
+ """
+
+ return dto.FluxioEdge(
+ fromNodeId=FluxioIdentificatorConv().to_ice(bo.fromNodeId),
+ fromParameterId=FluxioIdentificatorConv().to_ice(bo.fromParameterId),
+ toNodeId=FluxioIdentificatorConv().to_ice(bo.toNodeId),
+ toParameterId=FluxioIdentificatorConv().to_ice(bo.toParameterId),
+ )
+
diff --git a/armarx_skills/manager/fluxio_exception.py b/armarx_skills/manager/fluxio_exception.py
new file mode 100644
index 0000000000000000000000000000000000000000..a879ce489e26fb88a7716cab879b949976bdfa7b
--- /dev/null
+++ b/armarx_skills/manager/fluxio_exception.py
@@ -0,0 +1,73 @@
+import dataclasses as dc
+from enum import Enum
+from typing import List
+
+from armarx_skills.manager import dto
+from armarx_memory.ice_conv.ice_converter import IceConverter
+
+
+class FluxioExceptionType(str, Enum):
+ BAD_REQUEST = "BAD_REQUEST"
+ UNAUTHORIZED = "UNAUTHORIZED"
+ FORBIDDEN = "FORBIDDEN"
+ NOT_FOUND = "NOT_FOUND"
+ UNKNOWN = "UNKNOWN"
+
+
+@dc.dataclass
+class FluxioException(BaseException):
+ """
+ A class representing a Fluxio Exception.
+
+ Attributes:
+ message (str): The reason for the exception
+ type (FluxioExceptionType): The type of the exception
+ contextList (List[str]): A list of context strings
+ """
+
+ message: str
+ type: FluxioExceptionType
+ contextList: List[str] = dc.field(default_factory=list)
+
+
+class FluxioExceptionConv(IceConverter):
+ """
+ A class representing a FluxioProviderConv object, which is a subclass of IceConverter.
+
+ Attributes:
+ - None
+
+ Methods:
+
+ """
+
+ def __init__(self):
+ """
+ Initializes a new FluxioProviderConv object.
+
+ Parameters:
+ - None
+
+ Return:
+ - None
+ """
+ super().__init__()
+
+ def _from_ice(self, dt: dto.FluxioException):
+ return FluxioException(dt.message, FluxioExceptionType(dt.type), dt.contextList)
+
+ def _to_ice(self, bo: FluxioException):
+ return dto.FluxioException(bo.reason, bo.type, bo.contextList)
+
+ @classmethod
+ def _import_dto(cls):
+ """
+ This method is a class method that returns the FluxioIdentificator data transfer object (DTO).
+
+ Return:
+ dto.FluxioIdentificator: The FluxioIdentificator data transfer object.
+
+ Raises:
+ None
+ """
+ return dto.FluxioException
diff --git a/armarx_skills/manager/fluxio_identificator.py b/armarx_skills/manager/fluxio_identificator.py
new file mode 100644
index 0000000000000000000000000000000000000000..bacd280ed2765227f080e190155f927a573b2d79
--- /dev/null
+++ b/armarx_skills/manager/fluxio_identificator.py
@@ -0,0 +1,97 @@
+import dataclasses as dc
+
+from armarx_skills.manager import dto
+from armarx_memory.ice_conv.ice_converter import IceConverter
+
+
+@dc.dataclass
+class Identificator:
+ """
+ A class representing an Identificator object.
+
+ Attributes:
+ id (str): The unique identifier for the object.
+ hint (str): A hint or description for the object.
+
+ Methods:
+ None
+
+ Example:
+ >>> ident = Identificator(id="12345", hint="This is a sample identificator")
+ """
+
+ id: str
+ hint: str
+
+
+class FluxioIdentificatorConv(IceConverter):
+ """
+ A class for converting identifiers in the Fluxio system.
+
+ This class inherits from IceConverter and provides methods for converting identifiers used in the Fluxio system.
+
+ Attributes:
+ - None
+
+ Methods:
+ - __init__: Initializes the FluxioIdentificatorConv object.
+ - _import_dto: Imports the DTO class for the converter.
+ - _from_ice: Converts the DTO to the Identificator object.
+ - _to_ice: Converts the Identificator object to the DTO.
+
+ """
+
+ def __init__(self):
+ """
+ Initializes the FluxioIdentificatorConv object.
+
+ Parameters:
+ - None
+
+ Return:
+ - None
+ """
+ super().__init__()
+
+ @classmethod
+ def _import_dto(cls):
+ """
+ This method is a class method that returns the FluxioIdentificator data transfer object (DTO).
+
+ Return:
+ dto.FluxioIdentificator: The FluxioIdentificator data transfer object.
+
+ Raises:
+ None
+ """
+ return dto.FluxioIdentificator
+
+ def _from_ice(self, dt: dto.FluxioIdentificator) -> Identificator:
+ """
+ Converts a FluxioIdentificator object to an Identificator object.
+
+ Parameters:
+ dt (dto.FluxioIdentificator): The FluxioIdentificator object to convert.
+
+ Return:
+ Identificator: The converted Identificator object.
+
+ Raises:
+ None
+ """
+ return Identificator(id=dt.id, hint=dt.hint)
+
+ def _to_ice(self, bo: Identificator) -> dto.FluxioIdentificator:
+ """
+ Converts an Identificator object to a FluxioIdentificator object.
+
+ Parameters:
+ bo (Identificator): The Identificator object to be converted.
+
+ Return:
+ dto.FluxioIdentificator: The converted FluxioIdentificator object.
+
+ Raises:
+ None
+ """
+ return dto.FluxioIdentificator(id=bo.id, hint=bo.hint)
diff --git a/armarx_skills/manager/fluxio_node.py b/armarx_skills/manager/fluxio_node.py
new file mode 100644
index 0000000000000000000000000000000000000000..a28870080e82f6e24e69ac112e6c1e399dd6f5cc
--- /dev/null
+++ b/armarx_skills/manager/fluxio_node.py
@@ -0,0 +1,109 @@
+import dataclasses as dc
+from enum import Enum
+
+from armarx_skills.manager import dto
+from armarx_memory.ice_conv.ice_converter import IceConverter
+
+
+class NodeTypes(str, Enum):
+ SUBSKILL = "SUBSKILL"
+ PARAMETER = "PARAMETER"
+ CONTROL = "CONTROL"
+
+
+@dc.dataclass
+class Node:
+ """
+ A class representing a node in a Fluxio compositer skill.
+
+ Attributes:
+ - nodeType (NodeTypes): The type of the node.
+ - nodeId (str): The unique identifier of the node.
+ - xPos (float): The x-coordinate of the node.
+ - yPos (float): The y-coordinate of the node.
+ """
+
+ nodeType: NodeTypes
+ nodeId: str
+ xPos: float
+ yPos: float
+
+
+class FluxioNodeConv(IceConverter):
+ """
+ A class representing a Fluxio Node Converter.
+
+ This class extends the IceConverter class and is used to convert data from Fluxio nodes.
+
+ Attributes:
+ - None
+
+ Methods:
+ - __init__: Initializes the FluxioNodeConv object.
+ - _import_dto: Imports the DTO class for the converter.
+ - _from_ice: Converts the DTO to the Node object.
+ - _to_ice: Converts the Node object to the DTO.
+
+ """
+
+ def __init__(self):
+ """
+ Initializes the FluxioNodeConv object.
+
+ Parameters:
+ - None
+
+ Return:
+ - None
+ """
+ super().__init__()
+
+ @classmethod
+ def _import_dto(cls):
+ """
+ This method is a class method that returns the FluxioNode data transfer object (DTO).
+
+ Return:
+ dto.FluxioNode: The FluxioNode data transfer object.
+
+ Raises:
+ None
+ """
+ return dto.FluxioNode
+
+ def _from_ice(self, dt: dto.FluxioNode) -> Node:
+ """
+ Converts a FluxioNode object to a Node object.
+
+ Parameters:
+ dt (dto.FluxioNode): The FluxioNode object to convert.
+
+ Return:
+ Node: The converted Node object.
+
+ Raises:
+ None
+ """
+ return Node(
+ nodeType=NodeTypes(dt.nodeType),
+ nodeId=dt.nodeId,
+ xPos=dt.xPos,
+ yPos=dt.yPos,
+ )
+
+ def _to_ice(self, bo: Node) -> dto.FluxioNode:
+ """
+ Converts a Node object to a FluxioNode object.
+
+ Parameters:
+ bo (Node): The Node object to be converted.
+
+ Return:
+ dto.FluxioNode: The converted FluxioNode object.
+
+ Raises:
+ None
+ """
+ return dto.FluxioNode(
+ nodeType=bo.nodeType, nodeId=bo.nodeId, xPos=bo.xPos, yPos=bo.yPos
+ )
diff --git a/armarx_skills/manager/fluxio_parameter.py b/armarx_skills/manager/fluxio_parameter.py
new file mode 100644
index 0000000000000000000000000000000000000000..efa54fb11f4da41602f52ead0e68a8d430a40982
--- /dev/null
+++ b/armarx_skills/manager/fluxio_parameter.py
@@ -0,0 +1,121 @@
+import dataclasses as dc
+from typing import List
+
+from armarx_skills.manager import dto
+from armarx_skills.manager.fluxio_value import Value, FluxioValueConv
+from armarx_skills.manager.fluxio_type_identificator import TypeIdentificator, FluxioTypeIdentificatorConv
+from armarx_memory.ice_conv.ice_converter import IceConverter
+
+
+@dc.dataclass
+class Parameter:
+ """
+ A class representing a parameter of a Fluxio skill.
+
+ Attributes:
+ - id (str): The unique identifier for the parameter.
+ - name (str): The name of the parameter.
+ - description (str): A description of the parameter.
+ - type (TypeIdentificator): The type of the parameter.
+ - required (bool): Indicates whether the parameter is required.
+ - isInput (bool): Indicates whether the parameter is an input parameter.
+ - values (List[Value]): A list of possible values for the parameter.
+ """
+
+ id: str
+ name: str
+ description: str
+ # type: str # TODO: Update to a more specific type.
+ type: TypeIdentificator
+ required: bool
+ isInput: bool
+ values: List[Value]
+
+
+class FluxioParameterConv(IceConverter):
+ """
+ A class for converting parameters in Fluxio format to Ice format.
+
+ This class inherits from IceConverter and provides methods for converting parameters in Fluxio format to Ice format.
+
+ Attributes:
+ - None
+
+ Methods:
+ - __init__(): Initializes the FluxioParameterConv class.
+ - _import_dto(): Imports the DTO class for the converter.
+ - _from_ice(): Converts the DTO to the Parameter object.
+ - _to_ice(): Converts the Parameter object to the DTO.
+
+ """
+
+ def __init__(self):
+ """
+ Initializes the FluxioParameterConv class.
+
+ Parameters:
+ - None
+
+ Return:
+ - None
+ """
+ super().__init__()
+
+ @classmethod
+ def _import_dto(cls):
+ """
+ This method is a class method that returns the FluxioParameter data transfer object (DTO).
+
+ Return:
+ dto.FluxioParameter: The FluxioParameter data transfer object.
+
+ Raises:
+ None
+ """
+ return dto.FluxioParameter
+
+ def _from_ice(self, dt: dto.FluxioParameter) -> Parameter:
+ """
+ Converts a FluxioParameter object to a Parameter object.
+
+ Parameters:
+ dt (dto.FluxioParameter): The FluxioParameter object to convert.
+
+ Return:
+ Parameter: The converted Parameter object.
+
+ Raises:
+ None
+ """
+ return Parameter(
+ id=dt.id,
+ name=dt.name,
+ description=dt.description,
+ type=FluxioTypeIdentificatorConv().from_ice(dt.type),
+ required=dt.required,
+ isInput=dt.isInput,
+ values=[FluxioValueConv().from_ice(v) for v in dt.values],
+ )
+
+ def _to_ice(self, bo: Parameter) -> dto.FluxioParameter:
+ """
+ Converts a Parameter object to a FluxioParameter object.
+
+ Parameters:
+ bo (Parameter): The Parameter object to be converted.
+
+ Return:
+ dto.FluxioParameter: The converted FluxioParameter object.
+
+ Raises:
+ None
+ """
+ return dto.FluxioParameter(
+ id=bo.id,
+ name=bo.name,
+ description=bo.description,
+ type=FluxioTypeIdentificatorConv().to_ice(bo.type),
+ required=bo.required,
+ isInput=bo.isInput,
+ values=[FluxioValueConv().to_ice(v) for v in bo.values],
+ )
diff --git a/armarx_skills/manager/fluxio_parameter_node.py b/armarx_skills/manager/fluxio_parameter_node.py
new file mode 100644
index 0000000000000000000000000000000000000000..e8e8d2ccaf72b1309379a080d89b4891a54c9878
--- /dev/null
+++ b/armarx_skills/manager/fluxio_parameter_node.py
@@ -0,0 +1,130 @@
+import dataclasses as dc
+
+from armarx_skills.manager import dto
+from armarx_memory.ice_conv.ice_converter import IceConverter
+from armarx_skills.manager.fluxio_node import Node, NodeTypes
+from armarx_skills.manager.fluxio_identificator import (
+ Identificator,
+ FluxioIdentificatorConv,
+)
+
+
+@dc.dataclass
+class ParameterNode(Node):
+ """
+ A class representing a Parameter Node in a Fluxio compositer skill.
+
+ Attributes:
+ - nodeType (str): The type of the node.
+ - nodeId (str): The unique identifier of the node.
+ - xPos (float): The x-coordinate of the node.
+ - yPos (float): The y-coordinate of the node.
+ - parameterId (Identificator): The identifier of the parameter associated with the node.
+ """
+
+ parameterId: Identificator
+
+ def __init__(
+ self,
+ nodeType: NodeTypes,
+ nodeId: str,
+ xPos: float,
+ yPos: float,
+ parameterId: Identificator,
+ ) -> None:
+ """
+ Initialize a ParameterNode object.
+
+ Parameters:
+ - nodeType (str): The type of the node.
+ - nodeId (str): The unique identifier of the node.
+ - xPos (float): The x-coordinate of the node.
+ - yPos (float): The y-coordinate of the node.
+ - parameterId (Identificator): The identifier of the parameter associated with the node.
+ """
+ super().__init__(nodeType, nodeId, xPos, yPos)
+ self.parameterId = parameterId
+
+
+class FluxioParameterNodeConv(IceConverter):
+ """
+ A class representing a Fluxio Parameter Node Converter.
+
+ This class inherits from IceConverter and is used to convert Fluxio Parameter Nodes.
+
+ Attributes:
+ - None
+
+ Methods:
+ - __init__: Initializes the FluxioParameterNodeConv object.
+ - _import_dto: Imports the DTO class for the converter.
+ - _from_ice: Converts the DTO to the ParameterNode object.
+ - _to_ice: Converts the ParameterNode object to the DTO.
+ """
+
+ def __init__(self):
+ """
+ Initializes the FluxioParameterNodeConv object.
+
+ Parameters:
+ - None
+
+ Return:
+ - None
+ """
+ super().__init__()
+
+ @classmethod
+ def _import_dto(cls):
+ """
+ This method is a class method that returns the FluxioNode data transfer object (DTO).
+
+ Return:
+ dto.FluxioNode: The FluxioNode data transfer object.
+
+ Raises:
+ None
+ """
+ return dto.FluxioNode
+
+ def _from_ice(self, dt: dto.FluxioNode) -> ParameterNode:
+ """
+ Converts a dto.FluxioNode object to a ParameterNode object.
+
+ Parameters:
+ dt (dto.FluxioNode): The dto.FluxioNode object to convert.
+
+ Returns:
+ ParameterNode: The converted ParameterNode object.
+
+ Raises:
+ None
+ """
+ return ParameterNode(
+ nodeType=NodeTypes(dt.nodeType),
+ nodeId=dt.nodeId,
+ xPos=dt.xPos,
+ yPos=dt.yPos,
+ parameterId=FluxioIdentificatorConv().from_ice(dt.parameterId),
+ )
+
+ def _to_ice(self, bo: ParameterNode) -> dto.FluxioNode:
+ """
+ Converts a ParameterNode object to a FluxioNode object.
+
+ Parameters:
+ bo (ParameterNode): The ParameterNode object to be converted.
+
+ Returns:
+ dto.FluxioNode: The converted FluxioNode object.
+
+ Raises:
+ None
+ """
+ return dto.FluxioNode(
+ nodeType=bo.nodeType,
+ nodeId=bo.nodeId,
+ xPos=bo.xPos,
+ yPos=bo.yPos,
+ parameterId=FluxioIdentificatorConv().to_ice(bo.parameterId),
+ )
diff --git a/armarx_skills/manager/fluxio_profile.py b/armarx_skills/manager/fluxio_profile.py
new file mode 100644
index 0000000000000000000000000000000000000000..d1d20710c099071009541d40e0e0ae37fa0b8d49
--- /dev/null
+++ b/armarx_skills/manager/fluxio_profile.py
@@ -0,0 +1,118 @@
+import dataclasses as dc
+
+from armarx_skills.manager import dto
+from armarx_skills.manager.fluxio_identificator import (
+ Identificator,
+ FluxioIdentificatorConv,
+)
+from armarx_memory.ice_conv.ice_converter import IceConverter
+
+
+@dc.dataclass
+class Profile:
+ """
+ A class representing a settings profile for skill parameter values.
+
+ Attributes:
+ - id (str): The unique identifier for the profile.
+ - name (str): The name of the user.
+ - description (str): A description of the user.
+ - parentId (Identificator | None): The identifier of the parent profile, if any.
+ """
+
+ id: str
+ name: str
+ description: str
+ parentId: Identificator | None = None
+
+
+class FluxioProfileConv(IceConverter):
+ """
+ A class for converting Fluxio profiles.
+
+ This class extends IceConverter and provides methods for converting Fluxio profiles.
+
+ Attributes:
+ - fluxio_identificator_conv: An instance of FluxioIdentificatorConv used for converting Fluxio identifiers.
+
+ Methods:
+ - __init__: Initializes the FluxioProfileConv class and sets up the fluxio_identificator_conv attribute.
+ - _import_dto: Imports the DTO class for the converter.
+ - _from_ice: Converts the DTO to the Profile object.
+ - _to_ice: Converts the Profile object to the DTO.
+ """
+
+ def __init__(self):
+ """
+ Initializes the FluxioProfileConv class.
+
+ Parameters:
+ - None
+
+ Return:
+ - None
+ """
+ super().__init__()
+ self.fluxio_identificator_conv = FluxioIdentificatorConv()
+
+ @classmethod
+ def _import_dto(cls):
+ """
+ This method is a class method that returns the FluxioProfile data transfer object (DTO) class.
+
+ Return:
+ dto.FluxioProfile: The FluxioProfile data transfer object class.
+
+ Raises:
+ None
+ """
+ return dto.FluxioProfile
+
+ def _from_ice(self, dt: dto.FluxioProfile) -> Profile:
+ """
+ Converts a dto.FluxioProfile object to a Profile object.
+
+ Parameters:
+ dt (dto.FluxioProfile): The dto.FluxioProfile object to be converted.
+
+ Return:
+ Profile: The converted Profile object.
+
+ Raises:
+ None
+ """
+ return Profile(
+ id=dt.id,
+ name=dt.name,
+ description=dt.description,
+ parentId=None
+ if (not dt.parentIdHasValue)
+ else self.fluxio_identificator_conv.from_ice(dt.parentId),
+ )
+
+ def _to_ice(self, bo: Profile) -> dto.FluxioProfile:
+ """
+ Converts a Profile object to a FluxioProfile object.
+
+ Parameters:
+ bo (Profile): The Profile object to be converted.
+
+ Return:
+ dto.FluxioProfile: The converted FluxioProfile object.
+
+ Raises:
+ None
+
+ If the parentId attribute of the Profile object is None, parentIdHasValue is set to False and parentId is set to an empty dictionary.
+ If the parentId attribute is not None, parentIdHasValue is set to True and parentId is converted using the fluxio_identificator_conv.to_ice method.
+
+ """
+ return dto.FluxioProfile(
+ id=bo.id,
+ name=bo.name,
+ description=bo.description,
+ parentIdHasValue=False if (bo.parentId is None) else True,
+ parentId={"id": "", "hint": ""}
+ if (bo.parentId is None)
+ else self.fluxio_identificator_conv.to_ice(bo.parentId),
+ )
diff --git a/armarx_skills/manager/fluxio_provider.py b/armarx_skills/manager/fluxio_provider.py
new file mode 100644
index 0000000000000000000000000000000000000000..b0db087ed8e98a06fdfa6dab24f77b3839b3f844
--- /dev/null
+++ b/armarx_skills/manager/fluxio_provider.py
@@ -0,0 +1,89 @@
+import dataclasses as dc
+
+from armarx_skills.manager import dto
+from armarx_memory.ice_conv.ice_converter import IceConverter
+
+
+@dc.dataclass
+class Provider:
+ """
+ A class representing a provider for skills.
+
+ Attributes:
+ id (str): The unique identifier for the provider.
+ name (str): The name of the provider.
+ """
+
+ id: str
+ name: str
+
+
+class FluxioProviderConv(IceConverter):
+ """
+ A class representing a FluxioProviderConv object, which is a subclass of IceConverter.
+
+ Attributes:
+ - None
+
+ Methods:
+ - __init__: Initializes a new FluxioProviderConv object.
+ - _import_dto: Imports the DTO class for the converter.
+ - _from_ice: Converts the DTO to the Provider object.
+ - _to_ice: Converts the Provider object to the DTO.
+ """
+
+ def __init__(self):
+ """
+ Initializes a new FluxioProviderConv object.
+
+ Parameters:
+ - None
+
+ Return:
+ - None
+ """
+ super().__init__()
+
+ @classmethod
+ def _import_dto(cls):
+ """
+ This method is a class method that imports and returns the FluxioProvider data transfer object (DTO).
+
+ Return:
+ dto.FluxioProvider: The FluxioProvider data transfer object.
+
+ Raises:
+ None
+ """
+ return dto.FluxioProvider
+
+ def _from_ice(self, dt: dto.FluxioProvider) -> Provider:
+ """
+ Converts a data transfer object (DTO) representing a FluxioProvider to a Provider object.
+
+ Parameters:
+ dt (dto.FluxioProvider): The data transfer object representing a FluxioProvider.
+
+ Return:
+ Provider: A Provider object created from the data transfer object.
+
+ Raises:
+ None
+ """
+ return Provider(id=dt.id, name=dt.name)
+
+ def _to_ice(self, bo: Provider) -> dto.FluxioProvider:
+ """
+ Converts a Provider business object to a FluxioProvider data transfer object.
+
+ Parameters:
+ bo (Provider): The Provider business object to convert.
+
+ Return:
+ dto.FluxioProvider: The converted FluxioProvider data transfer object.
+
+ Raises:
+ None
+ """
+ return dto.FluxioProvider(id=bo.id, name=bo.name)
+
diff --git a/armarx_skills/manager/fluxio_skill.py b/armarx_skills/manager/fluxio_skill.py
new file mode 100644
index 0000000000000000000000000000000000000000..5d0c0e5eb081f31a2a72ebf9890104db1e278c15
--- /dev/null
+++ b/armarx_skills/manager/fluxio_skill.py
@@ -0,0 +1,182 @@
+import dataclasses as dc
+from typing import List
+
+from armarx_skills.manager import dto
+from armarx_skills.manager.fluxio_identificator import (
+ Identificator,
+ FluxioIdentificatorConv,
+)
+from armarx_skills.manager.fluxio_parameter import Parameter, FluxioParameterConv
+from armarx_skills.manager.fluxio_node import Node, FluxioNodeConv, NodeTypes
+from armarx_skills.manager.fluxio_parameter_node import (
+ ParameterNode,
+ FluxioParameterNodeConv,
+)
+from armarx_skills.manager.fluxio_subskill_node import (
+ SubSkillNode,
+ FluxioSubSkillNodeConv,
+)
+from armarx_skills.manager.fluxio_control_node import ControlNode, FluxioControlNodeConv
+from armarx_skills.manager.fluxio_edge import Edge, FluxioEdgeConv
+from armarx_memory.ice_conv.ice_converter import IceConverter
+
+from armarx_core.time.duration import Duration, DurationIceConv
+
+
+@dc.dataclass
+class Skill:
+ """
+ Represents a Fluxio skill entity.
+
+ Attributes:
+ - id (str): The unique identifier of the skill.
+ - name (str): The name of the skill.
+ - description (str): A brief description of the skill.
+ - timeout (Duration): The timeout of the skill.
+ - lastChanged (str): The timestamp when the skill was last modified.
+ - executable (bool): Indicates if the skill is executable.
+ - native (bool): Indicates if the skill is native and not composite.
+ - skillProviderId (Identificator): The identifier of the skill provider.
+ - parameters (List[Parameter]): A list of parameters required by the skill.
+ - nodes (List[Node], optional): A list of nodes associated with the skill.
+ - edges (List[Edge], optional): A list of edges associated with the skill.
+ """
+
+ id: str
+ name: str
+ description: str
+ timeout: int
+ lastChanged: str
+ executable: bool
+ native: bool
+ skillProviderId: Identificator
+ parameters: List[Parameter]
+ nodes: List[Node] | None = None
+ edges: List[Edge] | None = None
+
+
+class FluxioSkillConv(IceConverter):
+ """
+ A class for converting Fluxio skills to Ice skills.
+
+ This class inherits from IceConverter and provides methods for converting Fluxio skills to Ice skills.
+
+ Attributes:
+ - None
+
+ Methods:
+ - __init__: Initializes the FluxioSkillConv object.
+ - _import_dto: Imports the DTO class for the converter.
+ - _from_ice: Converts the DTO to the Skill object.
+ - _to_ice: Converts the Skill object to the DTO.
+
+ """
+
+ def __init__(self):
+ """
+ Initializes the FluxioSkillConv object.
+
+ Parameters:
+ - None
+
+ Return:
+ - None
+ """
+ super().__init__()
+ self.duration_conv = DurationIceConv()
+
+ @classmethod
+ def _import_dto(cls):
+ """
+ This method is a class method that returns the FluxioSkill data transfer object (DTO).
+
+ Return:
+ dto.FluxioSkill: The FluxioSkill data transfer object.
+
+ Raises:
+ None
+ """
+ return dto.FluxioSkill
+
+ def _from_ice(self, dt: dto.FluxioSkill) -> Skill:
+ """
+ Converts a dto.FluxioSkill object to a Skill object.
+
+ Parameters:
+ dt (dto.FluxioSkill): The dto.FluxioSkill object to be converted.
+
+ Return:
+ Skill: The converted Skill object.
+
+ Raises:
+ None
+ """
+ ret = Skill(
+ id=dt.id,
+ name=dt.name,
+ description=dt.description,
+ timeout=self.duration_conv.from_ice(dt.timeout).toMilliSeconds(),
+ lastChanged=dt.lastChanged,
+ executable=dt.executable,
+ native=dt.native,
+ skillProviderId=FluxioIdentificatorConv().from_ice(dt.skillProviderId),
+ parameters=[FluxioParameterConv().from_ice(p) for p in dt.parameters],
+ nodes=[],
+ edges=[FluxioEdgeConv().from_ice(e) for e in dt.edges]
+ if dt.edgesHasValue is True
+ else [],
+ )
+
+ if dt.nodesHasValue is True:
+ for n in dt.nodes:
+ if n.nodeType == NodeTypes.PARAMETER:
+ ret.nodes.append(FluxioParameterNodeConv().from_ice(n))
+ elif n.nodeType == NodeTypes.SUBSKILL:
+ ret.nodes.append(FluxioSubSkillNodeConv().from_ice(n))
+ elif n.nodeType == NodeTypes.CONTROL:
+ ret.nodes.append(FluxioControlNodeConv().from_ice(n))
+ else:
+ ret.nodes.append(FluxioNodeConv().from_ice(n))
+ return ret
+
+ def _to_ice(self, bo: Skill) -> dto.FluxioSkill:
+ """
+ Converts a Skill business object to a FluxioSkill data transfer object.
+
+ Parameters:
+ bo (Skill): The Skill business object to be converted.
+
+ Return:
+ dto.FluxioSkill: The converted FluxioSkill data transfer object.
+
+ Raises:
+ None
+ """
+ ret = dto.FluxioSkill(
+ id=bo.id,
+ name=bo.name,
+ description=bo.description,
+ timeout=self.duration_conv.to_ice(Duration(microSeconds=bo.timeout * 1000)),
+ lastChanged=bo.lastChanged,
+ executable=bo.executable,
+ native=bo.native,
+ skillProviderId=FluxioIdentificatorConv().to_ice(bo.skillProviderId),
+ parameters=[FluxioParameterConv().to_ice(p) for p in bo.parameters],
+ nodesHasValue=False if (bo.nodes is None) else True,
+ nodes=[],
+ edgesHasValue=False if (bo.edges is None) else True,
+ edges=[]
+ if (bo.edges is None)
+ else [FluxioEdgeConv().to_ice(e) for e in bo.edges],
+ )
+ if bo.nodes is not None:
+ for n in bo.nodes:
+ if isinstance(n, ParameterNode):
+ ret.nodes.append(FluxioParameterNodeConv().to_ice(n))
+ elif isinstance(n, SubSkillNode):
+ ret.nodes.append(FluxioSubSkillNodeConv().to_ice(n))
+ elif isinstance(n, ControlNode):
+ ret.nodes.append(FluxioControlNodeConv().to_ice(n))
+ else:
+ ret.nodes.append(FluxioNodeConv().to_ice(n))
+ return ret
diff --git a/armarx_skills/manager/fluxio_skill_status_update.py b/armarx_skills/manager/fluxio_skill_status_update.py
new file mode 100644
index 0000000000000000000000000000000000000000..d4608bebbad5ce4d739251474f25d8b30f556b5b
--- /dev/null
+++ b/armarx_skills/manager/fluxio_skill_status_update.py
@@ -0,0 +1,103 @@
+import dataclasses as dc
+
+from armarx_skills.manager import dto
+from armarx_core.time.date_time import DateTimeIceConverter
+from armarx_memory.ice_conv.ice_converter import IceConverter
+from armarx_skills.core.execution_status import ExecutionStatus, ExecutionStatusConv
+from armarx_skills.manager.fluxio_identificator import (
+ Identificator,
+ FluxioIdentificatorConv,
+)
+
+
+@dc.dataclass
+class SkillStatusUpdate:
+ """
+ A class representing an update to the status of a skill execution.
+
+ Attributes:
+ timestamp (int): The timestamp of the status update. Generated by armarx::DateTime::Now().
+ executionId (str): The unique identifier of the skill execution.
+ subSkillNodeId (Identificator): The identifier of the sub-skill node. If composite skill then the id is the skill id itself.
+ status (ExecutionStatus): The status of the skill execution.
+ """
+
+ timestamp: int
+ executionId: str
+ subSkillNodeId: Identificator
+ status: ExecutionStatus
+
+
+class FluxioSkillStatusUpdateConv(IceConverter):
+ """
+ A class for converting Fluxio skill status updates.
+
+ This class inherits from IceConverter and provides methods for converting Fluxio skill status updates.
+
+ Attributes:
+ - date_time_conv (DateTimeIceConverter): A converter for date and time values.
+ - status_conv (ExecutionStatusConv): A converter for execution status values.
+ - fluxio_identificator_conv (FluxioIdentificatorConv): A converter for Fluxio identificator values.
+ """
+
+ def __init__(self):
+ """
+ Initializes the FluxioSkillStatusUpdateConv object.
+
+ This method initializes the converters for date and time values, execution status values, and Fluxio identificator values.
+ """
+ super().__init__()
+ self.date_time_conv = DateTimeIceConverter()
+ self.status_conv = ExecutionStatusConv()
+ self.fluxio_identificator_conv = FluxioIdentificatorConv()
+
+ @classmethod
+ def _import_dto(cls):
+ """
+ Returns the data transfer object (DTO) class for FluxioSkillStatusUpdate.
+
+ Return:
+ dto.FluxioSkillStatusUpdate: The class representing the data transfer object for FluxioSkillStatusUpdate.
+ """
+ return dto.FluxioSkillStatusUpdate
+
+ def _from_ice(self, dt: dto.FluxioSkillStatusUpdate) -> SkillStatusUpdate:
+ """
+ Converts a dto.FluxioSkillStatusUpdate object to a SkillStatusUpdate object.
+
+ Parameters:
+ dt (dto.FluxioSkillStatusUpdate): The dto object to be converted.
+
+ Return:
+ SkillStatusUpdate: The converted SkillStatusUpdate object.
+
+ Raises:
+ None
+ """
+ return SkillStatusUpdate(
+ timestamp=self.date_time_conv.from_ice(dt.timestamp),
+ executionId=dt.executionId,
+ subSkillNodeId=self.fluxio_identificator_conv.from_ice(dt.subSkillNodeId),
+ status=self.status_conv.from_ice(dt.status),
+ )
+
+ def _to_ice(self, bo: SkillStatusUpdate) -> dto.FluxioSkillStatusUpdate:
+ """
+ Converts a SkillStatusUpdate object to a FluxioSkillStatusUpdate object.
+
+ Parameters:
+ bo (SkillStatusUpdate): The SkillStatusUpdate object to be converted.
+
+ Return:
+ FluxioSkillStatusUpdate: The converted FluxioSkillStatusUpdate object.
+
+ Raises:
+ None
+ """
+
+ return dto.FluxioSkillStatusUpdate(
+ timestamp=self.date_time_conv.to_ice(bo.timestamp),
+ executionId=bo.executionId,
+ subSkillNodeId=self.fluxio_identificator_conv.to_ice(bo.subSkillNodeId),
+ status=self.status_conv.to_ice(bo.status),
+ )
diff --git a/armarx_skills/manager/fluxio_subskill_node.py b/armarx_skills/manager/fluxio_subskill_node.py
new file mode 100644
index 0000000000000000000000000000000000000000..6f27ec60851ed330a839c28311d25f8db1fcf5ff
--- /dev/null
+++ b/armarx_skills/manager/fluxio_subskill_node.py
@@ -0,0 +1,135 @@
+import dataclasses as dc
+
+from armarx_skills.manager import dto
+from armarx_memory.ice_conv.ice_converter import IceConverter
+from armarx_skills.manager.fluxio_node import Node, NodeTypes
+from armarx_skills.manager.fluxio_identificator import (
+ Identificator,
+ FluxioIdentificatorConv,
+)
+
+
+@dc.dataclass
+class SubSkillNode(Node):
+ """
+ A class representing a node in a composite skill that contains a sub-skill.
+
+ Attributes:
+ - skillId: Identificator - The identifier of the sub-skill associated with this node
+
+ """
+
+ skillId: Identificator
+
+ def __init__(
+ self,
+ nodeType: NodeTypes,
+ nodeId: str,
+ xPos: float,
+ yPos: float,
+ skillId: Identificator,
+ ):
+ """
+ Initialize a Node object with the given parameters.
+
+ Parameters:
+ nodeType (NodeTypes): The type of the node.
+ nodeId (str): The unique identifier of the node.
+ xPos (float): The x-coordinate position of the node.
+ yPos (float): The y-coordinate position of the node.
+ skillId (Identificator): The identifier of the skill associated with the node.
+
+ Return:
+ None
+
+ Raises:
+ None
+ """
+ super().__init__(nodeType, nodeId, xPos, yPos)
+ self.skillId = skillId
+
+
+class FluxioSubSkillNodeConv(IceConverter):
+ """
+ A class representing a Fluxio SubSkill Node Converter.
+
+ This class inherits from IceConverter and is used to convert Fluxio SubSkill Nodes.
+
+ Attributes:
+ - None
+
+ Methods:
+ - __init__: Initializes the FluxioSubSkillNodeConv object.
+ - _import_dto: Imports the DTO class for the converter.
+ - _from_ice: Converts the DTO to the SubSkillNode object.
+ - _to_ice: Converts the SubSkillNode object to the DTO.
+
+ """
+
+ def __init__(self):
+ """
+ Initializes the FluxioSubSkillNodeConv object.
+
+ Parameters:
+ - None
+
+ Return:
+ - None
+ """
+ super().__init__()
+
+ @classmethod
+ def _import_dto(cls):
+ """
+ This method is a class method that returns the FluxioNode data transfer object (DTO).
+
+ Return:
+ dto.FluxioNode: The FluxioNode data transfer object.
+
+ Raises:
+ None
+ """
+ return dto.FluxioNode
+
+ def _from_ice(self, dt: dto.FluxioNode) -> SubSkillNode:
+ """
+ Converts a dto.FluxioNode object to a SubSkillNode object.
+
+ Parameters:
+ dt (dto.FluxioNode): The FluxioNode object to convert.
+
+ Return:
+ SubSkillNode: The converted SubSkillNode object.
+
+ Raises:
+ None
+ """
+ return SubSkillNode(
+ nodeType=NodeTypes(dt.nodeType),
+ nodeId=dt.nodeId,
+ xPos=dt.xPos,
+ yPos=dt.yPos,
+ skillId=FluxioIdentificatorConv().from_ice(dt.skillId),
+ )
+
+ def _to_ice(self, bo: SubSkillNode) -> dto.FluxioNode:
+ """
+ Converts a SubSkillNode object to a FluxioNode object.
+
+ Parameters:
+ bo (SubSkillNode): The SubSkillNode object to be converted.
+
+ Returns:
+ dto.FluxioNode: The converted FluxioNode object.
+
+ Raises:
+ None
+ """
+
+ return dto.FluxioNode(
+ nodeType=bo.nodeType,
+ nodeId=bo.nodeId,
+ xPos=bo.xPos,
+ yPos=bo.yPos,
+ skillId=FluxioIdentificatorConv().to_ice(bo.skillId),
+ )
diff --git a/armarx_skills/manager/fluxio_type_aron_converter.py b/armarx_skills/manager/fluxio_type_aron_converter.py
new file mode 100644
index 0000000000000000000000000000000000000000..35b70ffad86fcc2fcf5b67674be286c43baa6930
--- /dev/null
+++ b/armarx_skills/manager/fluxio_type_aron_converter.py
@@ -0,0 +1,268 @@
+import warnings
+
+from armarx_memory.aron.aron_ice_types import AronTypeIceTypes
+
+from .fluxio_types import (
+ FluxioBasicType,
+ FluxioDoubleGenericType,
+ FluxioGenericType,
+ FluxioObjectType,
+ FluxioTypeIdentifier,
+ FluxioEnumType,
+)
+
+
+class FluxioTypeAronConverter:
+ @classmethod
+ def aron_type_to_fluxio_type(
+ cls, aron_type: AronTypeIceTypes.GenericType
+ ) -> (
+ FluxioBasicType | FluxioGenericType | FluxioDoubleGenericType | FluxioObjectType
+ ):
+ ret = None
+ maybe = aron_type.maybe != AronTypeIceTypes._maybe["NONE"]
+ if isinstance(aron_type, AronTypeIceTypes.Int):
+ defaultValue = aron_type.defaultValue
+ ret = FluxioBasicType(
+ name="Integer", description="Integer", default_value=defaultValue
+ )
+ elif isinstance(aron_type, AronTypeIceTypes.Long):
+ defaultValue = aron_type.defaultValue
+ ret = FluxioBasicType(
+ name="Long", description="Long", default_value=defaultValue
+ )
+ elif isinstance(aron_type, AronTypeIceTypes.Float):
+ defaultValue = aron_type.defaultValue
+ ret = FluxioBasicType(
+ name="Float", description="Float", default_value=defaultValue
+ )
+ elif isinstance(aron_type, AronTypeIceTypes.Double):
+ defaultValue = aron_type.defaultValue
+ ret = FluxioBasicType(
+ name="Double", description="Double", default_value=defaultValue
+ )
+ elif isinstance(aron_type, AronTypeIceTypes.Bool):
+ defaultValue = aron_type.defaultValue
+ ret = FluxioBasicType(
+ name="Boolean", description="Boolean", default_value=defaultValue
+ )
+ elif isinstance(aron_type, AronTypeIceTypes.String):
+ defaultValue = aron_type.defaultValue
+ ret = FluxioBasicType(
+ name="String", description="String", default_value=defaultValue
+ )
+ elif isinstance(aron_type, AronTypeIceTypes.List):
+ generic = cls.aron_type_to_fluxio_type(aron_type.acceptedType)
+ ret = FluxioGenericType(
+ name="List",
+ description=f"List with elements of type '{generic.name}'",
+ generic=generic,
+ optional=maybe,
+ max_number_of_elements=0,
+ )
+ elif isinstance(aron_type, AronTypeIceTypes.Quaternion):
+ elementType = str(aron_type.elementType)
+ generic = FluxioBasicType(
+ name=elementType,
+ description=elementType,
+ )
+ description = f"Quaternion with elements of type '{generic.name}'"
+ default = aron_type.defaultValue
+ ret = FluxioGenericType(
+ name="Quaternion",
+ description=description,
+ generic=generic,
+ max_number_of_elements=4,
+ optional=maybe,
+ default_value=default,
+ )
+ elif isinstance(aron_type, AronTypeIceTypes.Dict):
+ generic1 = FluxioBasicType(name="String", description="String")
+ generic2 = cls.aron_type_to_fluxio_type(aron_type.acceptedType)
+ ret = FluxioDoubleGenericType(
+ name="Dict",
+ description="Dictionary with keys of type 'String'"
+ + " and values of type '"
+ + generic2.name
+ + "'",
+ generic1=generic1,
+ generic2=generic2,
+ )
+ elif isinstance(aron_type, AronTypeIceTypes.Object):
+ if aron_type.objectName == "Event":
+ ret = FluxioBasicType(name="Event", description="Event")
+ else:
+ parent = (
+ cls.aron_type_to_fluxio_type(aron_type.parent)
+ if aron_type.parent
+ else None
+ )
+ ret = FluxioObjectType(
+ name=aron_type.objectName,
+ type_identifier=FluxioTypeIdentifier.OBJECT,
+ parent=parent,
+ )
+ for k, v in aron_type.elementTypes.items():
+ child = cls.aron_type_to_fluxio_type(v)
+ ret.add_attribute(name=k, type=child)
+ elif isinstance(aron_type, AronTypeIceTypes.Matrix):
+ rows = aron_type.rows
+ cols = aron_type.cols
+ mat_type = str(aron_type.elementType)
+ description = (
+ f"Matrix with dimensions {rows}x{cols} and elements of type {mat_type}"
+ )
+ generic = FluxioBasicType(name=mat_type, description=mat_type)
+ inner_type = FluxioGenericType(
+ name="row",
+ description="",
+ max_number_of_elements=rows,
+ generic=generic,
+ )
+ ret = FluxioGenericType(
+ name="Matrix",
+ description=description,
+ max_number_of_elements=cols,
+ generic=inner_type,
+ )
+ elif isinstance(aron_type, AronTypeIceTypes.IntEnum):
+ ret = FluxioEnumType(
+ name=aron_type.enumName, description=aron_type.enumName
+ )
+ for k, v in aron_type.acceptedValues.items():
+ ret.add_attribute(name=k, value=v)
+ else:
+ # TODO: this won't work because aron_type can't be converted to a strings
+ message = f"Unknown type: {aron_type}"
+ warnings.warn(message=message, category=RuntimeWarning)
+ raise Exception(message)
+ # return ret
+ ret.optional = maybe
+ return ret
+
+ @classmethod
+ def fluxio_type_to_aron(
+ cls,
+ fluxio_type: (
+ FluxioBasicType
+ | FluxioGenericType
+ | FluxioDoubleGenericType
+ | FluxioObjectType
+ ),
+ ) -> any:
+ ret = None
+ identifier = fluxio_type.type_identifier
+ if identifier == FluxioTypeIdentifier.BASIC:
+ ret = cls.__fluxio_basic_type_to_aron(fluxio_type)
+ elif identifier == FluxioTypeIdentifier.GENERIC:
+ ret = cls.__fluxio_generic_type_to_aron(fluxio_type)
+ elif identifier == FluxioTypeIdentifier.DOUBLE_GENERIC:
+ ret = cls.__fluxio_double_generic_type_to_aron(fluxio_type)
+ elif identifier == FluxioTypeIdentifier.OBJECT:
+ ret = cls.__fluxio_obj_type_to_aron(fluxio_type)
+ elif identifier == FluxioTypeIdentifier.ENUM:
+ ret = cls.__fluxio_enum_type_to_aron(fluxio_type)
+ else:
+ warnings.warn(
+ message="Unknown type identifier: " + identifier,
+ category=RuntimeWarning,
+ )
+ return ret
+ return ret
+
+ @classmethod
+ def __fluxio_basic_type_to_aron(
+ cls, fluxio_type: FluxioBasicType
+ ) -> AronTypeIceTypes.GenericType:
+ ret = None
+ maybe = "OPTIONAL" if fluxio_type.optional else "NONE"
+ if fluxio_type.name == "Integer":
+ ret = AronTypeIceTypes.int(default_value=fluxio_type.default_value)
+ elif fluxio_type.name == "Event":
+ maybe = "OPTIONAL" if fluxio_type.optional else "NONE"
+ ret = AronTypeIceTypes.object(
+ object_name="Event", element_types={}, maybe=maybe,
+ )
+ elif fluxio_type.name == "Long":
+ ret = AronTypeIceTypes.long(default_value=fluxio_type.default_value)
+ elif fluxio_type.name == "Float":
+ ret = AronTypeIceTypes.float(default_value=fluxio_type.default_value)
+ elif fluxio_type.name == "Double":
+ ret = AronTypeIceTypes.double(default_value=fluxio_type.default_value)
+ elif fluxio_type.name == "Boolean":
+ ret = AronTypeIceTypes.bool(default_value=fluxio_type.default_value)
+ elif fluxio_type.name == "String":
+ ret = AronTypeIceTypes.string(default_value=fluxio_type.default_value)
+ else:
+ warnings.warn(
+ message="Unknown basic type: " + fluxio_type.name,
+ category=RuntimeWarning,
+ )
+ return ret
+ ret.maybe = AronTypeIceTypes._maybe[maybe]
+ return ret
+
+ @classmethod
+ def __fluxio_generic_type_to_aron(
+ cls, fluxio_type: FluxioGenericType
+ ) -> AronTypeIceTypes.GenericType:
+ ret = None
+ maybe = "OPTIONAL" if fluxio_type.optional else "NONE"
+ if fluxio_type.name == "List":
+ generic = cls.fluxio_type_to_aron(fluxio_type.generic)
+ ret = AronTypeIceTypes.list(accepted_type=generic, maybe=maybe)
+ elif fluxio_type.name == "Quaternion":
+ ret = AronTypeIceTypes.quaternion(
+ element_type=fluxio_type.generic.name,
+ maybe=maybe,
+ default_value=fluxio_type.default_value or "",
+ )
+ elif fluxio_type.name == "Matrix":
+ rows = fluxio_type.generic.max_number_of_elements
+ cols = fluxio_type.max_number_of_elements
+ mat_type = fluxio_type.generic.generic.name
+ maybe = "OPTIONAL" if fluxio_type.optional else "NONE"
+ ret = AronTypeIceTypes.matrix(
+ rows=rows, cols=cols, element_type=mat_type, maybe=maybe
+ )
+ return ret
+
+ @classmethod
+ def __fluxio_double_generic_type_to_aron(
+ cls,
+ fluxio_type: FluxioDoubleGenericType,
+ ) -> AronTypeIceTypes:
+ ret = None
+ if fluxio_type.name == "Dict":
+ generic2 = cls.fluxio_type_to_aron(fluxio_type.generic2)
+ maybe = "OPTIONAL" if fluxio_type.optional else "NONE"
+ ret = AronTypeIceTypes.dict(accepted_type=generic2, maybe=maybe)
+ return ret
+
+ @classmethod
+ def __fluxio_obj_type_to_aron(
+ cls, fluxio_type: FluxioObjectType
+ ) -> AronTypeIceTypes.GenericType:
+ atr = {}
+ for attribute in fluxio_type.attributes:
+ atr[attribute.name] = cls.fluxio_type_to_aron(attribute.type)
+ maybe = "OPTIONAL" if fluxio_type.optional else "NONE"
+ parent = (
+ cls.fluxio_type_to_aron(fluxio_type.parent) if fluxio_type.parent else None
+ )
+ ret = AronTypeIceTypes.object(
+ object_name=fluxio_type.name, element_types=atr, maybe=maybe, parent=parent
+ )
+ return ret
+
+ @classmethod
+ def __fluxio_enum_type_to_aron(
+ cls, fluxio_type: FluxioEnumType
+ ) -> AronTypeIceTypes.GenericType:
+ maybe = "OPTIONAL" if fluxio_type.optional else "NONE"
+ ret = AronTypeIceTypes.int_enum(name=fluxio_type.name, maybe=maybe)
+ dict = {}
+ for k, v in fluxio_type.attributes.items():
+ dict[k] = v
+ ret.acceptedValues = dict
+ return ret
diff --git a/armarx_skills/manager/fluxio_type_identificator.py b/armarx_skills/manager/fluxio_type_identificator.py
new file mode 100644
index 0000000000000000000000000000000000000000..bd13435b59c0faa5e7101d179104b26a13460de1
--- /dev/null
+++ b/armarx_skills/manager/fluxio_type_identificator.py
@@ -0,0 +1,27 @@
+import dataclasses as dc
+
+from armarx_skills.manager import dto
+from armarx_memory.ice_conv.ice_converter import IceConverter
+
+
+@dc.dataclass
+class TypeIdentificator:
+ skillId: str
+ parameterId: str
+
+
+class FluxioTypeIdentificatorConv(IceConverter):
+ def __init__(self):
+ super().__init__()
+
+ @classmethod
+ def _import_dto(cls):
+ return dto.FluxioTypeIdentificator
+
+ def _from_ice(self, dt: dto.FluxioTypeIdentificator) -> TypeIdentificator:
+ return TypeIdentificator(skillId=dt.skillId, parameterId=dt.parameterId)
+
+ def _to_ice(self, bo: TypeIdentificator) -> dto.FluxioTypeIdentificator:
+ return dto.FluxioTypeIdentificator(
+ skillId=bo.skillId, parameterId=bo.parameterId
+ )
diff --git a/armarx_skills/manager/fluxio_types.py b/armarx_skills/manager/fluxio_types.py
new file mode 100644
index 0000000000000000000000000000000000000000..532d56a85a36351f8390dc2d787c21e5427366ea
--- /dev/null
+++ b/armarx_skills/manager/fluxio_types.py
@@ -0,0 +1,181 @@
+from typing import Optional
+import dataclasses as dc
+from enum import Enum
+
+from armarx_skills.manager.fluxio_type_identificator import TypeIdentificator
+
+
+class AronTypes:
+ INT = "_ARON_INT"
+ LONG = "_ARON_LONG"
+ FLOAT = "_ARON_FLOAT"
+ DOUBLE = "_ARON_DOUBLE"
+ BOOL = "_ARON_BOOL"
+ STRING = "_ARON_STRING"
+ LIST = "_ARON_LIST"
+ QUATERNION = "_ARON_QUATERNION"
+ DICT = "_ARON_DICT"
+ MATRIX = "_ARON_MATRIX"
+ ENUM = "_ARON_INT_ENUM"
+ OBJECT = "_ARON_OBJECT"
+
+
+class AronOptionals:
+ NONE = "type::maybe::NONE"
+ OPTIONAL = "type::maybe::OPTIONAL"
+
+
+class FluxioTypeIdentifier(str, Enum):
+ BASIC = "BASIC"
+ GENERIC = "GENERIC"
+ DOUBLE_GENERIC = "DOUBLE_GENERIC"
+ OBJECT = "OBJECT"
+ ENUM = "ENUM"
+
+
+@dc.dataclass
+class FluxioBasicType:
+ origin: TypeIdentificator
+ name: str
+ description: str
+ type_identifier: FluxioTypeIdentifier
+ optional: bool
+ default_value: str | None
+
+ def __init__(
+ self,
+ name: str,
+ origin=TypeIdentificator("", ""),
+ description: str = "",
+ type_identifier: FluxioTypeIdentifier = FluxioTypeIdentifier.BASIC,
+ optional: bool = False,
+ default_value: str | None = None,
+ ):
+ self.origin = origin
+ self.name = name
+ self.description = description
+ self.type_identifier = type_identifier
+ self.optional = optional
+ self.default_value = default_value
+
+
+@dc.dataclass
+class FluxioGenericType(FluxioBasicType):
+ generic: FluxioBasicType
+ max_number_of_elements: int
+
+ def __init__(
+ self,
+ name: str,
+ generic: FluxioBasicType,
+ origin=TypeIdentificator("", ""),
+ description: str = "",
+ type_identifier: FluxioTypeIdentifier = FluxioTypeIdentifier.GENERIC,
+ default_value: str = "",
+ optional: bool = False,
+ max_number_of_elements: int = 0,
+ ):
+ super().__init__(
+ origin=origin,
+ name=name,
+ description=description,
+ optional=optional,
+ default_value=default_value,
+ type_identifier=type_identifier,
+ )
+ self.generic = generic
+ self.max_number_of_elements = max_number_of_elements
+
+
+@dc.dataclass
+class FluxioDoubleGenericType(FluxioBasicType):
+ generic1: FluxioBasicType
+ generic2: FluxioBasicType
+ max_number_of_elements: int
+
+ def __init__(
+ self,
+ name: str,
+ generic1: FluxioBasicType,
+ generic2: FluxioBasicType,
+ origin=TypeIdentificator("", ""),
+ description: str = "",
+ type_identifier: FluxioTypeIdentifier = FluxioTypeIdentifier.DOUBLE_GENERIC,
+ default_value: str = "",
+ optional: bool = False,
+ max_number_of_elements: int = 0,
+ ):
+ super().__init__(
+ origin=origin,
+ name=name,
+ description=description,
+ optional=optional,
+ default_value=default_value,
+ type_identifier=type_identifier,
+ )
+ self.generic1 = generic1
+ self.generic2 = generic2
+ self.max_number_of_elements = max_number_of_elements
+
+
+@dc.dataclass
+class FluxioObjectAttribute:
+ name: str
+ type: FluxioBasicType
+
+
+@dc.dataclass
+class FluxioObjectType(FluxioBasicType):
+ attributes: list[FluxioObjectAttribute]
+ parent: Optional["FluxioObjectType"]
+
+ def __init__(
+ self,
+ name: str,
+ origin=TypeIdentificator("", ""),
+ description: str = "",
+ type_identifier: FluxioTypeIdentifier = FluxioTypeIdentifier.OBJECT,
+ optional: bool = False,
+ default_value: str | None = None,
+ parent: Optional["FluxioObjectType"] = None,
+ ):
+ super().__init__(
+ origin=origin,
+ name=name,
+ description=description,
+ optional=optional,
+ default_value=default_value,
+ type_identifier=type_identifier,
+ )
+ self.parent = parent
+ self.attributes = []
+
+ def add_attribute(self, name: str, type: FluxioBasicType):
+ self.attributes.append(FluxioObjectAttribute(name=name, type=type))
+
+
+@dc.dataclass
+class FluxioEnumType(FluxioBasicType):
+ attributes: dict[str, int]
+
+ def __init__(
+ self,
+ name: str,
+ origin=TypeIdentificator("", ""),
+ description: str = "",
+ type_identifier: FluxioTypeIdentifier = FluxioTypeIdentifier.ENUM,
+ optional: bool = False,
+ default_value: str | None = None,
+ ):
+ super().__init__(
+ origin=origin,
+ name=name,
+ description=description,
+ optional=optional,
+ default_value=default_value,
+ type_identifier=type_identifier,
+ )
+ self.attributes = {}
+
+ def add_attribute(self, name: str, value: int):
+ self.attributes[name] = value
diff --git a/armarx_skills/manager/fluxio_value.py b/armarx_skills/manager/fluxio_value.py
new file mode 100644
index 0000000000000000000000000000000000000000..a8eb5ebf37ec32bf083c281f502d7796adeecbc8
--- /dev/null
+++ b/armarx_skills/manager/fluxio_value.py
@@ -0,0 +1,103 @@
+import dataclasses as dc
+
+from armarx_skills.manager import dto
+from armarx_skills.manager.fluxio_identificator import (
+ Identificator,
+ FluxioIdentificatorConv,
+)
+from armarx_memory.ice_conv.ice_converter import IceConverter
+from armarx_skills.manager.fluxio_data import FluxioBaseData
+from armarx_skills.manager.fluxio_data_aron_converter import FluxioDataAronConverter
+
+
+@dc.dataclass
+class Value:
+ """
+ A class representing a value with a profile ID and content.
+
+ Attributes:
+ profileId (Identificator): The profile ID associated with the value.
+ content (FluxioData): The content of the value.
+ """
+
+ profileId: Identificator
+ content: FluxioBaseData
+
+
+class FluxioValueConv(IceConverter):
+ """
+ A class for converting values to and from Fluxio format.
+
+ This class inherits from IceConverter and provides methods for converting values to and from Fluxio format.
+
+ Attributes:
+ - None
+
+ Methods:
+ - __init__(): Initializes the FluxioValueConv object.
+ - _import_dto(): Imports the DTO class for the converter.
+ - _from_ice(): Converts the DTO to the Value object.
+ - _to_ice(): Converts the Value object to the DTO.
+
+ """
+
+ def __init__(self):
+ """
+ Initializes the FluxioValueConv object.
+
+ Parameters:
+ - None
+
+ Return:
+ - None
+ """
+ super().__init__()
+
+ @classmethod
+ def _import_dto(cls):
+ """
+ This method is a class method that returns the FluxioValue data transfer object (DTO).
+
+ Return:
+ dto.FluxioValue: The FluxioValue data transfer object.
+
+ Raises:
+ None
+ """
+ return dto.FluxioValue
+
+ def _from_ice(self, dt: dto.FluxioValue) -> Value:
+ """
+ Converts a dto.FluxioValue object to a Value object.
+
+ Parameters:
+ dt (dto.FluxioValue): The dto.FluxioValue object to convert.
+
+ Return:
+ Value: The converted Value object.
+
+ Raises:
+ None
+ """
+ return Value(
+ profileId=FluxioIdentificatorConv().from_ice(dt.profileId),
+ content=FluxioDataAronConverter.aron_data_to_fluxio_data(dt.content),
+ )
+
+ def _to_ice(self, bo: Value) -> dto.FluxioValue:
+ """
+ Converts a Value object to a FluxioValue object.
+
+ Parameters:
+ bo (Value): The Value object to be converted.
+
+ Return:
+ dto.FluxioValue: The converted FluxioValue object.
+
+ Raises:
+ None
+ """
+ return dto.FluxioValue(
+ profileId=FluxioIdentificatorConv().to_ice(bo.profileId),
+ content=FluxioDataAronConverter.fluxio_data_to_aron(bo.content),
+ )
diff --git a/armarx_skills/manager/skill_manager.py b/armarx_skills/manager/skill_manager.py
index 12fc340144578994d3cfc7526e74bd169b78a9c1..8fdb92e064410acfcd638dc02c7ecff292e831d4 100644
--- a/armarx_skills/manager/skill_manager.py
+++ b/armarx_skills/manager/skill_manager.py
@@ -1,11 +1,11 @@
import Ice
-import typing as ty
+from typing import List, Optional, Callable, Dict
from armarx_core import ice_manager
from armarx_core.reconnecting_proxy_client import ReconnectingProxyClient
-from armarx_memory.aron.aron_ice_types import AronIceTypes
+from armarx_memory.aron.aron_ice_types import AronDataIceTypes, AronTypeIceTypes
from armarx_skills.core import execution_status as es
from armarx_skills.core.execution_status import ExecutionStatus
@@ -19,16 +19,48 @@ from armarx_skills.manager import provider_info as pi
from armarx_skills.manager import skill_execution_id as sei
from armarx_skills.manager import skill_execution_request as ser
from armarx_skills.manager import skill_status_update as ssu
-from armarx_skills.manager.skill_id import SkillID, SkillIdConv
+from armarx_skills.manager.skill_id import SkillID, SkillIdConv, ProviderID
from armarx_skills.manager import skill_description as sd
+from armarx_skills.manager.fluxio_skill_status_update import (
+ SkillStatusUpdate,
+ FluxioSkillStatusUpdateConv,
+)
+from armarx_skills.manager.fluxio_skill import Skill, FluxioSkillConv
+from armarx_skills.manager.fluxio_value import Value, FluxioValueConv
+from armarx_skills.manager.fluxio_parameter import Parameter, FluxioParameterConv
+from armarx_skills.manager.fluxio_profile import Profile, FluxioProfileConv
+from armarx_skills.manager.fluxio_provider import Provider, FluxioProviderConv
+from armarx_skills.manager.fluxio_exception import FluxioException, FluxioExceptionConv
+from armarx_skills.manager.fluxio_identificator import (
+ Identificator,
+ FluxioIdentificatorConv,
+)
+from armarx_skills.manager.fluxio_type_identificator import (
+ TypeIdentificator,
+ FluxioTypeIdentificatorConv,
+)
+from armarx_skills.manager.fluxio_type_aron_converter import FluxioTypeAronConverter
+from armarx_skills.manager.fluxio_data_aron_converter import FluxioDataAronConverter
+from armarx_skills.manager.fluxio_types import FluxioObjectType, FluxioTypeIdentifier
+
+
+def fluxio_wrapper(self, func):
+ def wrapper(*args, **kwargs):
+ try:
+ return func(*args, **kwargs)
+ except dto.FluxioException as e:
+ raise self.fluxio_exception_converter.from_ice(e)
+
+ return wrapper
+
class SkillManager:
DEFAULT_ICE_OBJECT_NAME = "SkillMemory"
def __init__(
- self,
- proxy: dti.SkillManagerInterfacePrx = None,
+ self,
+ proxy: dti.SkillManagerInterfacePrx = None,
):
self.proxy = proxy
@@ -40,54 +72,391 @@ class SkillManager:
self.skill_description_conv = sd.SkillDescriptionConv()
self.skill_description_map_conv = sd.SkillDescriptionMapConv()
self.abort_skill_result_conv = asr.AbortSkillResultConv()
+ self.fluxio_skill_conv = FluxioSkillConv()
+ self.fluxio_profile_conv = FluxioProfileConv()
+ self.fluxio_value_conv = FluxioValueConv()
+ self.fluxio_provider_conv = FluxioProviderConv()
+ self.fluxio_skill_status_update_conv = FluxioSkillStatusUpdateConv()
+ self.fluxio_exception_converter = FluxioExceptionConv()
+ self.fluxio_identificator_conv = FluxioIdentificatorConv()
+ self.fluxio_type_identificator_conv = FluxioTypeIdentificatorConv()
+ self.fluxio_parameter_conv = FluxioParameterConv()
+
+ def __getattribute__(self, name):
+ attr = object.__getattribute__(self, name)
+ if callable(attr):
+ return fluxio_wrapper(self, attr)
+ return attr
@classmethod
def wait_for_manager(
- cls,
- name=DEFAULT_ICE_OBJECT_NAME,
+ cls,
+ name=DEFAULT_ICE_OBJECT_NAME,
) -> "SkillManager":
return cls(proxy=ice_manager.wait_for_proxy(dti.SkillManagerInterfacePrx, name))
@classmethod
def get_manager(
- cls,
- name=DEFAULT_ICE_OBJECT_NAME,
- ) -> ty.Optional["SkillManager"]:
+ cls,
+ name=DEFAULT_ICE_OBJECT_NAME,
+ ) -> Optional["SkillManager"]:
try:
return cls(proxy=ice_manager.get_proxy(dti.SkillManagerInterfacePrx, name))
except Ice.ConnectionRefusedException:
return None
- def add_provider(
+ ### Fluxio related methods
+ # Everything skill related
+
+ def get_skills(
+ self,
+ ) -> List[Skill]:
+ skills_ice = self.get_skills_ice()
+ skills = []
+
+ for s in skills_ice:
+ skill = self.fluxio_skill_conv.from_ice(s)
+ skills.append(skill)
+
+ return skills
+
+ def get_skills_ice(self) -> dto.FluxioSkillList:
+ return self.proxy.getSkillList()
+
+ def get_skill(
+ self,
+ skill_id: str,
+ ) -> Skill | None:
+ skill_ice = self.get_skill_ice(skill_id)
+
+ if skill_ice is None:
+ return None
+
+ return self.fluxio_skill_conv.from_ice(skill_ice)
+
+ def get_skill_ice(self, skill_id: str) -> Optional[dto.FluxioSkill]:
+ return self._ice_unset_to_none(self.proxy.getSkill(id=skill_id))
+
+ def update_skill(
+ self,
+ user_id: str,
+ skill: Skill,
+ ) -> None:
+ skill_ice = self.fluxio_skill_conv.to_ice(skill)
+ return self.update_skill_ice(user_id=user_id, skill=skill_ice)
+
+ def update_skill_ice(self, user_id: str, skill: dto.FluxioSkill) -> None:
+ return self.proxy.updateSkill(userId=user_id, skill=skill)
+
+ def update_skill_parameter_values(
+ self,
+ user_id: str,
+ skill_id: str,
+ parameter_id: str,
+ values: List[Value]
+ ) -> None:
+ values_ice = []
+ for v in values:
+ values_ice.append(self.fluxio_value_conv.to_ice(v))
+ return self.update_skill_parameter_values_ice(
+ user_id=user_id, skill_id=skill_id, parameter_id=parameter_id, values=values_ice
+ )
+
+ def update_skill_parameter_values_ice(
self,
- provider_info: pi.ProviderInfo,
+ user_id: str,
+ skill_id: str,
+ parameter_id: str,
+ values: dto.FluxioValueList
+ ) -> None:
+ return self.proxy.updateSkillParameterValues(
+ userId=user_id, skillId=skill_id, parameterId=parameter_id, values=values
+ )
+
+ def delete_skill(
+ self,
+ skill_id: str,
+ user_id: str,
+ dry_run: bool,
+ ) -> List[Identificator] | None:
+ affected_skills_ice = self.delete_skill_ice(
+ skill_id=skill_id, user_id=user_id, dry_run=dry_run
+ )
+ affected_skills = []
+
+ if affected_skills_ice is None:
+ return None
+
+ for res in affected_skills_ice:
+ affected_skills.append(self.fluxio_identificator_conv.from_ice(res))
+
+ return affected_skills
+
+ def delete_skill_ice(
+ self, skill_id: str, user_id: str, dry_run: bool
+ ) -> Optional[dto.FluxioIdentificatorList]:
+ return self.proxy.deleteSkill(skillId=skill_id, userId=user_id, dryRun=dry_run)
+
+ def get_skill_mutex(
+ self,
+ skill_id: str,
+ user_id: str,
+ ) -> bool:
+ return self.get_skill_mutex_ice(skill_id=skill_id, user_id=user_id)
+
+ def get_skill_mutex_ice(self, skill_id: str, user_id: str) -> bool:
+ return self.proxy.getSkillMutex(skillId=skill_id, userId=user_id)
+
+ def delete_skill_mutex(
+ self,
+ skill_id: str,
+ user_id: str,
+ ) -> None:
+ return self.delete_skill_mutex_ice(skill_id=skill_id, user_id=user_id)
+
+ def delete_skill_mutex_ice(self, skill_id: str, user_id: str) -> None:
+ return self.proxy.deleteSkillMutex(skillId=skill_id, userId=user_id)
+
+ def update_skill_parameter(
+ self,
+ user_id: str,
+ skill_id: str,
+ parameter: Parameter,
+ dry_run: bool,
+ ) -> List[Identificator] | None:
+ parameter_ice = self.fluxio_parameter_conv.to_ice(parameter)
+ return self.update_skill_parameter_ice(
+ user_id=user_id, skill_id=skill_id, parameter=parameter_ice,dry_run=dry_run
+ )
+
+ def update_skill_parameter_ice(
+ self, user_id: str, skill_id: str, parameter: dto.FluxioParameter, dry_run: bool
+ ) -> Optional[dto.FluxioIdentificatorList]:
+ return self.proxy.updateSkillParameter(
+ userId=user_id, skillId=skill_id, parameter=parameter, dryRun=dry_run
+ )
+
+ def delete_skill_parameter(
+ self,
+ user_id: str,
+ skill_id: str,
+ parameter_id: str,
+ dry_run: bool,
+ ) -> List[Identificator] | None:
+ return self.delete_skill_parameter_ice(
+ user_id=user_id, skill_id=skill_id, parameter_id=parameter_id,dry_run=dry_run
+ )
+
+ def delete_skill_parameter_ice(
+ self, user_id: str, skill_id: str, parameter_id: str, dry_run: bool
+ ) -> Optional[dto.FluxioIdentificatorList]:
+ return self.proxy.deleteSkillParameter(
+ userId=user_id, skillId=skill_id, parameterId=parameter_id, dryRun=dry_run
+ )
+
+ # Everything profile related
+ def get_profiles(
+ self,
+ ) -> List[Profile]:
+ profiles_ice = self.get_profiles_ice()
+ profiles = []
+
+ for p in profiles_ice:
+ profile = self.fluxio_profile_conv.from_ice(p)
+ profiles.append(profile)
+
+ # sort profiles by dependencies
+ save_break = 0
+ prof_len = len(profiles)
+ sorted_profiles = []
+ while len(profiles) > 0:
+ # save break with pover of 2
+ if save_break == 2 ** prof_len:
+ break
+ for p in profiles:
+ if p.parentId is None or p.parentId.id in [
+ sp.id for sp in sorted_profiles
+ ]:
+ sorted_profiles.append(p)
+ profiles.remove(p)
+ save_break += 1
+
+ return sorted_profiles
+
+ def get_profiles_ice(self) -> dto.FluxioProfileList:
+ return self.proxy.getProfileList()
+
+ def get_profile(
+ self,
+ profile_id: str,
+ ) -> Profile | None:
+ profile_ice = self.get_profile_ice(profile_id=profile_id)
+
+ if profile_ice is None:
+ return None
+
+ return self.fluxio_profile_conv.from_ice(profile_ice)
+
+ def get_profile_ice(self, profile_id: str) -> Optional[dto.FluxioProfile]:
+ return self._ice_unset_to_none(self.proxy.getProfile(id=profile_id))
+
+ def create_profile(
+ self,
+ profile: Profile,
+ ) -> Profile:
+ profile_ice = self.fluxio_profile_conv.to_ice(profile)
+ profile_with_id_ice = self.create_profile_ice(profile=profile_ice)
+
+ return self.fluxio_profile_conv.from_ice(profile_with_id_ice)
+
+ def create_profile_ice(self, profile: dto.FluxioProfile) -> dto.FluxioProfile:
+ return self.proxy.createProfile(profile=profile)
+
+ def update_profile(
+ self,
+ profile: Profile,
+ ) -> None:
+ profile_ice = self.fluxio_profile_conv.to_ice(profile)
+ return self.update_profile_ice(profile=profile_ice)
+
+ def update_profile_ice(self, profile: dto.FluxioProfile) -> None:
+ return self.proxy.updateProfile(profile=profile)
+
+ # Everything provider related
+ def get_providers(
+ self,
+ ) -> List[Provider]:
+ providers_ice = self.get_providers_ice()
+ providers = []
+
+ for p in providers_ice:
+ provider = self.fluxio_provider_conv.from_ice(p)
+ providers.append(provider)
+
+ return providers
+
+ def get_providers_ice(self) -> dto.FluxioProviderList:
+ return self.proxy.getProviderList()
+
+ def get_provider(
+ self,
+ provider_id: str,
+ ) -> Provider | None:
+ provider_ice = self.get_provider_ice(provider_id=provider_id)
+
+ if provider_ice is None:
+ return None
+
+ return self.fluxio_provider_conv.from_ice(provider_ice)
+
+ def get_provider_ice(self, provider_id: str) -> Optional[dto.FluxioProvider]:
+ return self._ice_unset_to_none(self.proxy.getProvider(id=provider_id))
+
+ def get_skills_of_provider(
+ self,
+ provider_id: str,
+ ) -> List[Skill] | None:
+ skills_ice = self.get_skills_of_provider_ice(provider_id=provider_id)
+
+ if skills_ice is None:
+ return None
+
+ skills = []
+
+ for s in skills_ice:
+ skill = self.fluxio_skill_conv.from_ice(s)
+ skills.append(skill)
+
+ return skills
+
+ def get_skills_of_provider_ice(
+ self, provider_id: str
+ ) -> Optional[dto.FluxioSkillList]:
+ list = self.proxy.getSkillsOfProvider(id=provider_id)
+
+ return self._ice_unset_to_none(list)
+
+ def add_new_skill_to_provider(
+ self,
+ user_id: str,
+ provider_id: str,
+ skill: Skill,
+ ) -> Skill | None:
+ skill_ice = self.fluxio_skill_conv.to_ice(skill)
+ skill_with_id_ice = self.add_new_skill_to_provider_ice(
+ user_id=user_id, provider_id=provider_id, skill=skill_ice
+ )
+
+ if skill_with_id_ice is None:
+ return None
+
+ return self.fluxio_skill_conv.from_ice(skill_with_id_ice)
+
+ def add_new_skill_to_provider_ice(
+ self, user_id: str, provider_id: str, skill: dto.FluxioSkill
+ ) -> Optional[dto.FluxioSkill]:
+ return self._ice_unset_to_none(
+ self.proxy.addSkillToProvider(
+ userId=user_id, providerId=provider_id, skill=skill
+ )
+ )
+
+ def get_types(self) -> AronTypeIceTypes.Object:
+ return self.proxy.getTypes()
+
+ def get_types_fluxio(self) -> FluxioObjectType:
+ types = self.get_types()
+ res = FluxioTypeAronConverter.aron_type_to_fluxio_type(types)
+ self.fix_fluxio_types_origin(res)
+ return res
+
+ def fix_fluxio_types_origin(self, fluxio_types: FluxioObjectType) -> None:
+ for attr in fluxio_types.attributes:
+ skillId = attr.name
+ if attr.type.type_identifier != FluxioTypeIdentifier.OBJECT:
+ return
+ if isinstance(attr.type, FluxioObjectType):
+ for attr2 in attr.type.attributes:
+ parameterId = attr2.name
+ attr2.type.origin = TypeIdentificator(skillId, parameterId)
+
+ def get_test_data(self) -> AronDataIceTypes.Dict:
+ res: AronDataIceTypes.Dict = self.proxy.getTestData()
+ ret = FluxioDataAronConverter.aron_data_to_fluxio_data(res)
+ return ret
+
+ #### End of Fluxio related methods
+
+ def add_provider(
+ self,
+ provider_info: pi.ProviderInfo,
) -> None:
provider_info_ice = self.provider_info_conv.to_ice(provider_info)
return self.add_provider_ice(provider_info_ice)
def add_provider_ice(
- self,
- provider_info: dto.ProviderInfo,
+ self,
+ provider_info: dto.ProviderInfo,
) -> None:
return self.proxy.addProvider(provider_info)
def remove_provider(
- self,
- provider_info: pi.ProviderInfo,
+ self,
+ provider_info: pi.ProviderInfo,
) -> None:
provider_info_ice = self.provider_info_conv.to_ice(provider_info)
return self.remove_provider_ice(provider_info_ice)
def remove_provider_ice(
- self,
- provider_info: dto.ProviderInfo,
+ self,
+ provider_info: dto.ProviderInfo,
) -> None:
return self.proxy.removeProvider(providerInfo=provider_info)
def get_skill_description(
- self,
- skill_id: SkillID,
- ) -> ty.Optional[sd.SkillDescription]:
+ self,
+ skill_id: SkillID,
+ ) -> Optional[sd.SkillDescription]:
skill_id_ice = self.skill_id_conv.to_ice(skill_id)
description_ice = self.get_skill_description_ice(skill_id=skill_id_ice)
@@ -98,27 +467,49 @@ class SkillManager:
return description
def get_skill_description_ice(
- self,
- skill_id: dto.SkillID,
- ) -> ty.Optional[dto.SkillDescription]:
+ self,
+ skill_id: dto.SkillID,
+ ) -> Optional[dto.SkillDescription]:
return self._ice_unset_to_none(self.proxy.getSkillDescription(skillId=skill_id))
def get_skill_descriptions(
- self,
+ self,
) -> sd.SkillDescriptionMap:
descriptions_ice = self.get_skill_descriptions_ice()
descriptions = self.skill_description_map_conv.from_ice(descriptions_ice)
return descriptions
def get_skill_descriptions_ice(
- self,
+ self,
) -> dto.SkillDescriptionMap:
return self.proxy.getSkillDescriptions()
+ def get_fluxio_skill_execution_status(
+ self,
+ execution_id: str,
+ ) -> List[SkillStatusUpdate] | None:
+ status_list_ice = self.get_fluxio_skill_execution_status_ice(
+ execution_id=execution_id
+ )
+ if status_list_ice is None:
+ return None
+ status_list = []
+ for status_ice in status_list_ice:
+ status = self.fluxio_skill_status_update_conv.from_ice(status_ice)
+ status_list.append(status)
+
+ return status_list
+
+ def get_fluxio_skill_execution_status_ice(
+ self,
+ execution_id: str,
+ ) -> Optional[dto.FluxioSkillStatusUpdateList]:
+ return self.proxy.getFluxioSkillExecutionStatus(executionId=execution_id)
+
def get_skill_execution_status(
- self,
- execution_id: sei.SkillExecutionID,
- ) -> ty.Optional[ssu.SkillStatusUpdate]:
+ self,
+ execution_id: sei.SkillExecutionID,
+ ) -> Optional[ssu.SkillStatusUpdate]:
execution_id_ice = self.skill_execution_id_conv.to_ice(execution_id)
status_ice = self.get_skill_execution_status_ice(execution_id_ice)
@@ -129,23 +520,25 @@ class SkillManager:
return status
def get_skill_execution_status_ice(
- self,
- execution_id: dto.SkillExecutionID,
- ) -> ty.Optional[dto.SkillStatusUpdate]:
+ self,
+ execution_id: dto.SkillExecutionID,
+ ) -> Optional[dto.SkillStatusUpdate]:
return self._ice_unset_to_none(self.proxy.getSkillExecutionStatus(execution_id))
def get_skill_execution_statuses(
- self,
+ self,
) -> ssu.SkillStatusUpdateMap:
updates_ice = self.get_skill_execution_statuses_ice()
updates = {
- self.skill_execution_id_conv.from_ice(k): self.skill_status_update_conv.from_ice(v)
+ self.skill_execution_id_conv.from_ice(
+ k
+ ): self.skill_status_update_conv.from_ice(v)
for k, v in updates_ice.items()
}
return updates
def get_skill_execution_statuses_ice(
- self,
+ self,
) -> dto.SkillStatusUpdateMap:
return self.proxy.getSkillExecutionStatuses()
@@ -155,7 +548,7 @@ class SkillManager:
asynchronous=False,
log=None,
dry=False,
- ) -> ty.Optional[ssu.SkillStatusUpdate]:
+ ) -> Optional[ssu.SkillStatusUpdate]:
# Return is optional because returning None when dry == True.
if log is not None:
@@ -179,7 +572,9 @@ class SkillManager:
ExecutionStatus.Preparing, {})
if log is not None:
- log.info(f"Skill {request.skill_id} terminated with status '{update.status.name}'.")
+ log.info(
+ f"Skill {request.skill_id} terminated with status '{update.status.name}'."
+ )
return update
@@ -187,7 +582,7 @@ class SkillManager:
self,
executor_name: str,
skill_id: SkillID,
- parameters_function: ty.Callable[[ty.Dict], None],
+ parameters_function: Callable[[Dict], None],
asynchronous=False,
log=None,
dry=False
@@ -218,11 +613,11 @@ class SkillManager:
dry=dry)
def execute_skill_with_default_parameters(
- self,
- executor_name: str,
- skill_id: SkillID,
- log=None,
- dry=False,
+ self,
+ executor_name: str,
+ skill_id: SkillID,
+ log=None,
+ dry=False,
) -> ssu.SkillStatusUpdate:
"""
Execute a skill with its default parameters.
@@ -235,22 +630,46 @@ class SkillManager:
dry=dry,
)
+ def execute_skill_fluxio(
+ self,
+ skill_id: str,
+ profile_id: str,
+ ) -> str | None:
+ execution_id = self.execute_skill_fluxio_ice(skill_id=skill_id, profile_id=profile_id)
+
+ if execution_id is None:
+ return None
+
+ return execution_id
+
+ def execute_skill_fluxio_ice(self, skill_id: str, profile_id: str) -> Optional[str]:
+ return self.proxy.executeFluxioSkill(skillId=skill_id, profileId=profile_id)
+
+ def abort_skill_fluxio(
+ self,
+ execution_id: str,
+ ):
+ self.abort_skill_fluxio_ice(execution_id=execution_id)
+
+ def abort_skill_fluxio_ice(self, execution_id: str):
+ self.proxy.abortFluxioSkill(executionId=execution_id)
+
def execute_skill_ice(
- self,
- request: dto.SkillExecutionRequest,
+ self,
+ request: dto.SkillExecutionRequest,
) -> dto.SkillStatusUpdate:
return self.proxy.executeSkill(skillExecutionRequest=request)
def execute_skill_async_ice(
- self,
- request: dto.SkillExecutionRequest,
+ self,
+ request: dto.SkillExecutionRequest,
) -> dto.SkillExecutionID:
return self.proxy.executeSkillAsync(skillExecutionRequest=request)
def update_skill_parameters_ice(
- self,
- execution_id: dto.SkillExecutionID,
- parameters: AronIceTypes.Dict,
+ self,
+ execution_id: dto.SkillExecutionID,
+ parameters: AronDataIceTypes.Dict,
) -> dto.SkillExecutionID:
return self.proxy.updateSkillParameters(
executionId=execution_id,
@@ -258,9 +677,9 @@ class SkillManager:
)
def abort_skill(
- self,
- skill_execution_id: sei.SkillExecutionID,
- blocking=True,
+ self,
+ skill_execution_id: sei.SkillExecutionID,
+ blocking=True,
) -> asr.AbortSkillResult:
skill_execution_id_ice = self.skill_execution_id_conv.to_ice(skill_execution_id)
@@ -273,8 +692,8 @@ class SkillManager:
return abort_result
def abort_skill_async(
- self,
- skill_execution_id: sei.SkillExecutionID,
+ self,
+ skill_execution_id: sei.SkillExecutionID,
) -> asr.AbortSkillResult:
skill_execution_id_ice = self.skill_execution_id_conv.to_ice(skill_execution_id)
@@ -286,9 +705,9 @@ class SkillManager:
return abort_result
def abort_skill_ice(
- self,
- execution_id: dto.SkillExecutionID,
- blocking=True,
+ self,
+ execution_id: dto.SkillExecutionID,
+ blocking=True,
) -> provider_dto.AbortSkillResult:
if blocking:
return self.proxy.abortSkill(executionId=execution_id)
@@ -296,23 +715,22 @@ class SkillManager:
return self.abort_skill_async_ice(execution_id=execution_id)
def abort_skill_async_ice(
- self,
- execution_id: dto.SkillExecutionID,
+ self,
+ execution_id: dto.SkillExecutionID,
) -> provider_dto.AbortSkillResult:
return self.proxy.abortSkillAsync(executionId=execution_id)
-
def abort_all_non_terminated_skills(
- self,
- abort_statuses=(
- es.ExecutionStatus.Constructing,
- es.ExecutionStatus.Initializing,
- es.ExecutionStatus.Preparing,
- es.ExecutionStatus.Running,
- ),
- blocking=True,
- log=None,
- ) -> ty.Dict[sei.SkillExecutionID, asr.AbortSkillResult]:
+ self,
+ abort_statuses=(
+ es.ExecutionStatus.Constructing,
+ es.ExecutionStatus.Initializing,
+ es.ExecutionStatus.Preparing,
+ es.ExecutionStatus.Running,
+ ),
+ blocking=True,
+ log=None,
+ ) -> Dict[sei.SkillExecutionID, asr.AbortSkillResult]:
# temporary fix until RobotAPI is merged
"""execution_ids_to_abort = list()
@@ -329,7 +747,9 @@ class SkillManager:
for execution_id in execution_ids_to_abort:
if log is not None:
log.info(f"Aborting skill execution {execution_id} ...")
- result = self.abort_skill(skill_execution_id=execution_id, blocking=blocking)
+ result = self.abort_skill(
+ skill_execution_id=execution_id, blocking=blocking
+ )
results[execution_id] = result
return results"""
@@ -342,25 +762,26 @@ class SkillManager:
class ReconnectingSkillManager(ReconnectingProxyClient):
-
def __init__(
- self,
- name=SkillManager.DEFAULT_ICE_OBJECT_NAME,
+ self,
+ name=SkillManager.DEFAULT_ICE_OBJECT_NAME,
):
super().__init__()
self.name = name
- def get(self, wait=True, log=None, logger=None) -> ty.Optional[SkillManager]:
+ def get(self, wait=True, log=None, logger=None) -> Optional[SkillManager]:
if logger is not None:
logger.info("The parameter 'logger' is deprecated. Use 'log' instead.")
if log is None:
log = logger
else:
- raise ValueError("The parameter 'logger' is deprecated. Use 'log' instead.")
+ raise ValueError(
+ "The parameter 'logger' is deprecated. Use 'log' instead."
+ )
return self._get(wait=wait, log=log)
- def _get_client(self, log=None) -> ty.Optional[SkillManager]:
+ def _get_client(self, log=None) -> Optional[SkillManager]:
if log is not None:
log.info(f"Waiting for skill manager '{self.name}' ...")
@@ -386,10 +807,10 @@ class ReconnectingSkillManager(ReconnectingProxyClient):
return client
-if __name__ == '__main__':
+if __name__ == "__main__":
+
def test_main():
manager = SkillManager.wait_for_manager()
manager.get_skill_execution_statuses()
-
test_main()
diff --git a/armarx_skills/provider/skill_provider_client.py b/armarx_skills/provider/skill_provider_client.py
index 64eaf277386b222944e9ee3b7f28c98bdf081a57..49e6233eb5bb0f0a6300cc4492f90e5745a4956c 100644
--- a/armarx_skills/provider/skill_provider_client.py
+++ b/armarx_skills/provider/skill_provider_client.py
@@ -2,7 +2,7 @@ import Ice
import typing as ty
-from armarx_memory.aron.aron_ice_types import AronIceTypes
+from armarx_memory.aron.aron_ice_types import AronDataIceTypes
from armarx_skills.provider import dti, dto
@@ -68,7 +68,7 @@ class SkillProviderClient:
def update_skill_parameters_ice(
self,
execution_id: dto.SkillExecutionID,
- parameters: AronIceTypes.Dict,
+ parameters: AronDataIceTypes.Dict,
) -> dto.ParameterUpdateResult:
return self.proxy.updateSkillParameters(
executionId=execution_id,
diff --git a/pyrightconfig.json b/pyrightconfig.json
new file mode 100644
index 0000000000000000000000000000000000000000..8fd86437bcb1a14ea177d01411c0afbd3d715577
--- /dev/null
+++ b/pyrightconfig.json
@@ -0,0 +1,4 @@
+{
+ "venvPath": ".",
+ "venv": ".venv"
+}
diff --git a/tests/armarx/armarx_memory/aron/test_conversion.py b/tests/armarx/armarx_memory/aron/test_conversion.py
index 80ac774ee8fdb434182e39076c8d8c1cfe24af14..4fca184cf6a590dbc7f39dc4fc9ecc7b82857cff 100644
--- a/tests/armarx/armarx_memory/aron/test_conversion.py
+++ b/tests/armarx/armarx_memory/aron/test_conversion.py
@@ -97,7 +97,7 @@ class ContainersOfDataclasses(AronDataclass):
def _test_preamble(cls):
- from armarx_memory.aron.aron_ice_types import AronIceTypes
+ from armarx_memory.aron.aron_ice_types import AronDataIceTypes
data_in = cls.make_test_data()
print(f"Input:\n{data_in}")
@@ -105,7 +105,7 @@ def _test_preamble(cls):
aron_ice = data_in.to_aron_ice()
print(f"Aron ice:\n{aron_ice}")
- assert isinstance(aron_ice, AronIceTypes.Dict)
+ assert isinstance(aron_ice, AronDataIceTypes.Dict)
data_out = cls.from_aron_ice(aron_ice)
print(f"Output:\n{data_out}")