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:

>>> fakeImgStream = imfusion.execute_algorithm('Stream.CreateStreamIoFakeImage')
>>> fakeImgStream
[<imfusion._bindings.ImageStream object at 0x...>]

To record this image stream, we can pass it into the StreamRecorderAlgorithm:

>>> stream_recorder = imfusion.app.add_algorithm('Stream.StreamRecorder', fakeImgStream)
>>> stream_recorder
<imfusion._bindings.StreamRecorderAlgorithm object at 0x...>

And call the imfusion.StreamRecorderAlgorithm.start() and imfusion.StreamRecorderAlgorithm.stop() methods:

>>> import time
>>> stream_recorder.start()
>>> time.sleep(1)
>>> stream_recorder.stop()

The recorded image can then be obtained as output of the stream recorder:

>>> 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

It’s currently not possible to subscribe to real-time updates of streams, but at least it’s possible to record them via a recorder.

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.execute_algorithm('Stream.CreateStreamIoFakeImage')
process_us_stream = imfusion.execute_algorithm('LiveUS.ProcessUltrasoundStream', fake_image_stream)
fake_tracking_stream = imfusion.execute_algorithm('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.execute_algorithm('US.UltrasoundCalibration', [sweep1[0], sweep2[0]], {'maxFrames': 20})
calibration_matrix = sweep1[0].tracking().calibration