Working with Streams

An imfusion.Stream is derived from imfusion.Data, so it can be returned by and passed into algorithms as an imfusion.Data.

For example, we can create a fake image stream like this:

from imfusion import stream

fakeImgStream = stream.FakeImageStream()

>>> fakeImgStream

<imfusion.stream.FakeImageStream object at 0x...>

To access data from the stream, we have two options with different use cases:

Query the stream for the next sample using imfusion.stream.SynchronousConsumer - for occasional, blocking reads.
Record the stream using imfusion.stream.StreamRecorderAlgorithm - for longer captures or when you need to ensure all data is captured.

Blocking access to the next sample

For scripts or tools that need occasional access to the next piece of stream data, use imfusion.stream.SynchronousConsumer. It connects to a Stream when you enter a context manager and disconnects when you leave.

Call imfusion.stream.SynchronousConsumer.get_next_data() with an optional timeout in seconds. The method blocks until a sample arrives or is cancelled otherwise.

Example:

with stream.SynchronousConsumer(fakeImgStream) as cap:
    frame = cap.get_next_data(1.0)  # wait up to one second

>>> frame

imfusion.SharedImageSet(size: ..., [
    imfusion.SharedImage(UBYTE width: 100 height: 100),
])

This can also be done in a loop:

with stream.SynchronousConsumer(fakeImgStream) as cap:
    while True:
        try:
            frame = cap.get_next_data(1.0)  # wait up to one second
            if frame is not None:
                # frame is e.g. imfusion.SharedImageSet
                print(frame)
        except Exception as e:
            print(e)
            break

However, the imfusion.stream.SynchronousConsumer is not meant for real-time, high-throughput applications. It does not queue every sample: if the stream produces data faster than you call get_next_data, intermediate samples may be skipped.

Stream Recording

To record the image stream data, we can pass it into the imfusion.stream.StreamRecorderAlgorithm. This will record the data into a imfusion.SharedImageSet, ensuring that all data is captured. However, the data is only available after the recording is stopped.

Example:

stream_recorder = stream.StreamRecorderAlgorithm(fakeImgStream)
stream_recorder.start()
time.sleep(1)
stream_recorder.stop()
recorded_image = stream_recorder.output()
>>> recorded_image

imfusion.SharedImageSet(size: ..., [
    imfusion.SharedImage(UBYTE width: 100 height: 100),
    imfusion.SharedImage(UBYTE width: 100 height: 100),
    ...
    imfusion.SharedImage(UBYTE width: 100 height: 100)
])]

Note

There is no general Python API to register push callbacks on arbitrary streams (live UI-style subscriptions)..

As a final example, we can create a script like this to record two ultrasound sweeps and use them to optimize the calibration matrix:

import time
import imfusion

imfusion.app = imfusion.ConsoleController()

fake_image_stream = imfusion.algorithm.execute('Stream.CreateStreamIoFakeImage')
process_us_stream = imfusion.algorithm.execute('LiveUS.ProcessUltrasoundStream', fake_image_stream)
fake_tracking_stream = imfusion.algorithm.execute('Stream.CreateStreamIoFakeTracking')
sweep_recorder = imfusion.app.add_algorithm('LiveUS.SweepRecorder', [process_us_stream[0], fake_tracking_stream[0]])

sweep_recorder.start()
time.sleep(1)
sweep_recorder.stop()
sweep1 = sweep_recorder.output()
imfusion.io.write(sweep1, 'sweep1.imf')

sweep_recorder.start()
time.sleep(1)
sweep_recorder.stop()
sweep2 = sweep_recorder.output()
imfusion.io.write(sweep2, 'sweep2.imf')

imfusion.algorithm.execute('US.UltrasoundCalibration', [sweep1[0], sweep2[0]], {'maxFrames': 20})
calibration_matrix = sweep1[0].tracking().calibration