pipeline

This pipeline creation tool is designed to help you create a pipeline to process biologcal signal data. It aims to provide an end-to-end solution for processing biological signal data. Ideally you should be able to use this tool to create a pipeline that takes your signal as an input and produce processed signal and a feature vector as output.

bio_channel

class Channel(signal, name, sampling_rate)

Bases: object

Biological signal channel class

Parameters:

• signal (ArrayLike) –

• name (str) –

• sampling_rate (float) –

copy()

Returns a copy of the channel :returns: copy – Copy of the channel :rtype: Channel

property duration

get_timestamp()

get_window(window_index)

Returns a window of the channel :param window_index: Index of the window to return :type window_index: int

Returns:

window – Window of the channel

Return type:

ArrayLike

get_window_timestamps()

property n_windows

property segmented

bio_data

class Bio_Data

Bases: object

Signal object with add and iterate process objects

add_channel(signal, channel_name=None, sampling_rate=None, modify_existed=False, is_event=False)

Add a channel to the signal :param signal: Signal to add :type signal: Union[ArrayLike, Channel ,Event_Channel] :param channel_name: Name of the channel :type channel_name: str :param sampling_rate: Sampling rate of the signal :type sampling_rate: Union[int, float] :param modify_existed: If True, if a channel with the same name already exists, it will be overwritten

If False, if a channel with the same name already exists, a new name will be generated

Parameters:

• is_event (bool) – If True, the signal will be converted to an Event_Channel object If False, the signal will be converted to a Channel object

• signal (_SupportsArray[dtype] | _NestedSequence[_SupportsArray[dtype]] | bool | int | float | complex | str | bytes | _NestedSequence[bool | int | float | complex | str | bytes] | Channel | Event_Channel) –

• channel_name (str | None) –

• sampling_rate (int | float | None) –

• modify_existed (bool) –

property channel_count

copy()

get_channel_names()

join(other, overwrite=False)

Join two Bio_Data objects :param other: Bio_Data object to join :type other: Bio_Data :param overwrite: If True, if a channel with the same name already exists, it will be overwritten

If False, if a channel with the same name already exists, a new name will be generated

Parameters:

• other (Bio_Data) –

• overwrite (bool) –

property multichannel

remove_channel(channel_name)

Remove a channel from the signal :param channel_name: Name of the channel to remove :type channel_name: str

rename_channel(old_name, new_name)

Rename a channel :param old_name: Old name of the channel :type old_name: str :param new_name: New name of the channel :type new_name: str

bio_process

class Bio_Process(process_method, process_name, *args, **kwargs)

Bases: object

process_args(**kwargs)

Process the input arguments

run(*args, **kwargs)

Run the process method on the input arguments

bioprocess_queue

class Process_List(name='Process_Queue')

Bases: object

add_process(process, input_signals=None, output_signals=None, is_event=False, *args, **kwargs)

get_process_by_name(name)

run_next(bio_data)

run_process_queue(bio_data)

Parameters:

bio_data (Bio_Data) –

Return type:

Bio_Data

run_single(args, kwargs)

channel_input

convert_channel(signal, sampling_rate=None, name=None, n_windows=1, n_signal=None, index=None)

event_channel

class Event_Channel(events, name, sampling_rate)

Bases: object

Biological signal channel class

Parameters:

• events (list) –

• name (str) –

• sampling_rate (float) –

copy()

property event_names

get_event(event_name)

Parameters:

event_name (str) –

get_timestamp()

get_window(window_index)

get_window_timestamps()

property n_events

property n_windows

property segmented

event_input

convert_event(event, name=None, sampling_rate=None, indicator=None, n_windows=1, n_signal=None)

convert_list_dict_to_dict_list(list_dict)

event_from_signal(signal, indicator, n_windows=1)

feature_extraction

class Feature(name, function, *args, **kwargs)

Bases: object

process(*args, **kwargs)

Return type:

DataFrame

feature_queue

pipeline_export

pipeline

class Bio_Pipeline(windowed_process=False, window_size=None, step_size=None)

Bases: object

add_feature_step(feature, input_signals, *args, **kwargs)

Parameters:

feature (Feature) –

clear_data()

clear_features()

clear_input()

convert_windows()

export_data(filename)

export_features(filename)

extract_features()

get_data()

get_features()

get_input()

run_pipeline()

set_input(data, sampling_rate=None, name=None, is_event=False, **kwargs)

Parameters:

data (Bio_Data | ArrayLike | Channel) –

set_window_parameters(window_size=10, step_size=5)