mapteksdk.workflows.connector_types module

Basic connector type subclasses.

These can be passed to WorkflowArgumentParser.declare_input_connector and WorkflowArgumentParser.declare_output_connector to determine which type of data the connector should accept. The names of these classes match the names of the connectors types as displayed in workflows.

python_type_to_connector_type(python_type: ConnectorType) → ConnectorType

python_type_to_connector_type(python_type: PortType) → PortType

python_type_to_connector_type(python_type: DynamicConnectorType) → DynamicConnectorType

python_type_to_connector_type(python_type: type[PythonPortTypes] | None) → PortType | DynamicConnectorType

Returns the corresponding ConnectorType subclass for a Python type.

This only contains mappings for the ConnectorType subclasses defined in this file.

Parameters:: python_type (Type) – The Python type to match to a basic connector type.
Returns:: The corresponding ConnectorType subclass from this file.
Return type:: ConnectorType
Raises:: KeyError – If there was no corresponding ConnectorType subclass.

python_type_to_dynamic_connector_type(python_type)

Map a Python type to the corresponding dynamic connector type.

Parameters:: python_type (type[PythonPortTypes])
Return type:: DynamicConnectorType

python_type_to_port_type(python_type)

Get the port type for a given Python type.

Parameters:: python_type (type[PythonPortTypes])
Return type:: PortType

class AnyConnectorTypeClass

Bases: DynamicConnectorType

The class of AnyConnectorType.

data_type()

The type of data supported by this port.

Return type:: DataType

from_string(string_value)

Convert a string value to the corresponding python type.

Raises:

TypeError – If string_value is not a supported type.
ValueError – If string_value is the correct type but an invalid value.

Parameters:

string_value (str)

Return type:

str

to_json(value)

Converts the value to a json-serializable value.

This is used to convert python values to json values to be passed to the workflow for output connectors.

Returns:

Json serializable representation of value.

Return type:

json-serializable

Raises:

TypeError – If value is not a supported type.
ValueError – If value is the correct type but an invalid value.

Parameters:

value (str)

to_default_json(value)

Converts the value to a json serializable default.

This allows for specifying a different representation for default values. The output of this function should not include lists.

Overwrite this function to raise an error to indicate that default values are not supported.

By default this calls to_json.

Returns:

String representation of value. None if there is no default value.

Return type:

str

Raises:

TypeError – If value is not a supported type.
ValueError – If value is the correct type but an invalid value.

Parameters:

value (Any)

type_string()

The data type when generating ports.

Return type:: str

class StringConnectorTypeClass

Bases: DynamicConnectorType, PortType

The class of StringConnectorType.

data_type()

The type of data supported by this port.

Return type:: DataType

from_string(string_value)

Convert a string value to the corresponding python type.

Raises:

TypeError – If string_value is not a supported type.
ValueError – If string_value is the correct type but an invalid value.

Parameters:

string_value (str)

Return type:

str

to_json(value)

Converts the value to a json-serializable value.

This is used to convert python values to json values to be passed to the workflow for output connectors.

Returns:

Json serializable representation of value.

Return type:

json-serializable

Raises:

TypeError – If value is not a supported type.
ValueError – If value is the correct type but an invalid value.

Parameters:

value (str)

to_workflow_native(value)

Convert value to the Workflow native format.

Parameters:: value (Any)
Return type:: JsonValues

from_workflow_native(json_value)

Convert from the Workflow native representation to the Python one.

Raises:

TypeError – If json_value is not a supported type.
ValueError – If json_value is the correct type but an invalid value.

Parameters:

json_value (JsonValues)

Return type:

str

to_default_json(value)

Converts the value to a json serializable default.

This allows for specifying a different representation for default values. The output of this function should not include lists.

Overwrite this function to raise an error to indicate that default values are not supported.

By default this calls to_json.

Returns:

String representation of value. None if there is no default value.

Return type:

str

Raises:

TypeError – If value is not a supported type.
ValueError – If value is the correct type but an invalid value.

Parameters:

value (Any)

type_string()

The data type when generating ports.

Return type:: str

class IntegerConnectorTypeClass

Bases: DynamicConnectorType, PortType

Class for IntegerConnectorType.

data_type()

The type of data supported by this port.

Return type:: DataType

from_string(string_value)

Convert a string value to the corresponding python type.

Raises:

TypeError – If string_value is not a supported type.
ValueError – If string_value is the correct type but an invalid value.

Parameters:

string_value (str)

Return type:

int

to_json(value)

Converts the value to a json-serializable value.

This is used to convert python values to json values to be passed to the workflow for output connectors.

Returns:

Json serializable representation of value.

Return type:

json-serializable

Raises:

TypeError – If value is not a supported type.
ValueError – If value is the correct type but an invalid value.

Parameters:

value (int)

to_workflow_native(value)

Convert value to the Workflow native format.

Parameters:: value (Any)
Return type:: JsonValues

from_workflow_native(json_value)

Convert from the Workflow native representation to the Python one.

Raises:

TypeError – If json_value is not a supported type.
ValueError – If json_value is the correct type but an invalid value.

Parameters:

json_value (JsonValues)

Return type:

int

to_default_json(value)

Converts the value to a json serializable default.

This allows for specifying a different representation for default values. The output of this function should not include lists.

Overwrite this function to raise an error to indicate that default values are not supported.

By default this calls to_json.

Returns:

String representation of value. None if there is no default value.

Return type:

str

Raises:

TypeError – If value is not a supported type.
ValueError – If value is the correct type but an invalid value.

Parameters:

value (Any)

type_string()

The data type when generating ports.

Return type:: str

class DoubleConnectorTypeClass

Bases: DynamicConnectorType, PortType

Class for DoubleConnectorType.

data_type()

The type of data supported by this port.

Return type:: DataType

from_string(string_value)

Convert a string value to the corresponding python type.

Raises:

TypeError – If string_value is not a supported type.
ValueError – If string_value is the correct type but an invalid value.

Parameters:

string_value (str)

Return type:

float

to_json(value)

Converts the value to a json-serializable value.

This is used to convert python values to json values to be passed to the workflow for output connectors.

Returns:

Json serializable representation of value.

Return type:

json-serializable

Raises:

TypeError – If value is not a supported type.
ValueError – If value is the correct type but an invalid value.

Parameters:

value (float)

to_workflow_native(value)

Convert value to the Workflow native format.

Parameters:: value (Any)
Return type:: JsonValues

from_workflow_native(json_value)

Convert from the Workflow native representation to the Python one.

Raises:

TypeError – If json_value is not a supported type.
ValueError – If json_value is the correct type but an invalid value.

Parameters:

json_value (JsonValues)

Return type:

float

to_default_json(value)

Converts the value to a json serializable default.

This allows for specifying a different representation for default values. The output of this function should not include lists.

Overwrite this function to raise an error to indicate that default values are not supported.

By default this calls to_json.

Returns:

String representation of value. None if there is no default value.

Return type:

str

Raises:

TypeError – If value is not a supported type.
ValueError – If value is the correct type but an invalid value.

Parameters:

value (Any)

type_string()

The data type when generating ports.

Return type:: str

class BooleanConnectorTypeClass

Bases: DynamicConnectorType, PortType

Class for BooleanConnectorType.

data_type()

The type of data supported by this port.

Return type:: DataType

from_string(string_value)

Convert a string value to the corresponding python type.

Raises:

TypeError – If string_value is not a supported type.
ValueError – If string_value is the correct type but an invalid value.

Parameters:

string_value (str)

Return type:

float

to_json(value)

Converts the value to a json-serializable value.

This is used to convert python values to json values to be passed to the workflow for output connectors.

Returns:

Json serializable representation of value.

Return type:

json-serializable

Raises:

TypeError – If value is not a supported type.
ValueError – If value is the correct type but an invalid value.

Parameters:

value (bool)

to_workflow_native(value)

Convert value to the Workflow native format.

Parameters:: value (Any)
Return type:: JsonValues

from_workflow_native(json_value)

Convert from the Workflow native representation to the Python one.

Raises:

TypeError – If json_value is not a supported type.
ValueError – If json_value is the correct type but an invalid value.

Parameters:

json_value (JsonValues)

Return type:

bool

to_default_json(value)

Converts the value to a json serializable default.

This allows for specifying a different representation for default values. The output of this function should not include lists.

Overwrite this function to raise an error to indicate that default values are not supported.

By default this calls to_json.

Returns:

String representation of value. None if there is no default value.

Return type:

str

Raises:

TypeError – If value is not a supported type.
ValueError – If value is the correct type but an invalid value.

Parameters:

value (Any)

type_string()

The data type when generating ports.

Return type:: str

class CSVStringConnectorTypeClass

Bases: DynamicConnectorType, PortType

Class for CSVStringConnectorType.

data_type()

The type of data supported by this port.

Return type:: DataType

type_string()

The data type when generating ports.

Return type:: str

from_string(string_value)

Convert a string value to the corresponding python type.

Raises:

TypeError – If string_value is not a supported type.
ValueError – If string_value is the correct type but an invalid value.

Parameters:

string_value (str)

Return type:

list

to_json(value)

Converts the value to a json-serializable value.

This is used to convert python values to json values to be passed to the workflow for output connectors.

Returns:

Json serializable representation of value.

Return type:

json-serializable

Raises:

TypeError – If value is not a supported type.
ValueError – If value is the correct type but an invalid value.

Parameters:

value (list)

to_default_json(value)

Converts the value to a json serializable default.

This allows for specifying a different representation for default values. The output of this function should not include lists.

Overwrite this function to raise an error to indicate that default values are not supported.

By default this calls to_json.

Returns:

String representation of value. None if there is no default value.

Return type:

str

Raises:

TypeError – If value is not a supported type.
ValueError – If value is the correct type but an invalid value.

to_workflow_native(value)

Convert value to the Workflow native format.

Parameters:: value (Any)
Return type:: JsonValues

from_workflow_native(json_value)

Convert from the Workflow native representation to the Python one.

Raises:

TypeError – If json_value is not a supported type.
ValueError – If json_value is the correct type but an invalid value.

Parameters:

json_value (JsonValues)

Return type:

list

class DateTimeConnectorTypeClass

Bases: DynamicConnectorType, PortType

Class for DateTimeConnectorType.

data_type()

The type of data supported by this port.

Return type:: DataType

from_string(string_value)

Convert a string value to the corresponding python type.

Raises:

TypeError – If string_value is not a supported type.
ValueError – If string_value is the correct type but an invalid value.

Parameters:

string_value (str)

Return type:

datetime

to_json(value)

Converts the value to a json-serializable value.

This is used to convert python values to json values to be passed to the workflow for output connectors.

Returns:

Json serializable representation of value.

Return type:

json-serializable

Raises:

TypeError – If value is not a supported type.
ValueError – If value is the correct type but an invalid value.

Parameters:

value (datetime | str)

to_default_json(value)

Converts the value to a json serializable default.

This allows for specifying a different representation for default values. The output of this function should not include lists.

Overwrite this function to raise an error to indicate that default values are not supported.

By default this calls to_json.

Returns:

String representation of value. None if there is no default value.

Return type:

str

Raises:

TypeError – If value is not a supported type.
ValueError – If value is the correct type but an invalid value.

Parameters:

value (datetime)

to_workflow_native(value)

Convert value to the Workflow native format.

Parameters:: value (datetime.datetime | str)
Return type:: JsonValues

from_workflow_native(json_value)

Convert from the Workflow native representation to the Python one.

Raises:

TypeError – If json_value is not a supported type.
ValueError – If json_value is the correct type but an invalid value.

Parameters:

json_value (JsonValues)

Return type:

datetime.datetime

type_string()

The data type when generating ports.

Return type:: str

class Point3DConnectorTypeClass

Bases: DynamicConnectorType, PortType

Class for Point3DConnectorType.

data_type()

The type of data supported by this port.

Return type:: DataType

from_string(string_value)

Convert a string value to the corresponding python type.

Raises:

TypeError – If string_value is not a supported type.
ValueError – If string_value is the correct type but an invalid value.

Parameters:

string_value (str)

Return type:

numpy.ndarray

to_json(value)

Converts the value to a json-serializable value.

This is used to convert python values to json values to be passed to the workflow for output connectors.

Returns:

Json serializable representation of value.

Return type:

json-serializable

Raises:

TypeError – If value is not a supported type.
ValueError – If value is the correct type but an invalid value.

Parameters:

value (numpy.ndarray)

to_workflow_native(value)

Convert value to the Workflow native format.

Parameters:: value (Any)
Return type:: list[float]

from_workflow_native(json_value)

Convert from the Workflow native representation to the Python one.

Raises:

TypeError – If json_value is not a supported type.
ValueError – If json_value is the correct type but an invalid value.

Parameters:

json_value (JsonValues)

Return type:

np.ndarray

to_default_json(value)

Converts the value to a json serializable default.

This allows for specifying a different representation for default values. The output of this function should not include lists.

Overwrite this function to raise an error to indicate that default values are not supported.

By default this calls to_json.

Returns:

String representation of value. None if there is no default value.

Return type:

str

Raises:

TypeError – If value is not a supported type.
ValueError – If value is the correct type but an invalid value.

Parameters:

value (Any)

type_string()

The data type when generating ports.

Return type:: str

class FileSystemConnectorTypeClass(type_string)

Bases: DynamicConnectorType, PortType

Connector type for file and directory connector types.

Parameters:: type_string (Literal['File'] | typing.Literal['Folder'])

data_type()

The type of data supported by this port.

Return type:: DataType

from_string(string_value)

Convert a string value to the corresponding python type.

Raises:

TypeError – If string_value is not a supported type.
ValueError – If string_value is the correct type but an invalid value.

Parameters:

string_value (str)

Return type:

Path

to_json(value)

Converts the value to a json-serializable value.

This is used to convert python values to json values to be passed to the workflow for output connectors.

Returns:

Json serializable representation of value.

Return type:

json-serializable

Raises:

TypeError – If value is not a supported type.
ValueError – If value is the correct type but an invalid value.

Parameters:

value (PathLike | str)

to_workflow_native(value)

Convert value to the Workflow native format.

Parameters:: value (Any)
Return type:: JsonValues

from_workflow_native(json_value)

Convert from the Workflow native representation to the Python one.

Raises:

TypeError – If json_value is not a supported type.
ValueError – If json_value is the correct type but an invalid value.

Parameters:

json_value (JsonValues)

Return type:

pathlib.Path

to_default_json(value)

Converts the value to a json serializable default.

This allows for specifying a different representation for default values. The output of this function should not include lists.

Overwrite this function to raise an error to indicate that default values are not supported.

By default this calls to_json.

Returns:

String representation of value. None if there is no default value.

Return type:

str

Raises:

TypeError – If value is not a supported type.
ValueError – If value is the correct type but an invalid value.

Parameters:

value (Any)

type_string()

The data type when generating ports.

Return type:: str

AnyConnectorType = <mapteksdk.workflows.connector_types.AnyConnectorTypeClass object>

Connector type representing no connector type set.

This corresponds to the connector type being blank on the workflows side. Input connectors of this type will accept any value from other connectors and the string representation of that value will be returned. Output connectors of this type will accept any value which can be converted into a string.

StringConnectorType = <mapteksdk.workflows.connector_types.StringConnectorTypeClass object>

Connector type corresponding to String on the workflows side.

This can be passed to declare_input_connector or declare_output_connector to declare the connector type as String. Passing the python type str is equivalent to passing this class.

Examples

This example sets the output connector “reversed” to contain a reversed version of the string from the input connector “string”

>>> from mapteksdk.workflows import (WorkflowArgumentParser,
...                                  StringConnectorType)
>>> parser = WorkflowArgumentParser()
>>> parser.declare_input_connector("string", StringConnectorType)
>>> parser.declare_output_connector("reversed", StringConnectorType)
>>> parser.parse_arguments()
>>> parser.set_output("reversed", parser["string"][::-1])

IntegerConnectorType = <mapteksdk.workflows.connector_types.IntegerConnectorTypeClass object>

Connector type corresponding to Integer on the workflows side.

Passing the connector type as int is equivalent to passing this to declare_input/output_connector.

Examples

This example creates a workflow component with an Integer input and output connector. The output connector “new_count” is set to the value of the input connector “count” plus one.

>>> from mapteksdk.workflows import (WorkflowArgumentParser,
...                                  IntegerConnectorType)
>>> parser = WorkflowArgumentParser()
>>> parser.declare_input_connector("count", IntegerConnectorType)
>>> parser.declare_output_connector("new_count", IntegerConnectorType)
>>> parser.parse_arguments()
>>> parser.set_output("new_count", parser["count"] += 1)

DoubleConnectorType = <mapteksdk.workflows.connector_types.DoubleConnectorTypeClass object>

Connector type corresponding to Double on the workflows side.

Passing the connector type as float is equivalent to passing this to declare_input/output_connector.

Examples

This example sets the value of the output connector “x_over_2” to the value of the input connector “x” divided by two.

>>> from mapteksdk.workflows import (WorkflowArgumentParser,
...                                  DoubleConnectorType)
>>> parser = WorkflowArgumentParser()
>>> parser.declare_input_connector("x", DoubleConnectorType)
>>> parser.declare_output_connector("x_over_2", DoubleConnectorType)
>>> parser.parse_arguments()
>>> parser.set_output("x_over_2", parser["x"] / 2)

BooleanConnectorType = <mapteksdk.workflows.connector_types.BooleanConnectorTypeClass object>

Connector type corresponding to Boolean on the workflows side.

Passing the connector type as bool is equivalent to passing this to declare_input/output_connector.

Examples

This example sets the output connector “not x” to be the inverse of the value passed to the “x” input connector.

>>> from mapteksdk.workflows import (WorkflowArgumentParser,
...                                  IntegerConnectorType)
>>> parser = WorkflowArgumentParser()
>>> parser.declare_input_connector("x", BooleanConnectorType)
>>> parser.declare_output_connector("not x", BooleanConnectorType)
>>> parser.parse_arguments()
>>> parser.set_output("not x", not parser["x"])

CSVStringConnectorType = <mapteksdk.workflows.connector_types.CSVStringConnectorTypeClass object>

Connector type corresponding to CSV String on the workflows side.

Passing the connector type as list is equivalent to passing this to declare_input/output_connector.

Examples

This example filters out every second element in the list from the input connector “values” and sets the filtered list to the output connector “second_values”.

>>> from mapteksdk.workflows import (WorkflowArgumentParser,
...                                  CSVStringConnectorType)
>>> parser = WorkflowArgumentParser()
>>> parser.declare_input_connector("values", CSVStringConnectorType)
>>> parser.declare_output_connector("second_values", CSVStringConnectorType)
>>> parser.parse_arguments()
>>> parser.set_output("second_values", parser["values"][::2])

DateTimeConnectorType = <mapteksdk.workflows.connector_types.DateTimeConnectorTypeClass object>

Connector type corresponding to DateTime on the Workflows side.

Passing the connector type as datetime.datetime is equivalent to passing this to declare_input/output_connector.

This does not currently support defaults.

Examples

This example adds 24 hours to the time from the input connector “today” and sets that time to the output connector “tomorrow”. Note that this may not give the same time on the next day due to daylight savings start/ending.

>>> import datetime
>>> from mapteksdk.workflows import (WorkflowArgumentParser,
...                                  DateTimeConnectorType)
>>> parser = WorkflowArgumentParser()
>>> parser.declare_input_connector("today", DateTimeConnectorType)
>>> parser.declare_output_connector("tomorrow", DateTimeConnectorType)
>>> parser.parse_arguments()
>>> tomorrow = parser["today"] + datetime.timedelta(days=1)
>>> parser.set_output("tomorrow", tomorrow)

Point3DConnectorType = <mapteksdk.workflows.connector_types.Point3DConnectorTypeClass object>

Connector type representing a 3D point in workflows.

An input connector of this type will return a numpy array of floats with shape (3, ) representing the point in the form [X, Y, Z].

Default values can be specified using any iterable as long as its length is three and all values can be converted to floats, though list or numpy arrays are generally preferable.

Given a script called “script.py” with an input connector of type Point3DConnectorType called “point”, to pass the point [1.2, 3.4, -1.3] via the command line you would type:

>>> py script.py --point=(1.2,3.4,-1.3)

Examples

This example sets the output connector “inverted_point” to the inverse of the point from the input connector “point”.

>>> from mapteksdk.workflows import (WorkflowArgumentParser,
...                                  Point3DConnectorType)
>>> parser = WorkflowArgumentParser()
>>> parser.declare_input_connector("point", Point3DConnectorType)
>>> parser.declare_output_connector("inverted_point", Point3DConnectorType)
>>> parser.parse_arguments()
>>> parser.set_output("inverted_point", -parser["point"])

FileConnectorType = <mapteksdk.workflows.connector_types.FileSystemConnectorTypeClass object>

Connector type representing a File.

If used as an input, the file path will be returned as a pathlib.Path object.

If used as an output, the file path can be provided as any path-like object.

Warning

This class does not validate the file paths received from a Workflow, nor does it validate file paths provided to output connectors. This means:

The file path given by an input connector may not point to an existing file.
The file path given by an input connector may be the path to a directory instead of a file.
An output connector will not raise an error if given the path to a file which does not exist.
An output connector will not raise an error if given the path to a directory.

Notes

When declaring connectors, pathlib.Path can be used instead of this class.

Examples

The following example demonstrates a script which accepts a file path from a workflow and then reads the CSV file at that file path using pandas.

>>> import pandas
>>> import pathlib
>>> from mapteksdk.workflows import WorkflowArgumentParser, FileConnectorType
>>> if __name__ == "__main__":
...   parser = WorkflowArgumentParser(
...     description="Example of reading a CSV file.")
...   parser.declare_input_connector(
...     "csv_file",
...     FileConnectorType,
...     description="Path to the CSV file to read."
...   )
...   parser.parse_arguments()
...   csv_file_path: pathlib.Path = parser["csv_file"]
...   if not csv_file_path.exists():
...     raise FileNotFoundError(
...       f"The CSV file does not exist: '{csv_file_path}'"
...     )
...   if csv_file_path.is_dir():
...     raise ValueError(
...       f"The CSV file cannot be a directory."
...     )
...   csv_data = pandas.read_csv(csv_file_path)
...   # Now do something with the CSV.

DirectoryConnectorType = <mapteksdk.workflows.connector_types.FileSystemConnectorTypeClass object>

Connector type representing a directory.

If used as an input, the path to the directory is returned as a pathlib.Path object.

If used as an output, the directory path can be provided as any path-like object.

Warning

This class does not validate the directory paths received from a Workflow, nor does it validate directory paths provided to output connectors. This means:

The directory path given by an input connector may not point to an existing directory.
The directory path given by an input connector may be the path to a file instead of a directory.
An output connector will not raise an error if given the path to a directory which does not exist.
An output connector will not raise an error if given the path to a file.

Examples

The following example demonstrates a workflow which accepts a directory input and produces an output list which contains every file within that directory.

>>> import pathlib
>>> from mapteksdk.workflows import (
...   WorkflowArgumentParser, DirectoryConnectorType)
>>> if __name__ == "__main__":
...   parser = WorkflowArgumentParser(
...     description="This script accepts a directory input from workflows. "
...       "It outputs a list containing all of the files within that "
...       "directory. This script will fail if the file input does not exist "
...       "or the user does not have permissions to access that directory."
...   )
...   parser.declare_input_connector(
...     "directory",
...     DirectoryConnectorType,
...     description="Path to the directory to list the files it contains."
...   )
...   parser.declare_output_connector(
...     "files_in_directory",
...     list,
...     description="Path to the files contained in the input directory."
...   )
...   parser.parse_arguments()
...   directory: pathlib.Path = parser["directory"]
...   # :NOTE: is_dir() will return false if the directory does not exist.
...   # Thus check if the path exists before checking if it is a directory.
...   if not directory.exists():
...     raise FileNotFoundError(
...       f"The input directory did not exist: '{directory}'"
...     )
...   if not directory.is_dir():
...     raise ValueError(
...       f"The input directory path pointed to a file: {directory}"
...     )
...   files_in_directory = [str(path) for path in directory.iterdir()]
...   parser.set_output("files_in_directory", files_in_directory)

data_type_to_port_type(data_type)

Get the port type which can handle the data_type.

Parameters:: data_type (DataType)
Return type:: PortType