Data loggers¶

Using the DataLogger class, variables can be logged programmatically to the host PC’s disk at up to the CPU task frequency. It supports the following file formats:

CSV text file (.csv)
MATLAB MAT file (.mat)
HDF5 file (.h5 or .hdf5)
MDF4 file (.mf4)

For each format, the internal data structure and parsing instructions are detailed in the Reading the recorded data section below.

Note

Data loggers can only be configured via Python scripts; they are not available in the Cockpit GUI.

Recording data¶

The following example demonstrates how to record data during an automated experiment.

# Create and configure a data logger
data_logger = project.add_data_logger(
   variables=["Ia", "Ib", "Ic"],
   path=r"C:\imperix\captures\recording.hdf5",
   sampling_frequency_hz=1_000.0,
   overwrite=True,  # Replace the file if it already exists.
)

# Start recording
data_logger.start()

# Execute automated experiment steps here...

# Stop recording
data_logger.stop()

# Remove the data logger from the project
data_logger.remove()

Long-term data logging¶

Unlike most other script snippets, this example requires an active project linked to a controller running a user code.

The script looks up a specific project (e.g., “Central PV inverter”) and starts logging the targeted variables. To prevent file size issues during extended captures, the logger is configured to create a new file every 24 hours without dropping any data points. The script will continue recording until the user presses Enter or closes the terminal.

Note

Currently, only one Python script can connect to a Cockpit instance at a time. Because this long-term logging script runs continuously, it will block other scripts from interacting with Cockpit on the same machine until the recording is stopped.

with Cockpit() as cockpit:

   # Look up the project by name
   project = cockpit.projects.get("Central PV inverter")

   if project is None:
      raise SystemExit(f"Project not found.")

   # Initialize the data logger
   data_logger = project.add_data_logger(
      variables=["Vdc", "Ia", "Ib", "Ic"],
      path=r"C:\imperix\captures\recording.h5",
      sampling_frequency_hz=1_000.0,
      max_file_duration_s=24 * 60 * 60, # Automatically start a new file every 24 hours
      overwrite=True,
   )

   # Start recording
   data_logger.start()

   # Keep the script alive and recording until the Enter key is pressed
   # Note: Cockpit's built-in script launcher does not support user input,
   # so this script must be run from an external terminal or IDE.
   input("Recording... Press Enter to stop.\n")

   # Stop recording and finish writing the current file
   data_logger.stop()
   data_logger.remove()

Reading the recorded data¶

This section describes the internal file structure for each supported recording format: CSV, HDF5, MAT, and MDF4. Python code examples are provided for each format to demonstrate how to extract the logged variables and timestamps into a pandas DataFrame for analysis.

Note

With the exception of the HDF5 format, files can be processed only once the recording is complete (after stop() is called). HDF5 uses Single Writer Multiple Reader (SWMR) mode, allowing data to be read while the recording is still active.

CSV¶

In CSV files, each value is stored as plain text, meaning it can be easily inspected in a text editor or opened in a spreadsheet application.

However, storing data as text uses significantly more disk space and requires more CPU overhead to write than binary formats. This makes CSV less suitable for high-frequency or long-term recordings.

A CSV recording starts with five header lines describing the capture, followed by a data table with one column per variable and one row per sample. For a recording of Ia, Ib, and Ic, the layout is:

General header
Date,<recording date>
Time resolution,<sample interval in seconds>
Cockpit version,<version>
Data
Timestamp,Ia,Ib,Ic
<timestamp>,<Ia value>,<Ib value>,<Ic value>
<timestamp>,<Ia value>,<Ib value>,<Ic value>
<timestamp>,<Ia value>,<Ib value>,<Ic value>
...

The following example shows how to read and filter a CSV file using the pandas package.

import pandas as pd

# Read the CSV file, skipping the first 5 header lines
csv_data = pd.read_csv(r"C:\imperix\captures\recording.csv", skiprows=5)

# Filter the DataFrame to extract only specific columns
csv_data = csv_data[["Timestamp", "Ia"]]

# Display the first five rows of the filtered data
print(csv_data.head())

HDF5¶

An HDF5 recording contains a General_header group with information about the recording, and a separate array for the timestamps and each variable. Values at the same position in these arrays belong to the same sample.

HDF5 files are written in Single Writer Multiple Reader (SWMR) mode, allowing external applications to safely read the dataset while the recording is in progress. This usage is described in the h5py SWMR documentation.

The layout below shows a recording with N samples:

recording.h5
├── General_header
│   ├── Date
│   ├── Time_resolution
│   └── Cockpit_version
├── Timestamp    (N values)
├── Ia           (N values)
├── Ib           (N values)
└── Ic           (N values)

The following example reads the Timestamp and Ia arrays into a table. The same code works with both .h5 and .hdf5 files:

import h5py
import pandas as pd

with h5py.File(r"C:\imperix\captures\recording.h5", "r") as recording:
   # [:] reads all the values stored in each 1D array
   hdf_data = pd.DataFrame({
      "Timestamp": recording["Timestamp"][:],
      "Ia": recording["Ia"][:],
   })

print(hdf_data.head())  # Display the first five rows

MAT¶

MAT recordings use the MATLAB 7.3 format. Because MATLAB 7.3 is based on HDF5, these files can also be read using the h5py package. They contain the same named arrays and recording information as standard HDF5 recordings.

The main difference is dimensional: MATLAB represents vectors with two dimensions, unlike the one-dimensional arrays used in the HDF5 format above. Each signal is saved as a column vector with shape (N, 1) in MATLAB. However, h5py reads the dimensions in reverse order, presenting it as a row with shape (1, N):

recording.mat (as read by h5py)
├── General_header
│   ├── Date
│   ├── Time_resolution
│   └── Cockpit_version
├── Timestamp    (1 row, N values)
├── Ia           (1 row, N values)
├── Ib           (1 row, N values)
└── Ic           (1 row, N values)

The following example reads the row of values for Timestamp and Ia into a table:

import h5py
import pandas as pd

with h5py.File(r"C:\imperix\captures\recording.mat", "r") as recording:
   # [0, :] extracts all values from the first (and only) row
   mat_data = pd.DataFrame({
      "Timestamp": recording["Timestamp"][0, :],
      "Ia": recording["Ia"][0, :],
   })

print(mat_data.head())  # Display the first five rows

MDF4¶

An MDF4 recording stores timestamps and user variables as named channels inside a Data Logger channel group. Each channel contains one value per sample.

The recording start time is stored in the file header, while the date, time resolution, and Cockpit version are stored as properties in the header comment. The diagram below maps out this structure:

recording.mf4
├── File header
│   ├── Recording start time
│   └── Header comment
│       ├── General_header.Date
│       ├── General_header.Time_resolution
│       └── General_header.Cockpit_version
└── Channel Group: "Data Logger"
    ├── Timestamp    (N values)
    ├── Ia           (N values)
    ├── Ib           (N values)
    └── Ic           (N values)

The following example loads the Timestamp and Ia channels into a pandas DataFrame using the asammdf package:

from asammdf import MDF

with MDF(r"C:\imperix\captures\recording.mf4") as recording:
   # Extract specific channels directly to a pandas DataFrame
   mdf_data = recording.to_dataframe(channels=["Timestamp", "Ia"])

print(mdf_data.head())  # Display the first five rows

API Reference¶

class imperix.DataLogger¶

Represents a data logger in Cockpit.

A data logger continuously saves acquired data from selected variables to a file on the host PC.

property path: str¶: Absolute output file path configured for this data logger.

property variables: list[str]¶: Variable names recorded by this data logger.

property sampling_frequency_hz: float¶: Sampling frequency in hertz.

property max_file_duration_s: float | None¶: Maximum duration of each file, or None when disabled.

get_status(*, timeout: float | None | UseDefaultTimeout = UseDefaultTimeout.TOKEN) → DataLoggerStatus¶

Returns the data logger’s current state and any failure message.

Parameters:: timeout – Maximum time to wait for a response in seconds.
Returns:: Current DataLoggerStatus.

start(*, timeout: float | None | UseDefaultTimeout = UseDefaultTimeout.TOKEN) → None¶

Starts the recording.

Parameters:: timeout – Maximum time to wait for a response in seconds.
Raises:: InvalidStateError – If the data logger is already running, its project or target is no longer available, or the output file cannot be opened.

stop(*, timeout: float | None | UseDefaultTimeout = UseDefaultTimeout.TOKEN) → None¶

Stops the recording.

Stopping the data logger finalizes the output file.

Parameters:: timeout – Maximum time to wait for a response in seconds.
Raises:: InvalidStateError – If Cockpit cannot finish writing the output file.

split(*, timeout: float | None | UseDefaultTimeout = UseDefaultTimeout.TOKEN) → None¶

Finalizes the current file and continues recording in a new one.

Parameters:: timeout – Maximum time to wait for a response in seconds.
Raises:: InvalidStateError – If the data logger is not running or Cockpit cannot open the next segment.

remove(*, timeout: float | None | UseDefaultTimeout = UseDefaultTimeout.TOKEN) → None¶

Stops and removes this data logger from the project.

The output file is not deleted.

Parameters:: timeout – Maximum time to wait for a response in seconds.

class imperix.DataLoggerStatus(state: Literal['stopped', 'running', 'failed'], error: str | None)¶

Current data logger state reported by Cockpit.

state¶

One of "stopped", "running", or "failed".

Type:: Literal[‘stopped’, ‘running’, ‘failed’]

error¶

Failure message, or None when no error is reported.

Type:: str | None

class imperix.DataLoggerCollection(cockpit: Cockpit, *, project_id: int)¶

Data loggers attached to a Cockpit project.

add(variables: str | Sequence[str], path: str | PathLike[str], sampling_frequency_hz: float, *, overwrite: bool = False, max_file_duration_s: float | None = None, timeout: float | None | UseDefaultTimeout = UseDefaultTimeout.TOKEN) → DataLogger¶

Configures a data logger for this project.

A data logger continuously saves acquired data from selected variables to a file on the host PC in CSV, MAT, HDF5, or MDF4 format.

Warning

CSV stores each value as readable text, which uses more disk space and takes more CPU to write. MAT, HDF5, or MDF4 file formats are therefore recommended when logging at high frequency or for long periods.

See the Data loggers page for more details.

Parameters:

variables – Variable name or variable names to record.
path – Output file path. The extension selects the file format (.csv, .mat, .h5, .hdf5, or .mf4).
sampling_frequency_hz – Sampling frequency in hertz. Must be greater than 0. This value saturates at the CPU task frequency.
overwrite – Replace an existing output file when True. Defaults to False.
max_file_duration_s – Maximum recorded-data duration of each physical file in seconds. Must be at least 1 second when set. None disables automatic splitting. Defaults to None.
timeout – Maximum time to wait for a response in seconds.

Returns:

The newly created DataLogger.

Raises:

NotFoundError – If one of the requested variables is not available.
InvalidStateError – If the project is not linked to a ready target or another data logger already uses the output path.

Example:

# Configure the data logger.
data_logger = project.data_loggers.add(
    ["speed_ref", "phase_current"],
    "C:/imperix/recordings/my_recording.csv",
    sampling_frequency_hz=1_000.0,
    overwrite=True,
    max_file_duration_s=60.0,
)

data_logger.start()
# Run the experiment.
data_logger.split()  # Optional manual file boundary.
data_logger.stop()
data_logger.remove()

list(*, timeout: float | None | UseDefaultTimeout = UseDefaultTimeout.TOKEN) → list[DataLogger]¶

Returns a list of all data loggers configured for this project.

Parameters:: timeout – Maximum time to wait for a response in seconds.
Returns:: DataLogger objects for this project.