DFS400 Pulse Testing Guide: Device Connection, Pulse Interpretation, and Data Export

This guide explains how to perform shock tests using the DFS400 SDK. It covers device connection, pulse receiving and conversion, and the export of pulse data and waveforms. To handle real-time data streaming from hardware, an asynchronous event-driven architecture is used.

Reference Code: pulse_test.py

1. Declare DFS400 Instance and Connect to Device

First, create an instance of DFS400 and connect to your device. Once the connection succeeds, a communication handle is returned for use in all subsequent API calls.

handle = device.connect(c_uint(-1), c_uint(5000))
if handle is None:
    print('Error: connect failed')
    sys.exit()

The above example sets the connection timeout of 5 seconds.
If the COM port number is known, you can specify it directly:

# device is on COM9 for example
handle = device.connect(9, c_uint(5000))

Verify connection by printing serial numbers:

print(f'Device is connected ({device.get_serial_number(handle)})')

2. Set The Device Load Resistance

Body resistance varies from person to person. To emulate different body resistance conditions, the DFSSetLoadResistance API allows you to configure the desired resistance value (default is 50Ω). The supported values are noted in DFS400.h. You can use the DFSGetLoadResistance API to get the current load resistance setting.

Example Code

#(dfs400.py)
@sdk_fn("DFSSetLoadResistance", [c_void_p, c_int], c_bool)
def set_resistance(self, handle, load_resistance) -> c_bool:
    pass

@sdk_fn("DFSGetLoadResistance", [c_void_p], c_int)
def get_load_resistance(self, handle) -> c_int:
    pass

#(pulse_test.py)
# Set and verify resistance
target_resistance = 50 
device.set_resistance(handle, int(target_resistance))
print(f'Resistance is set to {device.get_load_resistance(handle)} Ohm.')

3. Start Test

The DFSStartTest API starts the testing process. It requires the output pattern, a pulse-wait timeout, and a callback function to handle the test result. The timeout defines the number of milliseconds to wait for received data, or -1 to wait indefinitely. The callback function will be introduced next.

Example Code

testResultCb = TestResultCallback(TestResultHandler)
# Start the test with a timeout of 40 seconds. The TestResultHandler will handle the test result.
if not device.start_test(handle, ECGPattern.ECGPatternVT.value, c_uint(40000), testResultCb):
    print('Error: output ECG failed')
    break

4. TestResultCallback Function

The TestResultCallback function acts as a listener, whenever the DFS400 receives a pulse, the callback function is invoked. Here are some tasks that this callback function is expected to perform:

(1) Get the Pulse Result

Cast the raw result pointer to a TEST_RESULT structure to access the data fields in Python:

test_result = cast(result, POINTER(TEST_RESULT)).contents

The TEST_RESULT structure is defined in dfs400.py:

class TEST_RESULT (Structure):
    _fields_ = [
        ('J', c_double),
        ('data', POINTER(c_int)),
        ('size', c_uint),
        ('WaveformType', c_int),
        ('StartPosition', c_int),
        ('EndPosition', c_int)
    ]

(2) Data Conversion

The pulse data captured by the DFS400 is transmitted to the SDK as raw digital samples. To obtain the physical voltage, a linear scaling transformation must be applied by using hardware-specific calibration constants (voltage gain and voltage offset).

$V = (Raw \times Gain) + Offset$

vol_gain = c_double(0.0)
vol_offset = c_double(0.0)
sample_rate = device.get_sample_rate(handle)
device.get_raw_to_voltage_calibration_data(handle, byref(vol_gain), byref(vol_offset))

raw_adc_data_current_pulse = [test_result.data[i] for i in range(test_result.size)]

current_x = [float(i) / sample_rate for i in range(test_result.size)]
current_y = [(raw_adc_data_current_pulse[i] * vol_gain.value + vol_offset.value) for i in range(test_result.size)]

(3) Verify the Pulse Energy(J)

In practice, higher pulse energy is typically required as the load resistance increases. Therefore, this example employs an adaptive energy tolerance and target energy. Since the actual resistance may vary slightly between different DFS400 hardware, call device.get_resistance() to obtain the exact resistance.

current_resistance = device.get_resistance(handle)

# Adaptively set tolerance_percent and target_energy based on the resistance value
if current_resistance <= 50.0:
    tolerance_percent = 15.0
    target_energy = 150.0
elif current_resistance <= 100.0:
    tolerance_percent = 12.0
    target_energy = 200.0
elif current_resistance <= 150.0:
    tolerance_percent = 9.0
    target_energy = 250.0
else: # For resistance > 150.0 (e.g., 175, 200, or close values)
    tolerance_percent = 6.0
    target_energy = 300.0

lower_bound = target_energy * (1 - tolerance_percent / 100)
upper_bound = target_energy * (1 + tolerance_percent / 100)

actual_energy = test_result.J

if lower_bound <= actual_energy <= upper_bound:
    judgment_result = 'Pass'
else:
    judgment_result = 'Fail'

You may modify the tolerance and target energy values according to your testing requirements.

(4) Waveform Parameter Parsing

Once a pulse is captured, the SDK identifies its waveform type (e.g., BTE, MDP, MUD). Inside the TestResultCallback function, a result parsing function is invoked based on the detected waveform type to parse the returned result structure into waveform parameters.

The parameters mainly focus on these dimensions:

Peak Values: Maximum voltage(VMax) and maximum current (IMax)
Timing: Phase duration, start/end time
Morphology: The waveform slope and tilt percentage (the decay of the voltage)

(5) Data Export

Two CSV files are exported in this example:

Analysis Summary: Results of phase analysis and pulse energy judgment.
Raw Data: Detailed, sample-by-sample records of time, raw data, voltage, and current ( $I = V/R$ ).

Function Return

The return value of the callback function determines whether the device continues monitoring for subsequent pulses.

return True (continue testing): The pulse-wait timeout (initially set in start_test()) will be reset, if the caller wants the device to continue detecting the next shock without ending the current session.
return False (terminate testing): This will trigger a test result with a size of 0, which indicates the test is completed and the output waveform will be stopped automatically.

Example Code

def TestResultHandler(handle, result):
    """
    The callback runs in a background thread managed by the SDK.
    The function is called when the device returns test results.
    It processes the waveform data, exports the data to CSV, and prepares data for plotting in the main thread.
    """
    global latest_x, latest_y, latest_raw_data_pulse, latest_resistance, latest_waveform_type_str, latest_judgment_result
    global latest_analysis_csv_path, latest_raw_csv_path

    # 1) Get the Pulse Result
    test_result = cast(result, POINTER(TEST_RESULT)).contents


    if test_result.size == 0:
        print('Test completed. No more shocks will be received.')
        test_completed_event.set()
        return c_bool(False)

    print("New shock data received. Processing...")


    # 2) Data Conversion
    vol_gain = c_double(0.0)
    vol_offset = c_double(0.0)
    sample_rate = device.get_sample_rate(handle)
    device.get_raw_to_voltage_calibration_data(handle, byref(vol_gain), byref(vol_offset))

    # print("Calibration Voltage Gain:", vol_gain.value)
    # print("Calibration Voltage Offset:", vol_offset.value)

    raw_data_pulse = [test_result.data[i] for i in range(test_result.size)]

    current_x = [float(i) / sample_rate for i in range(test_result.size)]
    current_y = [(raw_data_pulse[i] * vol_gain.value + vol_offset.value) for i in range(test_result.size)]


    # 3) Verify the Pulse Energy(J)
    analysis_results = {}

    analysis_results['Waveform Type'] = waveform_types.get(test_result.WaveformType, 'unknown')
    analysis_results['Energy (J)'] = f'{test_result.J:.2f}'
    
    current_resistance = device.get_resistance(handle)
    analysis_results['Resistance (Ohm)'] = f'{current_resistance}'

    # Adaptively set tolerance_percent and target_energy based on the resistance value
    if current_resistance <= 50.0:
        tolerance_percent = 15.0
        target_energy = 150.0
    elif current_resistance <= 100.0:
        tolerance_percent = 12.0
        target_energy = 200.0
    elif current_resistance <= 150.0:
        tolerance_percent = 9.0
        target_energy = 250.0
    else: # For resistance > 150.0 (e.g., 175, 200, or close values)
        tolerance_percent = 6.0
        target_energy = 300.0

    lower_bound = target_energy * (1 - tolerance_percent / 100)
    upper_bound = target_energy * (1 + tolerance_percent / 100)
    
    actual_energy = test_result.J
    
    if lower_bound <= actual_energy <= upper_bound:
        judgment_result = 'Pass'
    else:
        judgment_result = 'Fail'
    
    analysis_results['Judgment Result'] = judgment_result

    print(f"Waveform Type = {analysis_results['Waveform Type']}")
    print(f"Energy = {analysis_results['Energy (J)']} J")
    print(f"Resistance: {analysis_results['Resistance (Ohm)']} Ohm")

    print(f"Target Energy: {target_energy:.2f} J")
    print(f"Tolerance: {tolerance_percent:.2f}%")
    print(f"Energy Range: {lower_bound:.2f} J ~ {upper_bound:.2f} J")
    print(f"Judgment Result: {analysis_results['Judgment Result']}")


    # 4) Waveform Parameters Parsing
    waveDump = {
        PulseWaveformType.PulseWaveformUnknown.value: show_pulse_waveform_unknown,
        PulseWaveformType.PulseWaveformMDP.value: show_pulse_waveform_mdp,
        PulseWaveformType.PulseWaveformMUD.value: show_pulse_waveform_mud,
        PulseWaveformType.PulseWaveformMTE.value: show_pulse_waveform_mte,
        PulseWaveformType.PulseWaveformBTE.value: show_pulse_waveform_bte,
        PulseWaveformType.PulseWaveformPBW.value: show_pulse_waveform_pbw,
        PulseWaveformType.PulseWaveformRBW.value: show_pulse_waveform_bte,
        PulseWaveformType.PulseWaveformABW.value: show_pulse_waveform_abw,
        PulseWaveformType.PulseWaveformSAW.value: show_pulse_waveform_bte
    }

    waveDump.get(test_result.WaveformType,
                 PulseWaveformType.PulseWaveformUnknown.value)(handle, result, analysis_results)

    
    # 5) Data Export
    # Store the paths of the exported files, they will be printed in the main thread
    latest_analysis_csv_path = export_results_to_csv(analysis_results, waveform_type_str, judgment_result)
    latest_raw_csv_path = export_raw_data_to_csv(current_x, raw_data_pulse, current_y, current_resistance, waveform_type_str, judgment_result)

    # Store the latest data for plotting in the main thread
    latest_x = current_x
    latest_y = current_y
    latest_raw_data_pulse = raw_data_pulse
    latest_resistance = current_resistance
    latest_waveform_type_str = waveform_type_str
    latest_judgment_result = judgment_result
    pulse_data_ready_event.set()


    # Return:
    # False - Terminate Testing. Trigger the size == 0 callback and will set test_completed_event.
    # True - Continue Detecting. The timeout will be reset
    return c_bool(False)

5. Multithreading

Since the callback is executed in a background thread, any data shared with the main thread should be synchronized using thread-safe mechanisms such as threading.Event.

Two events are used in our example:

pulse_data_ready_event: Set inside the callback after the test result data processing is finished. It signals the Main Thread that data is ready for waveform plotting.
test_completed_event: Set when the callback receives a test result with a size of 0. The main thread waits for this event before proceeding to the next test session.

Synchronization Workflow

Initialization: Before calling start_test(), both pulse_data_ready_event.clear() and test_completed_event.clear() are called to reset their states.
Main Thread Blocking: The main thread uses .wait(timeout) to block execution until the events are set. Note that the timeout for pulse_data_ready_event.wait() should be longer than the pulse-wait timeout set in start_test(), to ensure sufficient time for pulse data receiving and parsing.
Callback Thread: Once the data has been parsed and the CSVs have been written, pulse_data_ready_event.set() and test_completed_event.set() are called.

The following diagram shows the synchronization flow used in this example.

Notes: Hardware Protection & Cool-down

To maintain accuracy and protect the circuitry, allow a break of at least 15 seconds between pulse tests.

Example Code

print("Test completed. Cool down for 15s")
for i in range(15, 0, -1):
    print(i, end=' ', flush=True)
    time.sleep(1)

You can use the DFSIsOverTemperature API to check if the temperature of DFS400 is too high.

while device.is_over_temperature(handle):
    print("Device is over temperature. Wait 10 seconds and re-check...")
    time.sleep(10)
    
device.start_test(handle, ......