wristpy.processing.processing_utils

This module contains helper functions for general processing of timeseries data.

Functions

get_nonwear_measurements(calibrated_acceleration)

Non-wear measurement dispatcher.

synchronize_measurements(data_measurement, ...)

This function is used to match a Measurement object to a reference Measurement.
wristpy.processing.processing_utils._majority_vote_non_wear(nonwear_measurements: Sequence[Measurement], temporal_resolution: float = 60.0) → Measurement[source]

This function applies a majority vote on any number of nonwear Measurements.

The _time_fix function is used to ensure that all nonwear Measurements have the same start and endpoints. Then each nonwear Measurement is resampled to the same temporal_resolution. A majority vote is taken at each time point to determine the new nonwear Measurement. In case of an even number of inputs, the majority is rounded up. If only one nonwear Measurement is provided, it is returned as is.

Parameters:

  • nonwear_measurements – A sequence (ex. List, Tuple, …) of Measurement objects.

  • temporal_resolution – The temporal resolution of the output, in seconds. Defaults to 60.0.

Returns:

A new Measurement instance with the combined nonwear detection, at a new temporal resolution.

Raises:

ValueError – If the number of nonwear measurements is 0.

wristpy.processing.processing_utils._time_fix(nonwear: Measurement, end_time: datetime, start_time: datetime) → Measurement[source]

Helper function to fix the time of the nonwear measurements.

This function appends start/end points to the nonwear measurements based on previously computed reference start and end points.

Parameters:

  • nonwear – The nonwear measurement to adjust start/end time points.

  • end_time – The maximum end time of the nonwear measurements.

  • start_time – The minimum start time of the nonwear measurements.

Returns:

The nonwear measurement with the time fixed.

wristpy.processing.processing_utils.get_nonwear_measurements(calibrated_acceleration: Measurement, temperature: Measurement | None = None, non_wear_algorithms: Sequence[Literal['ggir', 'cta', 'detach']] = ['ggir']) → Measurement[source]

Non-wear measurement dispatcher.

This function chooses which non-wear detection algorithm(s) to use based on the provided sequence of valid algorithm names.

Parameters:

  • calibrated_acceleration – The calibrated acceleration data

  • temperature – Temperature data if available

  • non_wear_algorithms – One or more algorithm names to use.

Returns:

A non-wear Measurement object.

Raises:

ValueError – If the CTA or DETACH algorithm is requested without temperature data. If an unknown algorithm is specified.

wristpy.processing.processing_utils.synchronize_measurements(data_measurement: Measurement, reference_measurement: Measurement, epoch_length: float = 5.0) → Measurement[source]

This function is used to match a Measurement object to a reference Measurement.

This function ensures that a Measurement object and reference Measurement times are synced up. This is accomplished by first resampling a Measurement object to the specified temporal resolution. It also ensures that a Measurement object is a binary array, where 1 indicates nonwear and 0 indicates wear. It then truncates a Measurement object to match the reference Measurement time points.

Parameters:

  • data_measurement – The Measurement object that requires time syncing.

  • reference_measurement – The reference measurement to use for resampling.

  • epoch_length – The temporal resolution of the output, in seconds. Defaults to 5.0.

Returns:

A new Measurement instance with the cleaned up nonwear detection, returned as a boolean (True == nonwear).