medkit.core.pipeline ==================== .. py:module:: medkit.core.pipeline Classes ------- .. autoapisummary:: medkit.core.pipeline.PipelineCompatibleOperation medkit.core.pipeline.ProvCompatibleOperation medkit.core.pipeline.DescribableOperation medkit.core.pipeline.PipelineStep medkit.core.pipeline.Pipeline Module Contents --------------- .. py:class:: PipelineCompatibleOperation Bases: :py:obj:`typing_extensions.Protocol` Base class for protocol classes. Protocol classes are defined as:: class Proto(Protocol): def meth(self) -> int: ... Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing). For example:: class C: def meth(self) -> int: return 0 def func(x: Proto) -> int: return x.meth() func(C()) # Passes static type check See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as:: class GenProto(Protocol[T]): def meth(self) -> T: ... .. !! processed by numpydoc !! .. py:method:: run(*all_input_data: list[Any]) -> list[Any] | tuple[list[Any], Ellipsis] | None Run the operation. :Parameters: **all_input_data** : list of Any One or several list of data items to process (according to the number of input the operation needs) :Returns: list of Any or tuple of list, optional Tuple of list of all new data items created by the operation. Can be None if the operation does not create any new data items but rather modify existing items in-place (for instance by adding attributes to existing annotations). If there is only one list of created data items, it is possible to return directly that list without wrapping it in a tuple. .. !! processed by numpydoc !! .. py:class:: ProvCompatibleOperation Bases: :py:obj:`typing_extensions.Protocol` Base class for protocol classes. Protocol classes are defined as:: class Proto(Protocol): def meth(self) -> int: ... Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing). For example:: class C: def meth(self) -> int: return 0 def func(x: Proto) -> int: return x.meth() func(C()) # Passes static type check See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as:: class GenProto(Protocol[T]): def meth(self) -> T: ... .. !! processed by numpydoc !! .. py:method:: set_prov_tracer(prov_tracer: medkit.core.prov_tracer.ProvTracer) .. py:class:: DescribableOperation Bases: :py:obj:`typing_extensions.Protocol` Base class for protocol classes. Protocol classes are defined as:: class Proto(Protocol): def meth(self) -> int: ... Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing). For example:: class C: def meth(self) -> int: return 0 def func(x: Proto) -> int: return x.meth() func(C()) # Passes static type check See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as:: class GenProto(Protocol[T]): def meth(self) -> T: ... .. !! processed by numpydoc !! .. py:attribute:: description :type: medkit.core.operation_desc.OperationDescription .. py:class:: PipelineStep Pipeline step describing how a processing operation is connected to other. :Attributes: **operation** : PipelineCompatibleOperation The operation to use at that step **input_keys** : list of str For each input of `operation`, the key to use to retrieve the corresponding annotations (either retrieved from a document or generated by an earlier pipeline step) **output_keys** : list of str For each output of `operation`, the key used to pass output annotations to the next Pipeline step. Can be empty if `operation` doesn't return new annotations. **aggregate_input_keys** : bool, default=False If True, all the annotations from multiple input keys are aggregated in a single list. Defaults to False .. !! processed by numpydoc !! .. py:attribute:: operation :type: PipelineCompatibleOperation .. py:attribute:: input_keys :type: list[str] .. py:attribute:: output_keys :type: list[str] .. py:attribute:: aggregate_input_keys :type: bool :value: False .. py:method:: to_dict() -> dict[str, Any] .. py:class:: Pipeline(steps: list[PipelineStep], input_keys: list[str], output_keys: list[str], name: str | None = None, uid: str | None = None) Graph of processing operations. A pipeline is made of pipeline steps, connecting together different processing operations by the use of input/output keys. Each operation can be seen as a node and the keys are its edge. Two operations can be chained by using the same string as an output key for the first operation and as an input key to the second. Steps must be added in the order of execution, there isn't any sort of dependency detection mechanism. :Parameters: **steps** : list of PipelineStep List of pipeline steps These steps will be executed in the order in which they were added, so make sure to add first the steps generating data used by other steps. **input_keys** : list of str List of keys corresponding to the inputs passed to `run()` **output_keys** : list of str List of keys corresponding to the outputs returned by `run()` **name** : str, optional Name describing the pipeline (defaults to the class name) **uid** : str, optional Identifier of the pipeline .. !! processed by numpydoc !! .. py:property:: description :type: medkit.core.operation_desc.OperationDescription .. py:method:: set_prov_tracer(prov_tracer: medkit.core.prov_tracer.ProvTracer) .. py:method:: run(*all_input_data: list[Any]) -> list[Any] | tuple[list[Any], Ellipsis] | None Run the pipeline. :Parameters: **\*all_input_data** : list of Any Input data expected by the pipeline, must be of same length as the pipeline `input_keys`. For each input key, the corresponding input data must be a list of items than can be of any type. :Returns: list of Any or tuple of list, optional All output data returned by the pipeline, will be of same length as the pipeline `output_keys`. For each output key, the corresponding output will be a list of items that can be of any type. If the pipeline has only one output key, then the corresponding output will be directly returned, not wrapped in a tuple. If the pipeline doesn't have any output key, nothing (ie `None`) will be returned. .. !! processed by numpydoc !! .. py:method:: _perform_step(step: PipelineStep, data_by_key: dict[str, Any]) .. py:method:: _add_provenance(all_output_data: tuple[list[Any], Ellipsis]) .. py:method:: check_sanity()