A sequence of NumpyOps to run. They must all share the same mode. It also doesn't support scheduled ops at
the moment, though the Fuse itself may be scheduled.
required
Raises:
Type
Description
ValueError
If repeat or ops are invalid.
Source code in fastestimator\fastestimator\op\numpyop\meta\fuse.py
@traceable()classFuse(NumpyOp):"""Run a sequence of NumpyOps as a single Op. Args: ops: A sequence of NumpyOps to run. They must all share the same mode. It also doesn't support scheduled ops at the moment, though the Fuse itself may be scheduled. Raises: ValueError: If `repeat` or `ops` are invalid. """def__init__(self,ops:Union[NumpyOp,List[NumpyOp]])->None:ops=to_list(ops)iflen(ops)<1:raiseValueError("Fuse requires at least one op")inputs=[]outputs=[]mode=ops[0].modeforopinops:ifop.mode!=mode:raiseValueError(f"All Fuse ops must share the same mode, but got {mode} and {op.mode}")forinpinop.inputs:ifinpnotininputsandinpnotinoutputs:inputs.append(inp)foroutinop.outputs:ifoutnotinoutputs:outputs.append(out)super().__init__(inputs=inputs,outputs=outputs,mode=mode)self.ops=opsdef__getstate__(self)->Dict[str,List[Dict[Any,Any]]]:return{'ops':[elem.__getstate__()ifhasattr(elem,'__getstate__')else{}foreleminself.ops]}defforward(self,data:List[np.ndarray],state:Dict[str,Any])->List[np.ndarray]:data={key:elemforkey,eleminzip(self.inputs,data)}forward_numpyop(self.ops,data,state["mode"])return[data[key]forkeyinself.outputs]