Stream manipulation

Manipulate MorphoCut streams and show diagnostic information.

morphocut.stream.Enumerate(start: int = 0)[source]

Enumerate objects in the Stream.

Parameters:: start (int, default 0) – Start value of the counter.
Returns:: Variable[int] – Index (from start).

class morphocut.stream.Filter(predicate)[source]

Stream Filter objects in the Stream.

After this node, the stream will only contain objects for which function evaluates to True.

Parameters:: predicate (Variable or callable) – If the predicate is true, the object will stay in the stream. If a callable, it will receive a StreamObject and must return a bool.

Example

with Pipeline() as p:
    a = Unpack([1,2,3])
    # The stream now consists of three objects:
    # {a: 1}, {a: 2}, {a: 3}

    # Keep only objects where a>2
    Filter(a>2)
    # The stream now consists of 1 object:
    # {a: 3}

    ## OR:
    # Keep only objects where a>2.
    # Here, `obj` is the current StreamObject.
    # This form might be preferred for performance reasons.
    Filter(lambda obj: obj[a] > 2)

transform_stream(stream)[source]: Transform a stream. By default, this calls self.transform with appropriate parameters. transform has to be implemented by a subclass if transform_stream is not overridden. Override if the stream has to be altered in some way, i.e. objects are created, deleted or re-arranged.

morphocut.stream.FilterVariables(*variables)[source]

Stream Only keep the specified Variables in the stream.

This might reduce memory usage and speed up processing, especially when StreamObjects have to be sent to other processes.

morphocut.stream.Pack(size, *variables)[source]

Stream Pack values of subsequent objects in the stream into one tuple.

Parameters:

size (int or Variable) – Number of objects to aggregate.
*variables (Variable) – Variables to pack.

Returns:

One or more Variable – One output Variable per input Variable.

Example

with Pipeline() as p:
    a = Unpack([1,2,3])
    # The stream now consists of three objects:
    # {a: 1}, {a: 2}, {a: 3}
    a123 = Pack(3, a)
    # The stream now consists one object:
    # {a: 1, a123: (1,2,3)}

See also

Unpack

morphocut.stream.PrintObjects(*args: Variable)[source]

Print the contents of StreamObjects.

For debugging purposes only.

Parameters:: *args (Variable) – Variables to display.

morphocut.stream.Progress(description: str | Variable[str] | None = None, monitor_interval=None)[source]

Show a dynamically updating progress bar using tqdm.

Parameters:: description (str) – Description of the progress bar.

Example

with Pipeline() as pipeline:
    Progress("Description")

Output: Description|███████████████████████| [00:00, 2434.24it/s]

morphocut.stream.Slice(*args: int | None)[source]

Stream Slice the Stream.

Filter objects in the Stream based on their index.

Parameters:

start (int, optional) – Skip this many objects upfront.
stop (int, optional) – Stop at this index.
step (int, optional) – Skip this many objects in every step.

morphocut.stream.StreamBuffer(maxsize: int)[source]

Stream Buffer a stream to improve throughput by decoupling data production from data consumption.

This allows for continued processing of data while I/O-bound nodes are waiting for data. Buffering the stream enables asynchronous prefetching and minimizes the time spent waiting for data to be available.

Parameters:: maxsize (int) – The maximum number of items to buffer. If the buffer is full, further data production is paused until items are consumed.

Example

with Pipeline() as pipeline:
    # Produce data (preferably I/O-bound)
    ...
    StreamBuffer(maxsize=100)
    # Process data
    ...

morphocut.stream.Unpack(collection: Collection | Variable[Collection])[source]

Stream Unpack values from a collection into the Stream.

The result is basically the cross-product of the stream with the iterable.

Parameters:: collection (Collection or Variable) – An iterable to unpack.
Returns:: Variable – One value from the iterable.

Example

with Pipeline() as p:
    a = Unpack([1,2,3])
    # The stream now consists of three objects:
    # {a: 1}, {a: 2}, {a: 3}
    b = Unpack([1,2,3])
    # The stream now consists of nine objects:
    # {a: 1, b: 1}, {a: 1, b: 2}, {a: 1, b: 3},
    # {a: 2, b: 1}, {a: 2, b: 2}, {a: 2, b: 3},
    # {a: 3, b: 1}, {a: 3, b: 2}, {a: 3, b: 3}