Python Example

iris-sdk-sample.py

import iris.sdk
import numpy as np
import soundfile as sf
from pathlib import Path

from utils import check_licence_info, check_context_config, check_context_request, create_output_filepath #, file_to_string


def process_callback(event):
    match event.type:
        case iris.sdk.ProcessEventType.start:
            print(f"\rProcessing: started")
        case iris.sdk.ProcessEventType.process:
            print(f"Processing: {event.percentage:.2f}%", end='\r')
        case iris.sdk.ProcessEventType.end:
            print(f"\nProcessing: ended")


if __name__ == "__main__":
    # initialise iris sdk. either pass in the licence and key as strings or let the sdk discover
    # them from its default data location. See readme.md for default locations
    # licence = file_to_string("/full/path/to/iris.lic")
    # key = file_to_string("/full/path/to/iris.key")
    # info = iris.sdk.init(licence, key)
    info = iris.sdk.init()
    print(f"Initialised iris-sdk {info.version} for {info.name}")
    check_licence_info(info)

    # Load a wav file and fetch the sample rate and data format
    script_dir = Path(__file__).resolve().parent
    input_filepath = script_dir.parent / "sample_audio" / "breaking_point_f32_48k.wav"
    audio_in, sample_rate = sf.read(input_filepath, dtype='float32')
    file_info = sf.info(input_filepath)
    original_format = file_info.subtype

    # prepare an output buffer to hold the processed audio
    audio_out = np.zeros_like(audio_in, dtype=np.float32)

    # Create an audio processing context request
    context_request = iris.sdk.AudioContextRequest()
    context_request.processors = ["passthrough"]  # set the processor chain to use
    context_request.sample_rate = sample_rate  # set the sample rate of the input audio
    context_request.channel_count = file_info.channels  # set the channel count of the input audio
    context_request.resampler_mode = iris.sdk.ResamplerMode.HQ  # set the resampler quality (HQ or SPEED)
    check_context_request(context_request)

    # Create an audio processing context using the context request
    config = iris.sdk.create_audio_context(context_request)
    check_context_config(config)

    # Process the entire buffer in one go, accounting for any transport delay
    # If processing a stream, call iris.sdk.process when each new buffer of stream data arrives instead
    # In the streaming case, frames should be set to the size of the stream buffer size.
    # If each buffer is the same size, set fixed_frame_count to true in the context request
    # If the stream buffer size is known up front, set the buffer size in the context request
    # streaming pseudocode
    # while new_stream_data_available:
    #   process_in_buffer = get_stream_source()
    #   process_out_buffer = np.zeros_like(process_in_buffer, dtype=np.float32)
    #   iris.sdk.process(config.context_id, process_in_buffer, process_out_buffer, len(process_in_buffer))
    #   put_stream_sink(process_out_buffer)

    print(f"Processing {input_filepath}")
    iris.sdk.process_offline(config.context_id, audio_in, audio_out, file_info.frames, process_callback)

    # save the processed output alongside the input file in the original file format
    output_filepath = create_output_filepath(input_filepath, config.description)
    sf.write(output_filepath, audio_out, sample_rate, subtype=original_format)
    print(f"Saving {output_filepath}")

    # Release the context
    print(f"Cleaning up...")
    iris.sdk.release_audio_context(config.context_id)
    iris.sdk.cleanup()
    print(f"Done.")

utils.py

import os
import sys
from pathlib import Path

import iris.sdk


def file_to_string(file_path: str | Path) -> str:
    path = Path(file_path)
    with open(path, "r") as f:
        data = f.read()
    return data


def check_licence_info(info):
    print(f"Licence Info:")
    print(f"  {'Name:':<20}{info.name}")
    print(f"  {'SDK Version:':<20}{info.version}")
    print(f"  {'Expiry:':<20}{info.expiry}")
    print(f"  {'Licence status:':<20}{info.status}")
    print(f"")
    print('Available processors:', *info.available_processors, sep='\n- ')
    print(f"")
    if not info.status == iris.sdk.LicenceStatus.LICENCE_VALID:
        print('Invalid licence - exiting')
        sys.exit(0)


def check_context_request(request):
    print(f"Audio Context Request:")
    print(f"  {'Sample rate:':<20}{request.sample_rate}")
    print(f"  {'Buffer length:':<20}{request.buffer_length}")
    print(f"  {'Processors:':<20}{request.processors}")
    print(f"  {'Resampler Mode:':<20}{request.resampler_mode}")
    print(f"  {'Optimization:':<20}{request.optimization}")
    print(f"  {'Fixed frame count:':<20}{request.fixed_frame_count}")
    print(f"")


def check_context_config(config):
    print(f"Audio Context Configuration:")
    print(f"  {'Context id:':<20}{config.context_id}")
    print(f"  {'Description:':<20}{config.description}")
    print(f"  {'Sample rate:':<20}{config.sample_rate}")
    print(f"  {'Buffer length:':<20}{config.buffer_length}")
    print(f"  {'Channels:':<20}{config.channel_count}")
    if config.will_resample:
        print(f"  {'Resampling:':<20}{config.will_resample}")
        print(f"  {'Resampler Mode:':<20}{config.resampler_mode}")
    print(f"  {'Optimisation:':<20}{config.optimization}")

    if config.error == 0:
        print(f"  Context Parameters:")
        for group in config.parameters:
            print(f"    {group.name}")
            for param in group.parameter_descriptions:
                print(f"      -> {param.name}")

    else:
        print(f"  {'Error:':<20}{config.error}")
        print(f"")
        print('Context creation failed')
        sys.exit(0)

    print(f"")


def create_output_filepath(input_filepath, description):
    directory, filename = os.path.split(input_filepath)
    base_name, ext = os.path.splitext(filename)
    return os.path.join(directory, f"{base_name}_{description.replace(' -> ', '_')}{ext}")