Recording

exception pyflind.recording.base.FliRecorderError: Bases: Exception

class pyflind.recording.base.FliRecorder(filename: str, fs: float = 0, metadata: Mapping[str, str] = {})

Bases: ABC

Iteratively record a set of sampled signals to a file.

Abstract base class defining a common interface for recording to any type of file implemented by subclasses.

Parameters:

filename – name of file
fs – samplerate in Hz
metadata – string metadata to store with the recording

abstractmethod close() → None

Close the recording.

No more data can be written after calling close.

abstractmethod write_data(data: Mapping[str | int, Sequence[int | float]]) → int

Write a block of samples to the file.

The data is provided in a dictionary of sample arrays, each having the same length (number of samples). The keys represent signal names which correspond to fields/channels in the file. All calls must have the same fields and append the samples to the recording file.

Parameters:: data – block of samples to write
Returns:: number of samples written
Return type:: int

exception pyflind.recording.base.FliRecordingError: Bases: Exception

class pyflind.recording.base.FliRecording(filename: str, time_zero: bool = False, fs: float = 0.0, limit: int = 0, convert_field: bool = True)

Bases: ABC

Load a set of sampled signals from a file.

Abstract base class defining a common interface for loading a recording of any type of file implemented by subclasses.

A ‘time’ stream is automatically computed in a manner specific to the file type.

Data is retrievable by stream/channel/column name by indexing an object of this type. The object also supports iteration and will iterate through all streams except time. This makes for convenient signal processing or plotting, for example:

recording = FliRecordingSubClass('some_file.type')
t = recording['time']
for stream in recording:
    do_something(t, recording[stream])

Parameters:

filename – name of CSV file
time_zero – subtract the first timestamp from all times if True
fs – samplerate in Hz if known, otherwise computed from file
limit – max number of samples to read from file, 0 for all
convert_field – convert magnetic field channels to units of T if True

samplerate() → float

Get the file samplerate.

Samplerate is either provided at instantiation or computed as the mean period between samples.

Returns:: samplerate in Hz
Return type:: float

metadata() → Mapping[str, str]

Get any metadata from the recording.

Recording metadata is optional and may not be supported by all file types. If metadata isn’t supported or wasn’t recorded then an empty dict is returned.

Returns:: metadata dictionary
Return type:: Mapping[str,str]

stream_matches_critera(stream_list: Sequence[str], stream: str) → bool

Check if a stream name matches a list of criteria.

Parameters:

stream_list – list of stream names or substrings to match in stream names
stream – stream name to check for matches in stream_list criteria

Returns:

True if stream matches criteria in stream_list, False otherwise

Return type:

bool

match_streams(criteria: Sequence[str]) → Sequence[str]

Return a list of streams matching specified criteria.

Criteria list may include exact stream names and stream name substrings. Exact matches take priority and when found are removed from substring searches on remaining streams. An empty list of criteria will return all stream names.

Parameters:: criteria – list of stream names or substrings to find
Returns:: list of stream names found to match criteria
Return type:: Sequence[str]

class pyflind.recording.csv.CsvRecorder(filename: str, fs: float = 0, metadata: Mapping[str, str] = {})

Bases: FliRecorder

Iteratively record a set of sampled signals in a CSV formatted file.

The output file is opened on instantiation and the fields (columns) to save in the file are automatically determined on the first call to write_data. When recording is complete call close.

Metadata is not currently supported with CSV files.

Parameters:: filename – name of CSV file

close() → None

Close the recording.

No more data can be written after calling close.

write_data(data: Mapping[str | int, Sequence[int | float]]) → int

Write a block of samples to the file.

The data is provided in a dictionary of sample arrays, each having the same length (number of samples). The keys represent signal names which correspond to fields/columns in the CSV file. The CSV header is written on the first call to write_data which determines the set of fields and order of fields for the file. Each subsequent call to write_data must have the same fields and appends the samples to the CSV file.

Parameters:: data – block of samples to write
Returns:: number of samples written
Return type:: int

class pyflind.recording.csv.CsvRecording(filename: str, time_zero: bool = False, fs: float = 0.0, limit: int = 0, convert_field: bool = True)

Bases: FliRecording

Load a set of sampled signals from a CSV formatted file.

The CSV file is expected to represent a set of sampled signals or time series with uniform sampling rate. Several ways of of storing timestamps are supported to accommodate various formats used by FieldLine software as well as files from other sources. In all supported cases the time field(s) are converted to a floating point value in seconds and available through the ‘time’ key. If the file contains a ‘time’ column then it is interpreted as a float and used unaltered. Columns which are recognized as representing magnetic field values are automatically converted to units of Tesla as the file is read.

By default the samplerate of the file is computed as the mean time between samples using available timestamps. For short recordings with rx_t columns this may be inaccurate due to sample buffering during the recording. This can be solved by providing the samplerate if is known or ignoring the time stream and computing your own separately.

Data is retrievable by stream/channel/column name by indexing an object of this type. The object also supports iteration and will iterate through all streams except time. This makes for convenient signal processing or plotting, for example:

csv_recording = CsvRecording('some_file.csv')
t = csv_recording['time']
for stream in csv_recording:
    do_something(t, csv_recording[stream])

Parameters:

filename – name of CSV file
time_zero – subtract the first timestamp from all times if True
fs – samplerate in Hz, 0 to estimate from timestamps
limit – max number of samples to read from file, 0 for all
convert_field – convert magnetic field channels to units of T if True