hyped.core.module module

Defines the base class for creating data processing modules.

This module provides the DataFlowModule class, which serves as an abstract base class for defining modular data processing workflows. It simplifies the creation of complex data flows by encapsulating processing logic and automatically handling data flow construction.

class hyped.core.module.DataFlowModule(debug: bool = True)[source]

Bases: ABC

Base class for defining reusable data processing modules.

A DataFlowModule encapsulates a data processing workflow. Subclasses must implement the call() method, which defines the core logic of the module.

abstractmethod call(*args: Feature | Any | list[Any] | Scalar | Array, **kwargs: Feature | Any | list[Any] | Scalar | Array) → Feature | Any | list[Any] | Scalar | Array[source]

Defines the core logic of the data processing module.

Subclasses must implement this method to specify how input features are processed to produce output features. The signature of this method determines the input features of the data flow, and the return type determines the output feature.

Within this method, the flow property provides access to a DataFlow instance, giving access to functionality like ~.flow.DataFlow.collect.

Parameters:

  • *args (Feature) – Positional input features.

  • **kwargs (Feature) – Keyword input features.

Returns:

The output feature produced by the module.

Return type:

Feature

property flow: DataFlow | ExecutableDataFlow

The executable data flow for this module.

This property lazily builds and returns the data flow instance.

  • Within the call() method, this returns a mutable DataFlow instance, giving access to functionality like ~.flow.DataFlow.collect.

  • Outside of call(), this returns the build ExecutableDataFlow, which represents the fully constructed and optimized data flow, ready for execution.

Returns:

The data flow.

Return type:

DataFlow | ExecutableDataFlow

plot(node_format: str | Template = "[{{ node_id[:4] }}] {% if node_type == 'SOURCE_NODE' %}Source{% else %}{{ node_object }}{% endif %}", with_edge_labels: bool = True, edge_font_size: int = 6, node_font_size: int = 6, node_size: int = 5000, arrowsize: int = 25, color_map: dict[NodeType, str] = {}, legend: bool = True, legend_fontsize: int = 6, ax: None | Axes = None) → Axes[source]

Plot a data flow graph.

Parameters:

  • flow (DataFlow) – The data flow to plot.

  • node_format (str | Template) – The jinja template used to generate node labels.

  • with_edge_labels (bool) – Whether to include labels on the edges. Defaults to True.

  • edge_font_size (int) – The font size for edge labels. Defaults to 6.

  • node_font_size (int) – The font size for node labels. Defaults to 6.

  • node_size (int) – The size of the nodes. Defaults to 5_000.

  • arrowsize (int) – The size of the arrows on the edges. Defaults to 25.

  • color_map (dict[None | type, str]) – indicate custom color scheme based on the processor type. None refers to the source node.

  • legend (bool) – Whether to add a legend of the node types to the axes. Defaults to True.

  • legend_fontsize (int) – The font size for the legend. Defaults to 6.

  • ax (Optional[plt.Axes]) – Matplotlib axes object to draw the plot on. Defaults to None.

Returns:

The Matplotlib axes object with the plot.

Return type:

plt.Axes