safestructures.TensorProcessor

Bases: DataProcessor, ABC

Base class to process tensors.

Source code in src/safestructures/processors/base.py

class TensorProcessor(DataProcessor, ABC):
    """Base class to process tensors."""

    @abstractmethod
    def to_numpy(self, tensor: Any) -> np.ndarray:
        """Convert tensor to Numpy array.

        Args:
            tensor (Any): The tensor to convert to Numpy.

        Returns:
            np.ndarray: The tensor as a Numpy array.
        """
        pass

    def process_tensor(self, tensor: np.ndarray) -> str:
        """Process a tensor for serialization.

        Args:
            tensor (np.ndarray): The tensor as Numpy array.

        Returns:
            str: The tensor ID.
        """
        assert isinstance(tensor, np.ndarray), "Tensor must be np.ndarry at this point."
        _id = str(len(self.serializer.tensors))
        self.serializer.tensors[_id] = tensor
        return _id

    def serialize(self, tensor: Any) -> dict:
        """Overload `DataProcessor.serialize`."""
        tensor = self.to_numpy(tensor)
        return self.process_tensor(tensor)

    def deserialize(self, tensor_id: str, **kwargs) -> Any:
        """Overload `DataProcessor.deserialize`."""
        return self.serializer.tensors[tensor_id]

deserialize(tensor_id, **kwargs)

Overload DataProcessor.deserialize.

Source code in src/safestructures/processors/base.py

def deserialize(self, tensor_id: str, **kwargs) -> Any:
    """Overload `DataProcessor.deserialize`."""
    return self.serializer.tensors[tensor_id]

process_tensor(tensor)

Process a tensor for serialization.

Parameters:

Name	Type	Description	Default
tensor	ndarray	The tensor as Numpy array.	required

Returns:

Name	Type	Description
str	str	The tensor ID.

Source code in src/safestructures/processors/base.py

def process_tensor(self, tensor: np.ndarray) -> str:
    """Process a tensor for serialization.

    Args:
        tensor (np.ndarray): The tensor as Numpy array.

    Returns:
        str: The tensor ID.
    """
    assert isinstance(tensor, np.ndarray), "Tensor must be np.ndarry at this point."
    _id = str(len(self.serializer.tensors))
    self.serializer.tensors[_id] = tensor
    return _id

serialize(tensor)

Overload DataProcessor.serialize.

Source code in src/safestructures/processors/base.py

def serialize(self, tensor: Any) -> dict:
    """Overload `DataProcessor.serialize`."""
    tensor = self.to_numpy(tensor)
    return self.process_tensor(tensor)

to_numpy(tensor) abstractmethod

Convert tensor to Numpy array.

Parameters:

Name	Type	Description	Default
tensor	Any	The tensor to convert to Numpy.	required

Returns:

Type	Description
ndarray	np.ndarray: The tensor as a Numpy array.

Source code in src/safestructures/processors/base.py

@abstractmethod
def to_numpy(self, tensor: Any) -> np.ndarray:
    """Convert tensor to Numpy array.

    Args:
        tensor (Any): The tensor to convert to Numpy.

    Returns:
        np.ndarray: The tensor as a Numpy array.
    """
    pass