Overview

The imfusion.live_ultrasound package exposes Python bindings for streaming (live) ultrasound in ImFusion: wrapping ultrasound ImageStream instances for real-time processing and geometry, and recording streams into an UltrasoundSweep via SweepRecorderAlgorithm.

Importing imfusion.live_ultrasound also loads imfusion.stream and imfusion.ultrasound (see the module import logic in the bindings). Install the Live US stack with pip install 'imfusion-sdk[live_us,stream,ultrasound]' (see Installation).

For detailed documentation of specific classes and functions, use Python’s built-in help() or the docstrings on each symbol.

Examples

Short workflows below complement the full API listing in Reference.

Stream properties and state

An ImageStream (such as FakeImageStream from Stream.CreateStreamIoFakeImage) subclasses Stream and Data. Besides the lifecycle methods (open, start, pause, resume, stop, close), useful attributes include name, kind, modality, top_down, supports_pausing, and current_state. State values live on the nested enum Stream.State (also visible on concrete stream types).

dir(image_stream) lists everything exposed to Python (including enum members such as RUNNING on the class); the snippet below shows typical inspection and state transitions on the fake stream:

>>> import imfusion
>>> from imfusion import Data
>>> image_stream = imfusion.algorithm.execute('Stream.CreateStreamIoFakeImage')[0]
>>> image_stream
<imfusion.stream.FakeImageStream object at 0x...>
>>> image_stream.name
'Fake Image Stream'
>>> image_stream.kind == Data.Kind.IMAGE_STREAM
True
>>> image_stream.modality  # often NA until images define a modality
<Modality.NA: 0>
>>> image_stream.supports_pausing
True
>>> image_stream.open()
True
>>> image_stream.start()
True
>>> image_stream.current_state == image_stream.State.RUNNING
True
>>> image_stream.pause()
True
>>> image_stream.current_state == image_stream.State.PAUSED
True
>>> image_stream.resume()
True
>>> image_stream.stop()
True
>>> image_stream.close()
True

For a quick membership check without listing every symbol, image_stream.is_state_one_of([image_stream.State.RUNNING, image_stream.State.PAUSED]) is also available.

Processed ultrasound stream

ProcessedUltrasoundStream wraps a live ultrasound ImageStream and applies ProcessUltrasoundParameters (crop, mask, depth, and related options from the offline US module). Typical follow-ups are init_geometry(), detect_geometry(), or preset management.

import imfusion
import imfusion.live_ultrasound as live_us
import imfusion.ultrasound as us

image_stream = imfusion.algorithm.execute('Stream.CreateStreamIoFakeImage')[0]

params = us.ProcessUltrasoundParameters(
        apply_crop=True, # use FrameGeometry's bounding box for cropping
        depth=10.0, # set depth
        apply_depth=True #use depth for spacing
        )

processed_us_stream = live_us.ProcessedUltrasoundStream(
    image_stream=image_stream,
    processing_parameters=params,
    automatic_preset_change=False # don't switch preset when probe/preset changes are detected
    )

Sweep recording

To create UltrasoundSweeps from live ultrasound data, SweepRecorderAlgorithm records one ultrasound stream together with zero or more Stream instances (for example tracking), calibrates geometry, and produces an UltrasoundSweep. Pass the stream list and optional matrices / device metadata as supported by the constructor.

import numpy as np
import imfusion.live_ultrasound as live_us

# Streams are typically an ultrasound ImageStream plus optional TrackingStream instances.
image_stream = imfusion.algorithm.execute('Stream.CreateStreamIoFakeImage')[0]
tracking_stream = imfusion.algorithm.execute('Stream.CreateStreamIoFakeTracking')[0]

recorder = live_us.SweepRecorderAlgorithm(
    [image_stream, tracking_stream],
    reference_coordinate_system=np.eye(4),
    registration=np.eye(4),
    temporal_offset=0.0,
    live_stream_recording=False,
    device_name="",
)

    recorder.start()
    time.sleep(10.0)   # record for a bit
    recorder.stop()

    out = recorder.compute()
    sweeps = [d for d in out if isinstance(d, us.UltrasoundSweep)]

    print(f"Got {len(sweeps)} sweep(s)")

For execution and status handling, these types follow the usual ImFusion algorithm / stream patterns (for example compute() on algorithms where applicable); see the reference entries for each class.

Note: This module requires the ImFusion LiveUS plugin to be properly installed and licensed, and depends on the Stream and Ultrasound Python bindings.