Pipeline Design API#

This page describes the C and Python APIs that will be needed when using the pipeline design utility.

When designing a pipeline first create an instance of Pipeline, add threads to it with Pipeline.add_thread(). Then add DSP stages such as Biquad using CompositeStage.stage(). The pipeline can be visualised in a Jupyter Notebook using Pipeline.draw() and the xcore source code for the pipeline can be generated using generate_dsp_main().

audio_dsp.design.build_utils #

Utility functions for building and running the application within the Jupyter notebook.

This class packages a set of helper utilities for configuring, building, and running xcore applications using xcommon-cmake within Python.

Parameters:

source_dirstr | pathlib.Path | None: Specify a source directory for this build, passed as the -S parameter to CMake. If None passed or unspecified, defaults to the current working directory.
build_dirstr | pathlib.Path | None: Specify a build directory for this build, passed as the -B parameter to CMake. If None passed or unspecified, defaults to “build” within the current working directory.
bin_dirstr | pathlib.Path | None: Specify a binary output directory for this build. This should match what is configured to be the output directory from “cmake –build” within the application. If None passed or unspecified, defaults to “bin” within the current working directory.
project_namestr | None: The name of the project() specified in the project’s CMakeLists.txt. If None or unspecified, defaults to the name of the current working directory (so if in /app_example_name/, the project name is assumed to be app_example_name).
config_namestr | None: The name of the configuration to use from the project’s CMakeLists.txt. If None or unspecified, defaults to nothing - therefore the –target option to CMake will be just the project name, and the output binary will be assumed to be “<current working directory>/<bin_dir>/<project_name>.xe”. If specified, the –target option to CMake will be “<project name>_<config name>”, and the output binary will be assumed to be “<current working directory>/<bin_dir>/<config_name>/<project name>_<config name>.xe”.

build() → int

Invoke CMake’s build with the options specified in this class instance. Invokation will be of the form cmake --build <build_dir> --target <target_name>, where the target name is constructed as per this class’ docstring.

Returns:

returncode: Return code from the invokation of CMake. 0 if success.

configure() → int | None

Invoke CMake with the options specified in this class instance. Invokation will be of the form cmake -S <source_dir> -B <build_dir>. On first run, the invokation will also contain -G <generator>, where “generator” will be either “Ninja” if Ninja is present on the current system or “Unix Makefiles” if it is not.

Returns:

returncode: Return code from the invokation of CMake. 0 if success.

configure_build_run(xscope: bool = True) → None

Run, in order, this class’ .configure(), .build(), and .run() methods. If any return code from any of the three is nonzero, returns early. Otherwise, sleeps for 5 seconds after the .run() stage and prints “Done!”.

Parameters:

xscopebool: Passed directly to the call to .run(); determines whether to start an xscope server or not.

run(xscope: bool = True, hostname: str = 'localhost', port: str = '12345') → int

Invoke xrun or xgdb with the options specified in this class instance.

If xscope is True the invocation will be of the form:

xgdb -q --return-child-result --batch
-ex "connect --xscope-port <hostname>:<port> --xscope"
-ex "load"
-ex "continue"
<binary>

whereas if xscope if False the invocation will be of the form:

xrun <binary>

where the path to the binary is constructed as per this class’ docstring.

Parameters:

xscopebool: Specify whether to set up an xscope server or not.
hostnamestr: Hostname to use for the xscope server if xscope is True
portstr: Port to use for the xscope server if xscope is True

Returns:

returncode: Return code from the invokation of xrun or xgdb. 0 if success.

audio_dsp.design.composite_stage #

Contains the higher order stage class CompositeStage.

class audio_dsp.design.composite_stage.CompositeStage(graph: Graph, name: str = '')

This is a higher order stage.

Contains stages as well as other composite stages. A thread will be a composite stage. Composite stages allow:

drawing the detail with graphviz
process
frequency response

TODO: - Process method on the composite stage will need to know its inputs and the order of the inputs (which input index corresponds to each input edge). However a CompositeStage doesn’t know all of its inputs when it is created.

Parameters:

graphaudio_dsp.graph.Graph: instance of graph that all stages in this composite will be added to.
namestr: Name of this instance to use when drawing the pipeline, defaults to class name.

add_to_dot(dot)

Recursively adds composite stages to a dot diagram which is being constructed. Does not add the edges.

Parameters:

dotgraphviz.Diagraph: dot instance to add edges to.

composite_stage(name: str = '') → CompositeStage: Create a new composite stage that will be a included in the current composite. The new stage can have stages added to it dynamically.

contains_stage(stage: Stage) → bool

Recursively search self for the stage.

Returns:

bool: True if this composite contains the stage else False

draw(): Draws the stages and edges present in this instance of a composite stage.

get_all_stages() → list[audio_dsp.design.stage.Stage]

Get a flat list of all stages contained within this composite stage and the composite stages within.

Returns:

list of stages.

property o: StageOutputList

Outputs of this composite.

Dynamically computed by searching the graph for edges which originate in this composite and whose destination is outside this composite. Order not currently specified.

process(data): Execute the stages in this composite on the host.

Warning

Not implemented.

stage(stage_type: Type[_StageOrComposite], inputs: StageOutputList, label: str | None = None, **kwargs) → _StageOrComposite

Create a new stage or composite stage and register it with this composite stage.

Parameters:

stage_type: Must be a subclass of Stage or CompositeStage
inputs: Edges of the pipeline that will be connected to the newly created stage.
kwargsdict: Additional args are forwarded to the stages constructors (__init__)

Returns:

stage_type: Newly created stage or composite stage.

stages(stage_types: list[Type[_StageOrComposite]], inputs: StageOutputList) → list[_StageOrComposite]

Iterate through the provided stages and connect them linearly.

Returns a list of the created instances.

audio_dsp.design.graph #

Basic data structures for managing the pipeline graph.

class audio_dsp.design.graph.Edge

Graph node.

Attributes:

iduuid.UUID4: A unique identifier for this node.
sourceNode | None
destNode | None: source and dest are the graph nodes that this edge connects between.

set_dest(node: Node)

Set the dest node of this edge.

Parameters:

node: The instance to set as the dest.

set_source(node: Node)

Set the source node of this edge.

Parameters:

node: The instance to set as the source.

class audio_dsp.design.graph.Graph

A container of nodes and edges.

Attributes:

nodes: A list of the nodes in this graph.
edges: A list of the edges in this graph.

add_edge(edge) → None: Append an edge to this graph.

add_node(node: NodeSubClass) → None

Append a node to this graph.

The node’s index attribute is set here and therefore the node may not coexist in multiple graphs.

get_dependency_dict() → dict[NodeSubClass, set[NodeSubClass]]: Return a mapping of nodes to their dependencies ready for use with the graphlib utilities.

get_view(nodes: list[NodeSubClass]) → Graph[NodeSubClass]: Get a filtered view of the graph, including only the provided nodes and the edges which connect to them.

lock(): Lock the graph. Adding nodes or edges to a locked graph will cause a runtime exception. The graph is locked once the pipeline checksum is computed.

sort() → tuple[NodeSubClass, ...]

Sort the nodes in the graph based on the order they should be executed. This is determined by looking at the edges in the graph and resolving the order.

Returns:

tuple[Node]: Ordered list of nodes

class audio_dsp.design.graph.Node

Graph node.

Attributes:

iduuid.UUID4: A unique identifier for this node.
indexNone | int: node index in the graph. This is set by Graph when it is added to the graph.

audio_dsp.design.parse_config #

Script for use at build time to generate header files.

Use as:

python -m audio_dsp.design.parse_config -c CONFIG_DIR -o OUTPUT_DIR

audio_dsp.design.parse_config.main(args): Use the mako templates to build the autogenerated files.

audio_dsp.design.pipeline #

Top level pipeline design class and code generation functions.

class audio_dsp.design.pipeline.Pipeline(n_in, identifier='auto', frame_size=1, fs=48000, generate_xscope_task=False)

Top level class which is a container for a list of threads that are connected in series.

Parameters:

n_inint: Number of input channels into the pipeline
identifier: string: Unique identifier for this pipeline. This identifier will be included in the generated header file name (as “adsp_generated_<identifier>.h”), the generated source file name (as “adsp_generated_<identifier>.c”), and the pipeline’s generated initialisation and main functions (as “adsp_<identifier>_pipeline_init” and “adsp_<identifier>_pipeline_main”)
frame_sizeint: Size of the input frame of all input channels
fsint: Sample rate of the input channels
generate_xscope_taskbool: Determines whether the generated pipeline automatically instantiates a task to handle tuning over xscope. False by default. If False, the application code will need to explicitly call the “adsp_control_xscope_*” functions defined in adsp_control.h in order to handle tuning over xscope, such as that undertaken by the XScopeTransport() class.

Attributes:

ilist(StageOutput): The inputs to the pipeline should be passed as the inputs to the first stages in the pipeline
threadslist(Thread): List of all the threads in the pipeline
pipeline_stagePipelineStage | None: Stage corresponding to the pipeline. Needed for handling pipeline level control commands

add_pipeline_stage(thread): Add a PipelineStage stage for the pipeline.

static begin(n_in, identifier='auto', frame_size=1, fs=48000, generate_xscope_task=False)

Create a new Pipeline and get the attributes required for design.

Returns:

Pipeline, Thread, StageOutputList: The pipeline instance, the initial thread and the pipeline input edges.

draw(path: pathlib.Path | None = None)

Render a dot diagram of this pipeline.

If path is not none then the image will be saved to the named file instead of drawing to the jupyter notebook.

executor() → PipelineExecutor: Create an executor instance which can be used to simulate the pipeline.

generate_pipeline_hash(threads: list, edges: list)

Generate a hash unique to the pipeline and save it in the ‘checksum’ control field of the pipeline stage.

Parameters:

“threads”: list of [[(stage index, stage type name), …], …] for all threads in the pipeline
“edges”: list of [[[source stage, source index], [dest stage, dest index]], …] for all edges in the pipeline

next_thread() → None

Update the thread which stages will be added to.

This will always create a new thread.

resolve_pipeline()

Generate a dictionary with all of the information about the thread. Actual stage instances not included.

Returns:

dict: ‘identifier’: string identifier for the pipeline “threads”: list of [[(stage index, stage type name, stage memory use), …], …] for all threads “edges”: list of [[[source stage, source index], [dest stage, dest index]], …] for all edges “configs”: list of dicts containing stage config for each stage. “modules”: list of stage yaml configs for all types of stage that are present “labels”: dictionary {label: instance_id} defining mapping between the user defined stage labels and the index of the stage “xscope”: bool indicating whether or not to create an xscope task for control

set_outputs(output_edges: StageOutputList)

Set the pipeline outputs, configures the output channel index.

Parameters:

output_edgesIterable(None | StageOutput): configure the output channels and their indices. Outputs of the pipeline will be in the same indices as the input to this function. To have an empty output index, pass in None.

stage(stage_type: Type[audio_dsp.design.stage.Stage | audio_dsp.design.composite_stage.CompositeStage], inputs: StageOutputList, label: str | None = None, **kwargs) → StageOutputList

Add a new stage to the pipeline.

Parameters:

stage_type: The type of stage to add.
inputs: A StageOutputList containing edges in this pipeline.
label: An optional label that can be used for tuning and will also be converted into a macro in the generated pipeline. Label must be set if tuning or run time control is required for this stage.

property stages: Flattened list of all the stages in the pipeline.

validate()

TODO validate pipeline assumptions.

Thread connections must not lead to a scenario where the pipeline hangs
Stages must fit on thread
feedback must be within a thread (future)
All edges have the same fs and frame_size (until future enhancements)

class audio_dsp.design.pipeline.PipelineStage(**kwargs)

Stage for the pipelne. Does not support processing of data through it. Only used for pipeline level control commands, for example, querying the pipeline checksum.

add_to_dot(dot)

Override the CompositeStage.add_to_dot() function to ensure PipelineStage type stages are not added to the dot diagram.

Parameters:

dotgraphviz.Diagraph
dot instance to add edges to.

audio_dsp.design.pipeline.callonce(f): Decorate functions to ensure they only execute once despite being called multiple times.

audio_dsp.design.pipeline.generate_dsp_main(pipeline: Pipeline, out_dir='build/dsp_pipeline')

Generate the source code for adsp_generated_<x>.c.

Parameters:

pipelinePipeline: The pipeline to generate code for.
out_dirstr: Directory to store generated code in.

audio_dsp.design.pipeline_executor #

Utilities for processing the pipeline on the host machine.

class audio_dsp.design.pipeline_executor.ExecutionResult(result: ndarray, fs: float)

The result of processing samples through the pipeline.

Parameters:

result: The data produced by the pipeline.
fs: sample rate

Attributes:

data: ndarray containing the results of the pipeline.
fs: Sample rate.

play(channel: int)

Create a widget in the jupyter notebook to listen to the audio.

Warning

This will not work outside of a jupyter notebook.

Parameters:

channel: The channel to listen to.

plot(path: Optional[Union[Path, str]] = None)

Display a time domain plot of the result. Save to file if path is not None.

Parameters:

path: If path is not none then the plot will be saved to a file and not shown.

plot_magnitude_spectrum(path: Optional[Union[Path, str]] = None)

Display a spectrum plot of the result. Save to file if path is not None.

Parameters:

path: If path is not none then the plot will be saved to a file and not shown.

plot_spectrogram(path: Optional[Union[Path, str]] = None)

Display a spectrogram plot of the result. Save to file if path is not None.

Parameters:

path: If path is not none then the plot will be saved to a file and not shown.

to_wav(path: str | pathlib.Path): Save output to a wav file.

class audio_dsp.design.pipeline_executor.PipelineExecutor(graph: Graph[Stage], view_getter: Callable[[], PipelineView])

Utility for simulating the pipeline.

Parameters:

graph: The pipeline graph to simulate

log_chirp(length_s: float = 0.5, amplitude: float = 1, start: float = 20, stop: Optional[float] = None) → ExecutionResult

Generate a logarithmic chirp of constant amplitude and play through the simulated pipeline.

Parameters:

length_s: Length of generated chirp in seconds.
amplitude: Amplitude of the generated chirp, between 0 and 1.
start: Start frequency.
stop: Stop frequency. Nyquist if not set

Returns:

ExecutionResult: The output wrapped in a helpful container for viewing, saving, processing, etc.

process(data: ndarray) → ExecutionResult

Process the DSP pipeline on the host.

Parameters:

data: Pipeline input to process through the pipeline. The shape must match the number of channels that the pipeline expects as an input; if this is 1 then it may be a 1 dimensional array. Otherwise, it must have shape (num_samples, num_channels).

Returns:

ExecutionResult: A result object that can be used to visualise or save the output.

class audio_dsp.design.pipeline_executor.PipelineView(stages: Optional[list[audio_dsp.design.stage.Stage]], inputs: list[audio_dsp.design.stage.StageOutput], outputs: list[audio_dsp.design.stage.StageOutput])

A view of the DSP pipeline that is used by PipelineExecutor.

inputs: list[audio_dsp.design.stage.StageOutput]: Alias for field number 1

outputs: list[audio_dsp.design.stage.StageOutput]: Alias for field number 2

stages: Optional[list[audio_dsp.design.stage.Stage]]: Alias for field number 0

audio_dsp.design.plot #

Helper functions for displaying plots in the jupyter notebook pipeline design.

audio_dsp.design.plot.plot_frequency_response(f, h, name='', range=50)

Plot the frequency response.

Parameters:

fnumpy.ndarray: Frequencies (The X axis)
hnumpy.ndarray: Frequency response at the corresponding frequencies in f
namestr: String to include in the plot title, if not set there will be no title.
rangeint | float: Set the Y axis lower limit in dB, upper limit will be the maximum magnitude.

audio_dsp.design.stage #

The edges and nodes for a DSP pipeline.

class audio_dsp.design.stage.PropertyControlField(get, set=None)

For stages which have internal state they can register callbacks for getting and setting control fields.

property value

The current value of this control field.

Determined by executing the getter method.

class audio_dsp.design.stage.Stage(inputs: StageOutputList, config: Optional[Union[Path, str]] = None, name: Optional[str] = None, label: Optional[str] = None)

Base class for stages in the DSP pipeline. Each subclass should have a corresponding C implementation. Enables code generation, tuning and simulation of a stage.

The stages config can be written and read using square brackets as with a dictionary. This is shown in the below example, note that the config field must have been declared in the stages yaml file.

self[“config_field”] = 2 assert self[“config_field”] == 2

Parameters:

configstr | Path: Path to yaml file containing the stage definition for this stage. Config parameters are derived from this config file.
inputsIterable[StageOutput]: Pipeline edges to connect to self
namestr: Name of the stage. Passed instead of config when the stage does not have an associated config yaml file
labelstr: User defined label for the stage. Used for autogenerating a define for accessing the stage’s index in the device code

Attributes:

ilist[StageOutput]: This stages inputs.
fsint | None: Sample rate.
frame_sizeint | None: Samples in frame.
namestr: Stage name determined from config file
yaml_dictdict: config parsed from the config file
labelstr: User specified label for the stage
n_inint: number of inputs
n_outint: number of outputs
detailsdict: Dictionary of descriptive details which can be displayed to describe current tuning of this stage
dsp_blockNone | audio_dsp.dsp.generic.dsp_block: This will point to a dsp block class (e.g. biquad etc), to be set by the child class

add_to_dot(dot)

Add this stage to a diagram that is being constructed. Does not add the edges.

Parameters:

dotgraphviz.Diagraph: dot instance to add edges to.

property constants: Get a copy of the constants for this stage.

create_outputs(n_out)

Create this stages outputs.

Parameters:

n_outint: number of outputs to create.

get_config()

Get a dictionary containing the current value of the control fields which have been set.

Returns:

dict: current control fields

get_frequency_response(nfft=32768) → tuple[numpy.ndarray, numpy.ndarray]

Return the frequency response of this instance’s dsp_block attribute.

Parameters:

nfft: The length of the FFT

Returns:

ndarray, ndarray: Frequency values, Frequency response for this stage.

get_required_allocator_size()

Calculate the required statically-allocated memory in bytes for this stage. Formats this into a compile-time determinable expression.

Returns:

compile-time determinable expression of required allocator size.

property o: StageOutputList: This stage’s outputs. Use this object to connect this stage to the next stage in the pipeline. Subclass must call self.create_outputs() for this to exist.

plot_frequency_response(nfft=32768)

Plot magnitude and phase response of this stage using matplotlib. Will be displayed inline in a jupyter notebook.

Parameters:

nfftint: Number of frequency bins to calculate in the fft.

process(in_channels)

Run dsp object on the input channels and return the output.

Args:: in_channels: list of numpy arrays

Returns:

list of numpy arrays.

set_constant(field, value, value_type)

Define constant values in the stage. These will be hard coded in the autogenerated code and cannot be changed at runtime.

Parameters:

fieldstr: name of the field
valuendarray or int or float or list: value of the constant. This can be an array or scalar

set_control_field_cb(field, getter, setter=None)

Parameters:

fieldstr: name of the field
getterfunction: A function which returns the current value
setterfunction: A function which accepts 1 argument that will be used as the new value

class audio_dsp.design.stage.StageOutput(fs=48000, frame_size=1)

The Edge of a dsp pipeline.

Parameters:

fsint: Edge sample rate Hz
frame_sizeint: Number of samples per frame

Attributes:

sourceaudio_dsp.design.graph.Node: Inherited from Edge
destaudio_dsp.design.graph.Node: Inherited from Edge
source_indexint | None: The index of the edge connection to source.
fsint: see fs parameter
frame_sizeint: see frame_size parameter

property dest_index: int | None: The index of the edge connection to the dest.

class audio_dsp.design.stage.StageOutputList(edges: list[audio_dsp.design.stage.StageOutput | None] | None = None)

A container of StageOutput.

A stage output list will be created whenever a stage is added to the pipeline. It is unlikely that a StageOutputList will have to be explicitly created during pipeline design. However the indexing and combining methods shown in the example will be used to create new StageOutputList instances.

Parameters:

edges: list of StageOutput to create this list from.

Examples

This example shows how to combine StageOutputList in various ways:

# a and b are StageOutputList
a = some_stage.o
b = other_stage.o

# concatenate them
a + b

# Choose a single channel from 'a'
a[0]

# Choose channels 0 and 3 from 'a'
a[0, 3]

# Choose a slice of channels from 'a', start:stop:step
a[0:10:2]

# Combine channels 0 and 3 from 'a', and 2 from 'b'
a[0, 3] + b[2]

# Join 'a' and 'b', with a placeholder "None" in between
a + None + b

Attributes:

edgeslist[StageOutput]: To access the actual edges contained within this list then read from the edges attribute. All methods in this class return new StageOutputList instances (even when the length is 1).

class audio_dsp.design.stage.ValueControlField(value=None): Simple field which can be updated directly.

audio_dsp.design.stage.all_stages() → dict[str, Type[audio_dsp.design.stage.Stage]]: Get a dict containing all stages in scope.

audio_dsp.design.stage.find_config(name)

Find the config yaml file for a stage by looking for it in the default directory for built in stages.

Parameters:

namestr: Name of stage, e.g. a stage whose config is saved in “biquad.yaml” should pass in “biquad”.

Returns:

Path: Path to the config file.

audio_dsp.design.thread #

Contains classes for adding a thread to the DSP pipeline.

class audio_dsp.design.thread.DSPThreadStage(**kwargs)

Stage for the DSP thread. Does not support processing of data through it. Only used for DSP thread level control commands, for example, querying the max cycles consumed by the thread.

add_to_dot(dot)

Exclude this stage from the dot diagram.

Parameters:

dotgraphviz.Diagraph: dot instance to add edges to.

class audio_dsp.design.thread.Thread(id: int, **kwargs)

A composite stage used to represent a thread in the pipeline. Create using Pipeline.thread rather than instantiating directly.

Parameters:

idint: Thread index
kwargsdict: forwarded to __init__ of CompositeStage

Attributes:

idint: Thread index
thread_stageStage: DSPThreadStage stage

add_thread_stage(): Add to this thread the stage which manages thread level commands.