Recorder#

Overview#

pyXCP provides first-class support for XCP DAQ (Data Acquisition) including both online processing and offline recording. This page consolidates guidance and examples from discussions (disc_165) and examples in the repository to help you set up DAQ lists, select an appropriate Policy, record to the native .xmraw format, export CSV in real-time, and post-process recordings into formats like ASAM MDF, Parquet (Arrow), or SQLite.

Key concepts:

  • DAQ List: A list of measurements attached to an event on the ECU.

  • Policy: A strategy object that drives how DAQ data is handled (online vs offline). You pass the policy into the application context via ap.run(policy=...).

  • .xmraw: Proprietary, high-throughput recording format used by pyXCP. Convert later to other formats.

  • Timestamps: Two timestamps are supported per DAQ row: timestamp0 (host-side) and timestamp1 (device/ECU DAQ timestamp when available in the first ODT). The host timestamp (timestamp0) is normally generated by pyXCP; however, you can enable IEEE 1588/PTP hardware timestamps from the NIC by setting the configuration parameter c.Transport.Eth.ptp_timestamping=True. If hardware timestamps are not available or not enabled, pyXCP continues to generate the host timestamp.

    Timestamps are normalized, i.e.

    • Both values are in nano-seconds.

    • Both timestamps start with 0.

    • Overflows of the DAQ/ECU timestamp are internally handled (linearization).

Quick start example (run_daq)#

The example pyxcp/examples/run_daq.py demonstrates end-to-end DAQ:

from pyxcp.cmdline import ArgumentParser
from pyxcp.daq_stim import DaqList, DaqRecorder, DaqToCsv

ap = ArgumentParser(description="DAQ test")

# Define your DAQ lists (addresses below are examples; adjust for your ECU!)
DAQ_LISTS = [
    DaqList(
        name="pwm_stuff",
        event_num=2,
        stim=False,
        enable_timestamps=True,
        measurements=[
            ("channel1", 0x1BD004, 0, "F32"),
            ("period",   0x001C0028, 0, "F32"),
        ],
        priority=0,
        prescaler=1,
    )
]

# Choose a policy:
daq_parser = DaqToCsv(DAQ_LISTS)             # Online CSV per DAQ list
# daq_parser = DaqRecorder(DAQ_LISTS, "run_daq", 2)  # Offline .xmraw recording

with ap.run(policy=daq_parser) as x:
    x.connect()
    x.cond_unlock("DAQ")
    daq_parser.setup()   # allocate and arm DAQs on the slave
    daq_parser.start()   # start acquisition
    import time; time.sleep(60)
    daq_parser.stop()
    x.disconnect()

xcp_daq_recorder Scipt#

Usage:

xcp_daq_recorder <daq_config>.json -c <pyxcp_config>.py

Description: This script connects to an XCP slave and records DAQ data according to a JSON configuration file. The configuration uses the “daq_lists” key for the DAQ lists. For backward compatibility, a plain JSON array (without a container object) is also supported.

Expected JSON format (example):

{
  "daq_lists": [ ... ],            # list of DAQ lists (or alternatively: JSON root is the list)
  "output_type": "xmraw",          # "xmraw" or "csv"
  "output_file": "xcp_rec_0892026",# filename for xmraw output
  "runtime_seconds": 900           # runtime in seconds (optional)
}

Behavior:

  • output_type == “xmraw”: a DaqRecorder is created (binary/raw) and output_file is passed as filename.

  • output_type == “csv”: a DaqToCsv instance is created; output_file is not used, instead CSV files are created per DAQ list with names derived from the DAQ list name.

  • runtime_seconds determines the sleep time (time.sleep) during recording. If not provided, a default of 60 seconds is used.

Online vs offline and Policies#

  • Online processing: Use DaqToCsv (a DaqOnlinePolicy) to process data in-process and write CSV files on-the-fly. This is a convenient starting point for building custom online processing. Another option is Hdf5OnlinePolicy which records data to HDF5 files using the h5py library.

  • Offline recording: Use DaqRecorder to record into a high-throughput .xmraw file for later, flexible post-processing into multiple formats. This keeps runtime overhead low and allows multiple conversions later.

Pass the chosen policy through ap.run(policy=...). The application wiring ensures incoming DAQ frames are dispatched to the policy.

Timestamps#

If your ECU provides DAQ timestamps in the first ODT and the slave supports timestamps, the DAQ rows include two timestamps:

  • timestamp0: Host-side, generated by pyXCP, UTC 64-bit with nanosecond resolution.

  • timestamp1: ECU/device DAQ timestamp (scaled by the slave’s timestamp unit and ticks).

Example CSV header:

timestamp0,timestamp1,byteCounter,sbyteCounter,wordCounter,dwordCounter

Setting up DAQ lists#

Users primarily define two data structures:

  • pyxcp.daq_stim.DaqList: Corresponds to one XCP DAQ list.

  • measurements: A list of tuples specifying the variables to acquire.

Conceptually, DaqList provides these fields:

name: str                 # used for naming CSV files, SQL tables, MDF channel groups
event_num: int            # event the DAQ list attaches to
stim: bool                # DAQ (False) or STIM (True); STIM is not implemented yet
enable_timestamps: bool   # whether to include timestamps when selectable
measurements: list        # list[ (name, address, address_extension, datatype) ]
priority: int             # optional
prescaler: int            # optional

Data types are given as mnemonics (e.g., U8, I16, F32, F64, F16, BF16). To see all supported mnemonics:

python -c "from pyxcp.recorder import DATA_TYPES; print(DATA_TYPES)"

Note about floating point support: - 16-bit floating-point variables are supported where available from the compiler: F16 (half, 5-bit exponent / 10-bit mantissa) and BF16 (bfloat16, 8-bit exponent / 7-bit mantissa).

Allocation and optimization of ODTs is handled automatically by pyXCP (using bin-packing and continuous block construction internally).

Post-processing .xmraw recordings#

Use pyxcp.recorder.XcpLogFileDecoder as a base class to decode a recorded .xmraw file. Hook into on_daq_list() to consume rows list-wise per DAQ list.

from pathlib import Path
from pyxcp.recorder import XcpLogFileDecoder

class Decoder(XcpLogFileDecoder):
    def __init__(self, recording_file_name: str) -> None:
        self._out = Path(recording_file_name).with_suffix(".txt")

    def initialize(self) -> None:
        self._f = self._out.open("w")

    def finalize(self) -> None:
        self._f.close()

    def on_daq_list(self, daq_list_num: int, timestamp0: int, timestamp1: int, measurements: list) -> None:
        self._f.write(f"{timestamp0},{timestamp1},{measurements}\n")

Decoder("my_recording.xmraw").run()

Converters and examples#

The repository contains examples demonstrating common conversions of .xmraw data:

  • ex_arrow: Create Apache Parquet files (Arrow).

  • ex_mdf: Create ASAM MDF files.

  • ex_sqlite: Create SQLite3 databases.

See pyxcp/examples for these scripts. The converter infrastructure lives under pyxcp/recorder (see also recorder/converter in the source tree if present in your checkout).

Miscellaneous notes#

  • Build system: Poetry is used; use pip install -e . for editable installs.

  • Timestamps are produced by a C++ extension. Startup time (including timezone and DST offsets) is available as x.start_datetime within the application context:

    with ap.run() as x:
    print("Start DT:", x.start_datetime)

  • Re-using an existing interface: you can pass an existing CAN interface into the ArgumentParser context, but ensure your configuration matches the interface type:

    import can
from pyxcp.cmdline import ArgumentParser

can_if = can.Bus(interface="kvaser", channel="0", fd=False, bitrate=500000)
ap = ArgumentParser(description="external interface test")
with ap.run(transport_layer_interface=can_if) as x:
    x.connect()
    x.disconnect()

can_if.shutdown()  # Ownership is not transferred to pyXCP.

    The interface type is currently not deduced, so ensure the configuration matches the passed interface, e.g.:

    c.Transport.layer = 'CAN'

  • STIM is not implemented yet, but some infrastructure exists.