aicssegmentation.workflow package

Submodules

aicssegmentation.workflow.batch_workflow module

class aicssegmentation.workflow.batch_workflow.BatchWorkflow(workflow_definition: aicssegmentation.workflow.workflow_definition.WorkflowDefinition, input_dir: Union[str, pathlib.Path], output_dir: Union[str, pathlib.Path], channel_index: int = 0)

Bases: object

Represents a batch of workflows to process. This class provides the functionality to run batches of workflows using multiple image inputs from a input directory according to the steps defined in its WorkflowDefinition.

execute_all()

execute_next()

property failed_files

property input_dir

is_done() → bool

Indicates whether all files / steps have been executed

Use this to know when the batch workflow is complete if manually executing the workflow with execute_next()

Returns:

(bool): True if all files/steps have been executed, False if not

property output_dir

property processed_files

property total_files

write_log_file_summary()

Write a log file to the output folder.

aicssegmentation.workflow.segmenter_function module

class aicssegmentation.workflow.segmenter_function.FunctionParameter(name: str, widget_type: aicssegmentation.workflow.segmenter_function.WidgetType, data_type: str, min_value: Union[int, float] = None, max_value: Union[int, float] = None, increment: Union[int, float] = None, options: List[str] = None)

Bases: object

Represents an input parameter to a segmenter function

data_type: str = None

increment: Union[int, float] = None

max_value: Union[int, float] = None

min_value: Union[int, float] = None

name: str = None

options: List[str] = None

widget_type: WidgetType = None

class aicssegmentation.workflow.segmenter_function.SegmenterFunction(name: str, display_name: str, function: str, module: str, parameters: Dict[str, List[aicssegmentation.workflow.segmenter_function.FunctionParameter]] = None)

Bases: object

Represents an aicssegmentation function. Functions are the smallest executable entity in a workflow and directly map to a python callable

display_name: str = None

function: str = None

module: str = None

name: str = None

parameters: Dict[str, List[aicssegmentation.workflow.segmenter_function.FunctionParameter]] = None

class aicssegmentation.workflow.segmenter_function.WidgetType

Bases: enum.Enum

An enumeration.

DROPDOWN = 'drop-down'

SLIDER = 'slider'

static from_str(value: str)

aicssegmentation.workflow.workflow module

class aicssegmentation.workflow.workflow.Workflow(workflow_definition: aicssegmentation.workflow.workflow_definition.WorkflowDefinition, input_image: numpy.ndarray)

Bases: object

Represents an executable aics-segmentation workflow This class provides the functionality to run a workflow using an image input according to the steps defined in its WorkflowDefinition.

execute_all() → numpy.ndarray

Execute all steps in the Workflow Note: default parameters will be used to execute the steps. To execute a step

with user-provided parameters, use execute_next()

Params:

none

Returns:

(np.ndarray): Result of the final WorkflowStep.

execute_next(parameters: Dict[str, Any] = None) → numpy.ndarray

Execute the next workflow step.

Params:

parameters: Optional dictionary of parameter inputs to use when executing the step

If parameters are not provided, the step’s default parameters will be used

Returns:

result (np.ndarray): resultant image from running the next workflow step

execute_step(i: int, parameters: Dict[str, Any], selected_image: List[napari.layers.image.image.Image]) → numpy.ndarray

Args:

i: step number (0 indexed) that you want to run

Returns:

get_most_recent_result() → numpy.ndarray

Get the result from the last executed WorkflowStep.

Params:

none

Returns:

(np.ndarray): Result of the last executed WorkflowStep,

returns the starting image if no Workflowsteps have been run.

get_next_step() → aicssegmentation.workflow.workflow_step.WorkflowStep

Get the next step to be performed

Params:

none

Returns:

(WorkflowStep): next WorkflowStep object to perform on image None if all steps have already been executed

get_result(step_index: int) → numpy.ndarray

Get the result image for a workflow step.

You must call execute() on the workflow step in order to produce a result first before calling this function.

Params:

step_index (int): index of the WorkflowStep in the workflowengine to get the result image of.

Returns:

self.image (np.ndarray): Result of performing workflow step

on the given image None if step has not been executed yet.

is_done() → bool

Check if all WorkflowSteps have been executed.

Params:

none

Returns:

(bool): True if all WorkflowSteps have been executed, False if not

reset()

Reset the workflow so it can be run again

property workflow_definition

aicssegmentation.workflow.workflow_config module

exception aicssegmentation.workflow.workflow_config.ConfigurationException

Bases: Exception

Raised when errors are encountered reading from Configuration files

class aicssegmentation.workflow.workflow_config.WorkflowConfig

Bases: object

Provides access to structure workflow configuration

get_all_functions() → List[aicssegmentation.workflow.segmenter_function.SegmenterFunction]

Get the list of all available Functions from configuration

get_available_workflows() → List[str]

Get the list of all workflows available through configuration

get_workflow_definition(workflow_name: str) → aicssegmentation.workflow.workflow_definition.PrebuiltWorkflowDefinition

Get a WorkflowDefinition for the given workflow from the corresponding prebuilt json structure config

get_workflow_definition_from_config_file(file_path: pathlib.Path, workflow_name: str = None, prebuilt: bool = False) → aicssegmentation.workflow.workflow_definition.WorkflowDefinition

Get a WorkflowDefinition based off the given json configuration file

save_workflow_definition_as_json(workflow_definition: aicssegmentation.workflow.workflow_definition.WorkflowDefinition, output_file_path: pathlib.Path)

Save a WorkflowDefinition as a json config file

aicssegmentation.workflow.workflow_definition module

class aicssegmentation.workflow.workflow_definition.PrebuiltWorkflowDefinition(name: str, steps: List[aicssegmentation.workflow.workflow_step.WorkflowStep])

Bases: aicssegmentation.workflow.workflow_definition.WorkflowDefinition

Definition of a pre-built(default) aics-segmentation Workflow from our assets.

This class only defines the workflow (i.e. the workflow characteristics and steps) and is used either for building an executable Workflow object or to access information about the Workflow without needing to execute it

property diagram_image

name = None

steps = None

property thumbnail_post

property thumbnail_pre

class aicssegmentation.workflow.workflow_definition.WorkflowDefinition(name: str, steps: List[aicssegmentation.workflow.workflow_step.WorkflowStep])

Bases: object

Definition of a custom aics-segmentation Workflow loaded from file.

This class only defines the workflow (i.e. the workflow characteristics and steps) and is used either for building an executable Workflow object or to access information about the Workflow without needing to execute it

name: str = None

steps: List[WorkflowStep] = None

aicssegmentation.workflow.workflow_engine module

class aicssegmentation.workflow.workflow_engine.WorkflowEngine(workflow_config: aicssegmentation.workflow.workflow_config.WorkflowConfig = None)

Bases: object

aicssegmentation workflow engine Use this class to access and execute aicssegmentation structure workflows

get_executable_batch_workflow(workflow_name: str, input_dir: str, output_dir: str, channel_index: int = 0)

Get an executable BatchWorkflow object

inputs:

workflow_name (str): Name of the workflow to load input_dir (str|Path): Directory containing input files for the batch processing output_dir (str|Path): Output directory for the batch processing channel_index (int): Index of the channel to process in each image (usually a structure channel)

get_executable_batch_workflow_from_config_file(file_path: Union[str, pathlib.Path], input_dir: Union[str, pathlib.Path], output_dir: Union[str, pathlib.Path], channel_index: int = 0)

Get an executable batch workflow object from a configuration file

inputs:

file_path (str|Path): Path to the workflow configuration file input_dir (str|Path): Directory containing input files for the batch processing output_dir (str|Path): Output directory for the batch processing channel_index (int): Index of the channel to process in each image (usually a structure channel)

get_executable_workflow(workflow_name: str, input_image: numpy.ndarray) → aicssegmentation.workflow.workflow.Workflow

Get an executable workflow object

inputs:

workflow_name (str): Name of the workflow to load input_image (ndarray): input image for the workflow to execute on

get_executable_workflow_from_config_file(file_path: Union[str, pathlib.Path], input_image: numpy.ndarray) → aicssegmentation.workflow.workflow.Workflow

Get an executable workflow object from a configuration file

inputs:

file_path (str|Path): Path to the workflow configuration file input_image (ndarray): input image for the workflow to execute on

save_workflow_definition(workflow_definition: aicssegmentation.workflow.workflow_definition.WorkflowDefinition, output_file_path: Union[str, pathlib.Path])

property workflow_definitions

List of all workflow definitions

aicssegmentation.workflow.workflow_step module

class aicssegmentation.workflow.workflow_step.WorkflowStep(category: aicssegmentation.workflow.workflow_step.WorkflowStepCategory, function: aicssegmentation.workflow.segmenter_function.SegmenterFunction, step_number: int, parent: List[int], parameter_values: Dict[str, List] = None)

Bases: object

Represents a single step in an aicssegmentation Workflow

category: WorkflowStepCategory = None

execute(input_images: List[numpy.ndarray], parameters: Dict[str, Any] = None) → numpy.ndarray

Execute this workflow step on the given input image and return the result.

Params:

input_images (List[np.ndarray]): List of image inputs to perform this

workflow step on, generally parent image

parameters (Dict): Dictionary of parameters to pass to the

underlying function

Returns:

self.result (np.ndarray): Result of performing workflow step

on the given image.

function: SegmenterFunction = None

property name

parameter_values: Dict[str, List] = None

parent: List[int] = None

step_number: int = None

class aicssegmentation.workflow.workflow_step.WorkflowStepCategory

Bases: enum.Enum

An enumeration.

CORE = 'core'

POST_PROCESSING = 'postprocessing'

PRE_PROCESSING = 'preprocessing'

static from_str(value: str)

Module contents