numpyop

Delete

Bases: NumpyOp

Delete key(s) and their associated values from the data dictionary.

The system has special logic to detect instances of this Op and delete its inputs from the data dictionary.

Parameters:

Name	Type	Description	Default
keys	Union[str, List[str]]	Existing key(s) to be deleted from the data dictionary.	required
mode	Union[None, str, Iterable[str]]	What mode(s) to execute this Op in. For example, "train", "eval", "test", or "infer". To execute regardless of mode, pass None. To execute in all modes except for a particular one, you can pass an argument like "!infer" or "!train".	None
ds_id	Union[None, str, Iterable[str]]	What dataset id(s) to execute this Op in. To execute regardless of ds_id, pass None. To execute in all ds_ids except for a particular one, you can pass an argument like "!ds1".	None

Source code in fastestimator\fastestimator\op\numpyop\numpyop.py

@traceable()
class Delete(NumpyOp):
    """Delete key(s) and their associated values from the data dictionary.

    The system has special logic to detect instances of this Op and delete its `inputs` from the data dictionary.

    Args:
        keys: Existing key(s) to be deleted from the data dictionary.
        mode: What mode(s) to execute this Op in. For example, "train", "eval", "test", or "infer". To execute
            regardless of mode, pass None. To execute in all modes except for a particular one, you can pass an argument
            like "!infer" or "!train".
        ds_id: What dataset id(s) to execute this Op in. To execute regardless of ds_id, pass None. To execute in all
            ds_ids except for a particular one, you can pass an argument like "!ds1".
    """
    def __init__(self,
                 keys: Union[str, List[str]],
                 mode: Union[None, str, Iterable[str]] = None,
                 ds_id: Union[None, str, Iterable[str]] = None) -> None:
        super().__init__(inputs=keys, mode=mode, ds_id=ds_id)

    def forward(self, data: Union[np.ndarray, List[np.ndarray]], state: Dict[str, Any]) -> None:
        pass

    def forward_batch(self, data: Union[Tensor, List[Tensor]], state: Dict[str, Any]) -> None:
        pass

LambdaOp

Bases: NumpyOp

An Operator that performs any specified function as forward function.

Parameters:

Name	Type	Description	Default
fn	Callable	The function to be executed.	required
inputs	Union[None, str, Iterable[str]]	Key(s) from which to retrieve data from the data dictionary.	None
outputs	Union[None, str, Iterable[str]]	Key(s) under which to write the outputs of this Op back to the data dictionary.	None
mode	Union[None, str, Iterable[str]]	What mode(s) to execute this Op in. For example, "train", "eval", "test", or "infer". To execute regardless of mode, pass None. To execute in all modes except for a particular one, you can pass an argument like "!infer" or "!train".	None
ds_id	Union[None, str, Iterable[str]]	What dataset id(s) to execute this Op in. To execute regardless of ds_id, pass None. To execute in all ds_ids except for a particular one, you can pass an argument like "!ds1".	None

Source code in fastestimator\fastestimator\op\numpyop\numpyop.py

@traceable()
class LambdaOp(NumpyOp):
    """An Operator that performs any specified function as forward function.

    Args:
        fn: The function to be executed.
        inputs: Key(s) from which to retrieve data from the data dictionary.
        outputs: Key(s) under which to write the outputs of this Op back to the data dictionary.
        mode: What mode(s) to execute this Op in. For example, "train", "eval", "test", or "infer". To execute
            regardless of mode, pass None. To execute in all modes except for a particular one, you can pass an argument
            like "!infer" or "!train".
        ds_id: What dataset id(s) to execute this Op in. To execute regardless of ds_id, pass None. To execute in all
            ds_ids except for a particular one, you can pass an argument like "!ds1".
    """
    def __init__(self,
                 fn: Callable,
                 inputs: Union[None, str, Iterable[str]] = None,
                 outputs: Union[None, str, Iterable[str]] = None,
                 mode: Union[None, str, Iterable[str]] = None,
                 ds_id: Union[None, str, Iterable[str]] = None) -> None:
        super().__init__(inputs=inputs, outputs=outputs, mode=mode, ds_id=ds_id)
        self.fn = fn
        self.in_list = True

    def set_rua_level(self, magnitude_coef: float) -> None:
        """A method which will be invoked by the RUA Op to adjust the augmentation intensity.

        Args:
            magnitude_coef: The desired augmentation intensity (range [0-1]).
        """

    def forward(self, data: List[np.ndarray], state: Dict[str, Any]) -> Union[np.ndarray, List[np.ndarray]]:
        return self.fn(*data)

    def forward_batch(self, data: Union[Tensor, List[Tensor]], state: Dict[str,
                                                                           Any]) -> Union[np.ndarray, List[np.ndarray]]:
        return self.forward(data, state)

set_rua_level

A method which will be invoked by the RUA Op to adjust the augmentation intensity.

Parameters:

Name	Type	Description	Default
magnitude_coef	float	The desired augmentation intensity (range [0-1]).	required

Source code in fastestimator\fastestimator\op\numpyop\numpyop.py

def set_rua_level(self, magnitude_coef: float) -> None:
    """A method which will be invoked by the RUA Op to adjust the augmentation intensity.

    Args:
        magnitude_coef: The desired augmentation intensity (range [0-1]).
    """

NumpyOp

Bases: Op

An Operator class which takes and returns numpy data.

These Operators are used in fe.Pipeline to perform data pre-processing / augmentation. They may also be used in fe.Network to perform postprocessing on data.

Parameters:

Name	Type	Description	Default
inputs	Union[None, str, Iterable[str]]	Key(s) from which to retrieve data from the data dictionary.	None
outputs	Union[None, str, Iterable[str]]	Key(s) under which to write the outputs of this Op back to the data dictionary.	None
mode	Union[None, str, Iterable[str]]	What mode(s) to execute this Op in. For example, "train", "eval", "test", or "infer". To execute regardless of mode, pass None. To execute in all modes except for a particular one, you can pass an argument like "!infer" or "!train".	None
ds_id	Union[None, str, Iterable[str]]	What dataset id(s) to execute this Op in. To execute regardless of ds_id, pass None. To execute in all ds_ids except for a particular one, you can pass an argument like "!ds1".	None

Source code in fastestimator\fastestimator\op\numpyop\numpyop.py

@traceable()
class NumpyOp(Op):
    """An Operator class which takes and returns numpy data.

    These Operators are used in fe.Pipeline to perform data pre-processing / augmentation. They may also be used in
    fe.Network to perform postprocessing on data.

    Args:
        inputs: Key(s) from which to retrieve data from the data dictionary.
        outputs: Key(s) under which to write the outputs of this Op back to the data dictionary.
        mode: What mode(s) to execute this Op in. For example, "train", "eval", "test", or "infer". To execute
            regardless of mode, pass None. To execute in all modes except for a particular one, you can pass an argument
            like "!infer" or "!train".
        ds_id: What dataset id(s) to execute this Op in. To execute regardless of ds_id, pass None. To execute in all
            ds_ids except for a particular one, you can pass an argument like "!ds1".
    """
    def __init__(self,
                 inputs: Union[None, str, Iterable[str]] = None,
                 outputs: Union[None, str, Iterable[str]] = None,
                 mode: Union[None, str, Iterable[str]] = None,
                 ds_id: Union[None, str, Iterable[str]] = None) -> None:
        super().__init__(inputs=inputs, outputs=outputs, mode=mode, ds_id=ds_id)
        # in_place_edits tracks whether the .forward() method of this op will perform in-place edits of numpy arrays.
        # This is inferred automatically by the system and is used for memory management optimization. If you are
        # developing a NumpyOp which does in-place edits, the best practice is to set this to True in your init method.
        self.in_place_edits = False

    def forward(self, data: Union[np.ndarray, List[np.ndarray]],
                state: Dict[str, Any]) -> Union[np.ndarray, List[np.ndarray]]:
        """A method which will be invoked in order to transform data.

        This method will be invoked on individual elements of data before any batching / axis expansion is performed.

        Args:
            data: The arrays from the data dictionary corresponding to whatever keys this Op declares as its `inputs`.
            state: Information about the current execution context, for example {"mode": "train"}.

        Returns:
            The `data` after applying whatever transform this Op is responsible for. It will be written into the data
            dictionary based on whatever keys this Op declares as its `outputs`.
        """
        return data

    def forward_batch(self, data: Union[Tensor, List[Tensor]], state: Dict[str,
                                                                           Any]) -> Union[np.ndarray, List[np.ndarray]]:
        """A method which will be invoked in order to transform a batch of data.

        This method will be invoked on batches of data during network postprocessing. Note that the inputs may be numpy
        arrays or TF/Torch tensors. Outputs are expected to be Numpy arrays, though this is not enforced. Developers
        should probably not need to override this implementation unless they are building an op specifically intended
        for postprocessing.

        Args:
            data: The arrays from the data dictionary corresponding to whatever keys this Op declares as its `inputs`.
            state: Information about the current execution context, for example {"mode": "train"}.

        Returns:
            The `data` after applying whatever transform this Op is responsible for. It will be written into the data
            dictionary based on whatever keys this Op declares as its `outputs`.
        """
        if isinstance(data, List):
            data = [to_number(elem) for elem in data]
            batch_size = data[0].shape[0]
            data = [[elem[i] for elem in data] for i in range(batch_size)]
        else:
            data = to_number(data)
            data = [data[i] for i in range(data.shape[0])]
        results = [self.forward(elem, state) for elem in data]
        if self.out_list:
            results = [np.array(col) for col in [[row[i] for row in results] for i in range(len(results[0]))]]
        else:
            results = np.array(results)
        return results

forward

A method which will be invoked in order to transform data.

This method will be invoked on individual elements of data before any batching / axis expansion is performed.

Parameters:

Name	Type	Description	Default
data	Union[np.ndarray, List[np.ndarray]]	The arrays from the data dictionary corresponding to whatever keys this Op declares as its inputs.	required
state	Dict[str, Any]	Information about the current execution context, for example {"mode": "train"}.	required

Returns:

Type	Description
Union[np.ndarray, List[np.ndarray]]	The data after applying whatever transform this Op is responsible for. It will be written into the data
Union[np.ndarray, List[np.ndarray]]	dictionary based on whatever keys this Op declares as its outputs.

Source code in fastestimator\fastestimator\op\numpyop\numpyop.py

def forward(self, data: Union[np.ndarray, List[np.ndarray]],
            state: Dict[str, Any]) -> Union[np.ndarray, List[np.ndarray]]:
    """A method which will be invoked in order to transform data.

    This method will be invoked on individual elements of data before any batching / axis expansion is performed.

    Args:
        data: The arrays from the data dictionary corresponding to whatever keys this Op declares as its `inputs`.
        state: Information about the current execution context, for example {"mode": "train"}.

    Returns:
        The `data` after applying whatever transform this Op is responsible for. It will be written into the data
        dictionary based on whatever keys this Op declares as its `outputs`.
    """
    return data

forward_batch

A method which will be invoked in order to transform a batch of data.

This method will be invoked on batches of data during network postprocessing. Note that the inputs may be numpy arrays or TF/Torch tensors. Outputs are expected to be Numpy arrays, though this is not enforced. Developers should probably not need to override this implementation unless they are building an op specifically intended for postprocessing.

Parameters:

Name	Type	Description	Default
data	Union[Tensor, List[Tensor]]	The arrays from the data dictionary corresponding to whatever keys this Op declares as its inputs.	required
state	Dict[str, Any]	Information about the current execution context, for example {"mode": "train"}.	required

Returns:

Type	Description
Union[np.ndarray, List[np.ndarray]]	The data after applying whatever transform this Op is responsible for. It will be written into the data
Union[np.ndarray, List[np.ndarray]]	dictionary based on whatever keys this Op declares as its outputs.

Source code in fastestimator\fastestimator\op\numpyop\numpyop.py

def forward_batch(self, data: Union[Tensor, List[Tensor]], state: Dict[str,
                                                                       Any]) -> Union[np.ndarray, List[np.ndarray]]:
    """A method which will be invoked in order to transform a batch of data.

    This method will be invoked on batches of data during network postprocessing. Note that the inputs may be numpy
    arrays or TF/Torch tensors. Outputs are expected to be Numpy arrays, though this is not enforced. Developers
    should probably not need to override this implementation unless they are building an op specifically intended
    for postprocessing.

    Args:
        data: The arrays from the data dictionary corresponding to whatever keys this Op declares as its `inputs`.
        state: Information about the current execution context, for example {"mode": "train"}.

    Returns:
        The `data` after applying whatever transform this Op is responsible for. It will be written into the data
        dictionary based on whatever keys this Op declares as its `outputs`.
    """
    if isinstance(data, List):
        data = [to_number(elem) for elem in data]
        batch_size = data[0].shape[0]
        data = [[elem[i] for elem in data] for i in range(batch_size)]
    else:
        data = to_number(data)
        data = [data[i] for i in range(data.shape[0])]
    results = [self.forward(elem, state) for elem in data]
    if self.out_list:
        results = [np.array(col) for col in [[row[i] for row in results] for i in range(len(results[0]))]]
    else:
        results = np.array(results)
    return results

forward_numpyop

Call the forward function for list of NumpyOps, and modify the data dictionary in place.

Parameters:

Name	Type	Description	Default
ops	List[NumpyOp]	A list of NumpyOps to execute.	required
data	MutableMapping[str, Any]	The data dictionary.	required
state	Dict[str, Any]	Information about the current execution context, ex. {"mode": "train"}. Must contain at least the mode.	required
batched	bool	Whether the data is batched or not.	False

Source code in fastestimator\fastestimator\op\numpyop\numpyop.py

def forward_numpyop(ops: List[NumpyOp], data: MutableMapping[str, Any], state: Dict[str, Any],
                    batched: bool = False) -> None:
    """Call the forward function for list of NumpyOps, and modify the data dictionary in place.

    Args:
        ops: A list of NumpyOps to execute.
        data: The data dictionary.
        state: Information about the current execution context, ex. {"mode": "train"}. Must contain at least the mode.
        batched: Whether the `data` is batched or not.
    """
    for op in ops:
        op_data = get_inputs_by_op(op, data, copy_on_write=op.in_place_edits)
        try:
            op_data = op.forward_batch(op_data, state) if batched else op.forward(op_data, state)
        except ValueError as err:
            if err.args[0] == 'assignment destination is read-only':
                # If the numpy error text changes we'll need to make adjustments in the future
                op.in_place_edits = True
                op_data = get_inputs_by_op(op, data, copy_on_write=op.in_place_edits)
                op_data = op.forward_batch(op_data, state) if batched else op.forward(op_data, state)
            else:
                raise err
        if isinstance(op, Delete):
            for key in op.inputs:
                del data[key]
        if op.outputs:
            write_outputs_by_op(op, data, op_data)