Pipeline
- class ioiocore.pipeline.Pipeline(directory=None)[source]
Bases:
InterfaceA class representing a pipeline, inheriting from Interface. Manages nodes and their connections within the pipeline.
Supports both Node types (INode, ONode, IONode) and Chain types (IChain, OChain, IOChain).
- Parameters:
directory (str)
- __init__(directory=None)[source]
Initializes the pipeline.
- Parameters:
directory (
str) – Directory to be used for the pipeline (default is None).
- add_error_handler(handler)[source]
Register a callable invoked once when the run fails.
A failure is detected asynchronously, after start() has already returned, so there is nothing for start() to raise. Without a handler the only report is printed to the console – which a GUI application may never show. Use this to put the failure in front of the user, and raise_if_failed() where an exception is wanted instead.
The handler runs on the monitoring thread, so it should hand the news to whatever owns the user interface rather than doing work itself. An exception raised inside it is logged and swallowed: that thread is the only channel that reports failures and must not die reporting one.
Handlers run after get_condition() reports ERROR and after stop(), even one that raised, so whoever reads what a handler recorded sees the run over. The converse does not hold: a caller polling get_condition() can see ERROR before any handler has run; get_last_error() already returns the entry by then.
- Parameters:
handler (Callable) – Called with the failing log entry.
- Return type:
None
- add_node(node)[source]
Adds a node or chain to the pipeline.
- Parameters:
node (
ProcessingElement) – The node or chain to add to the pipeline.
- close()[source]
Dispose the pipeline and release logging resources.
Stops the pipeline if it is still running, then shuts down the logger (flushing pending entries, closing the log file and joining its writer thread). The logger survives ordinary start()/stop() cycles, so call close() (or use the pipeline as a context manager) once the pipeline is no longer needed. Safe to call more than once.
- static deserialize(data)[source]
Deserializes the pipeline from a dictionary.
- Parameters:
data (
dict) – A dictionary containing the serialized pipeline data.- Returns:
The deserialized pipeline object.
- Return type:
- get_backlog()[source]
Returns the maximum input-queue backlog across the pipeline.
This is the number of items queued on the most-backed-up input port of any element (including nodes inside chains). A steadily growing value indicates a consumer that cannot keep up with its input rate.
- Returns:
The maximum number of queued items on any input port.
- Return type:
int
- get_condition()[source]
Returns the current condition of the pipeline.
- Returns:
The current condition of the pipeline.
- Return type:
- get_elapsed_time()[source]
Returns the elapsed time since the pipeline started.
- Returns:
The elapsed time in seconds.
- Return type:
float
- get_last_error()[source]
Returns the last error recorded by the pipeline.
- Returns:
The failing entry, carrying the message, the node and the source location, or None if nothing has failed. Note this is an entry rather than a string: the annotation previously said str, which was wrong and hid the node and location a reader needs.
- Return type:
LogEntry or None
- get_state()[source]
Returns the current state of the pipeline.
- Returns:
The current state of the pipeline.
- Return type:
- is_logging_persistent()[source]
Whether the pipeline writes a persistent log file.
- Returns:
False if the log file could not be created and logging is in-memory only (no persistent audit trail).
- Return type:
bool
- log(msg, type=None)[source]
Writes a log entry to the pipeline.
- Parameters:
entry (str) – The log entry to write.
msg (str | Exception)
type (LogTypes)
- raise_if_failed()[source]
Raise if the run has failed, naming the node and the cause.
For a script that would otherwise carry on past a dead pipeline.
- Raises:
RuntimeError – If the pipeline is in an error condition.
- Return type:
None