csv_to_wfdb mishandles defaults, scalar broadcasting and checksums

Found while reviewing #591.

The code in csv_to_wfdb that fills in defaults and broadcasts scalars to all channels (

Lines 338 to 409 in b8914e1

    
           units = [units] * n_sig if type(units) is str else units 
        
           # Set the default `fmt` if none exists 
        
           if not fmt: 
        
               fmt = ["16"] * n_sig 
        
           fmt = [fmt] * n_sig if type(fmt) is str else fmt 
        
           if verbose: 
        
               print(f"Signal format: {fmt}") 
        
           # Set the default `adc_gain` if none exists 
        
           if not adc_gain: 
        
               adc_gain = [200] * n_sig 
        
           adc_gain = [adc_gain] * n_sig if type(adc_gain) is int else adc_gain 
        
           if verbose: 
        
               print(f"Signal ADC gain: {adc_gain}") 
        
           # Set the default `baseline` if none exists 
        
           if not baseline: 
        
               if adc_zero: 
        
                   baseline = [adc_zero] * n_sig 
        
               else: 
        
                   baseline = [0] * n_sig 
        
           baseline = [baseline] * n_sig if type(baseline) is int else baseline 
        
           if verbose: 
        
               print(f"Signal baseline: {baseline}") 
        
           # Convert `samps_per_frame` from int to list if necessary 
        
           samps_per_frame = ( 
        
               [samps_per_frame] * n_sig if type(samps_per_frame) is int else samps_per_frame 
        
           ) 
        
           # Convert `skew` from int to list if necessary 
        
           skew = [skew] * n_sig if type(skew) is int else skew 
        
           # Convert `byte_offset` from int to list if necessary 
        
           byte_offset = [byte_offset] * n_sig if type(byte_offset) is int else byte_offset 
        
           # Set the default `adc_res` if none exists 
        
           if not adc_res: 
        
               adc_res = [12] * n_sig 
        
           adc_res = [adc_res] * n_sig if type(adc_res) is int else adc_res 
        
           if verbose: 
        
               print(f"Signal ADC resolution: {adc_res}") 
        
           # Set the default `adc_zero` if none exists 
        
           if not adc_zero: 
        
               adc_zero = [0] * n_sig 
        
           adc_zero = [adc_zero] * n_sig if type(adc_zero) is int else adc_zero 
        
           if verbose: 
        
               print(f"Signal ADC zero: {adc_zero}") 
        
           # Set the default `init_value` 
        
           # NOTE: Initial value (and subsequently the digital signal) won't be correct 
        
           # unless the correct `baseline` and `adc_gain` are provided... this is just 
        
           # the best approximation 
        
           if not init_value: 
        
               init_value = p_signal[0, :] 
        
               init_value = baseline + (np.array(adc_gain) * init_value) 
        
               init_value = [int(i) for i in init_value.tolist()] 
        
           if verbose: 
        
               print(f"Signal initial value: {init_value}") 
        
           # Set the default `checksum` 
        
           if not checksum: 
        
               checksum = [int(np.sum(v) % 65536) for v in np.transpose(p_signal)] 
        
           if verbose: 
        
               print(f"Signal checksum: {checksum}") 
        
           # Set the default `block_size` 
        
           if not block_size: 
        
               block_size = [0] * n_sig 
        
           block_size = [block_size] * n_sig if type(block_size) is int else block_size

) has several problems:

Scalars are only broadcast when type(x) is int (or str). NumPy integers and floats such as adc_gain=200.5 are not broadcast to a list.
Defaults are applied with if not x:, so an explicit 0 is treated as missing. For example, baseline=0 with adc_zero=1024 gives a baseline of 1024.

When adc_zero is a list and baseline is not given, the default baseline becomes [adc_zero] * n_sig, a list of lists (

wfdb-python/src/wfdb/io/convert/csv.py

Lines 355 to 357 in b8914e1

    
           if not baseline: 
        
               if adc_zero: 
        
                   baseline = [adc_zero] * n_sig

).

With record_only=True, the default checksum is computed from the physical values rather than the digital samples (

wfdb-python/src/wfdb/io/convert/csv.py

Lines 401 to 402 in b8914e1

    
           if not checksum: 
        
               checksum = [int(np.sum(v) % 65536) for v in np.transpose(p_signal)]

), and init_value is an approximation from baseline + adc_gain * p_signal[0] (

wfdb-python/src/wfdb/io/convert/csv.py

Lines 393 to 396 in b8914e1

    
           if not init_value: 
        
               init_value = p_signal[0, :] 
        
               init_value = baseline + (np.array(adc_gain) * init_value) 
        
               init_value = [int(i) for i in init_value.tolist()]

).

The checksum and initial value could be computed from the digital signal (e.g. via Record.adc()) instead. This may be easiest to address together with #592.

Navigation Menu

Repository navigation

csv_to_wfdb mishandles defaults, scalar broadcasting and checksums #593

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	units = [units] * n_sig if type(units) is str else units

	# Set the default `fmt` if none exists
	if not fmt:
	fmt = ["16"] * n_sig
	fmt = [fmt] * n_sig if type(fmt) is str else fmt
	if verbose:
	print(f"Signal format: {fmt}")

	# Set the default `adc_gain` if none exists
	if not adc_gain:
	adc_gain = [200] * n_sig
	adc_gain = [adc_gain] * n_sig if type(adc_gain) is int else adc_gain
	if verbose:
	print(f"Signal ADC gain: {adc_gain}")

	# Set the default `baseline` if none exists
	if not baseline:
	if adc_zero:
	baseline = [adc_zero] * n_sig
	else:
	baseline = [0] * n_sig
	baseline = [baseline] * n_sig if type(baseline) is int else baseline
	if verbose:
	print(f"Signal baseline: {baseline}")

	# Convert `samps_per_frame` from int to list if necessary
	samps_per_frame = (
	[samps_per_frame] * n_sig if type(samps_per_frame) is int else samps_per_frame
	)

	# Convert `skew` from int to list if necessary
	skew = [skew] * n_sig if type(skew) is int else skew

	# Convert `byte_offset` from int to list if necessary
	byte_offset = [byte_offset] * n_sig if type(byte_offset) is int else byte_offset

	# Set the default `adc_res` if none exists
	if not adc_res:
	adc_res = [12] * n_sig
	adc_res = [adc_res] * n_sig if type(adc_res) is int else adc_res
	if verbose:
	print(f"Signal ADC resolution: {adc_res}")

	# Set the default `adc_zero` if none exists
	if not adc_zero:
	adc_zero = [0] * n_sig
	adc_zero = [adc_zero] * n_sig if type(adc_zero) is int else adc_zero
	if verbose:
	print(f"Signal ADC zero: {adc_zero}")

	# Set the default `init_value`
	# NOTE: Initial value (and subsequently the digital signal) won't be correct
	# unless the correct `baseline` and `adc_gain` are provided... this is just
	# the best approximation
	if not init_value:
	init_value = p_signal[0, :]
	init_value = baseline + (np.array(adc_gain) * init_value)
	init_value = [int(i) for i in init_value.tolist()]
	if verbose:
	print(f"Signal initial value: {init_value}")

	# Set the default `checksum`
	if not checksum:
	checksum = [int(np.sum(v) % 65536) for v in np.transpose(p_signal)]
	if verbose:
	print(f"Signal checksum: {checksum}")

	# Set the default `block_size`
	if not block_size:
	block_size = [0] * n_sig
	block_size = [block_size] * n_sig if type(block_size) is int else block_size

csv_to_wfdb mishandles defaults, scalar broadcasting and checksums #593

Description

Activity

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions