File Writer

class gpype.backend.sinks.file_writer.FileWriter(file_name, **kwargs)

Bases: INode

Threaded file writer for real-time data logging to CSV files.

Implements a file writer that operates in a separate background thread to prevent blocking the main signal processing pipeline. Data is queued and written asynchronously to maintain real-time performance. Automatically adds timestamps to filenames and includes sample indices with channel headers in CSV output.

Parameters:

file_name (str)

class Configuration(input_ports=None, **kwargs)

Bases: Configuration

Configuration class for FileWriter parameters.

Parameters:

input_ports (list)

class Keys

Bases: Keys

Configuration keys for file writer settings.

FILE_NAME = 'file_name'

__init__(file_name, **kwargs)

Initialize the file writer with specified filename and ports.

Parameters:

• file_name (str) – Base filename for data output. A timestamp will be automatically appended. Must have .csv extension.

• **kwargs – Additional arguments passed to parent INode class.

Raises:

ValueError – If file format is not supported (.csv required).

start()

Start the file writer and initialize background thread.

Creates the output file with timestamp, validates the file format, and starts the background worker thread for asynchronous writing.

Raises:

• ValueError – If the file format is not .csv.

• IOError – If the file cannot be created or opened.

stop()

Stop the file writer and clean up resources.

Signals the background thread to stop, waits for it to finish processing remaining data, and properly closes the file handle. Ensures all queued data is written before stopping.

setup(data, port_context_in)

Setup method called before processing begins.

Validates that the file handle is properly initialized and ready for writing operations.

Parameters:

• data (dict[str, ndarray]) – Dictionary of input data arrays from connected ports.

• port_context_in (dict[str, dict]) – Context information from input ports.

Return type:

dict[str, dict]

Returns:

Empty dictionary as this is a sink node with no outputs.

Raises:

RuntimeError – If file handle is not initialized (start() not called).

step(data)

Process incoming data by queuing it for background writing.

Parameters:

data (dict[str, ndarray]) – Dictionary containing input data arrays. Uses the default input port to retrieve data for writing.

Return type:

dict[str, ndarray]

Returns:

Empty dictionary as this is a sink node with no outputs.