medkit.core.pipeline

medkit.core.pipeline#

Classes#

`PipelineCompatibleOperation`	Base class for protocol classes.
`ProvCompatibleOperation`	Base class for protocol classes.
`DescribableOperation`	Base class for protocol classes.
`PipelineStep`	Pipeline step describing how a processing operation is connected to other.
`Pipeline`	Graph of processing operations.

Module Contents#

class medkit.core.pipeline.PipelineCompatibleOperation#

Bases: 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:
        ...

run(*all_input_data: list[Any]) → list[Any] | tuple[list[Any], Ellipsis] | None#

Run the operation.

Parameters:

all_input_datalist 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.

class medkit.core.pipeline.ProvCompatibleOperation#

Bases: 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:
        ...

set_prov_tracer(prov_tracer: medkit.core.prov_tracer.ProvTracer)#

class medkit.core.pipeline.DescribableOperation#

Bases: 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:
        ...

description: medkit.core.operation_desc.OperationDescription#

class medkit.core.pipeline.PipelineStep#

Pipeline step describing how a processing operation is connected to other.

Attributes:

operationPipelineCompatibleOperation: The operation to use at that step
input_keyslist 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_keyslist 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_keysbool, default=False: If True, all the annotations from multiple input keys are aggregated in a single list. Defaults to False

operation: PipelineCompatibleOperation#

input_keys: list[str]#

output_keys: list[str]#

aggregate_input_keys: bool = False#

to_dict() → dict[str, Any]#

class medkit.core.pipeline.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:

stepslist 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_keyslist of str: List of keys corresponding to the inputs passed to run()
output_keyslist of str: List of keys corresponding to the outputs returned by run()
namestr, optional: Name describing the pipeline (defaults to the class name)
uidstr, optional: Identifier of the pipeline

uid: str#

name: str | None#

steps: list[PipelineStep]#

input_keys: list[str]#

output_keys: list[str]#

_prov_tracer: medkit.core.prov_tracer.ProvTracer | None = None#

_sub_prov_tracer: medkit.core.prov_tracer.ProvTracer | None = None#

property description: medkit.core.operation_desc.OperationDescription#

set_prov_tracer(prov_tracer: medkit.core.prov_tracer.ProvTracer)#

run(*all_input_data: list[Any]) → list[Any] | tuple[list[Any], Ellipsis] | None#

Run the pipeline.

Parameters:

*all_input_datalist 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.

_perform_step(step: PipelineStep, data_by_key: dict[str, Any])#

_add_provenance(all_output_data: tuple[list[Any], Ellipsis])#

check_sanity()#