wristpy.core.orchestrator

  1"""Python based runner."""
  2
  3import itertools
  4import logging
  5import pathlib
  6from typing import Dict, Literal, Optional, Sequence, Tuple, Union
  7
  8from wristpy.core import config, exceptions, models
  9from wristpy.io.readers import readers
 10from wristpy.io.writers import writers
 11from wristpy.processing import (
 12    analytics,
 13    calibration,
 14    idle_sleep_mode_imputation,
 15    metrics,
 16    processing_utils,
 17)
 18
 19logger = config.get_logger()
 20
 21VALID_FILE_TYPES = (".csv", ".parquet")
 22
 23DEFAULT_ACTIVITY_THRESHOLDS = {
 24    "enmo": (0.0563, 0.1916, 0.6958),
 25    "mad": (0.029, 0.338, 0.604),
 26    "ag_count": (100, 3000, 5200),
 27    "mims": (10.558, 15.047, 19.614),
 28}
 29
 30
 31def run(
 32    input: Union[pathlib.Path, str],
 33    output: Optional[Union[pathlib.Path, str]] = None,
 34    thresholds: Optional[Sequence[Tuple[float, float, float]]] = None,
 35    calibrator: Union[
 36        None,
 37        Literal["ggir", "gradient"],
 38    ] = "gradient",
 39    epoch_length: float = 5,
 40    activity_metric: Sequence[Literal["enmo", "mad", "ag_count", "mims"]] = ["enmo"],
 41    nonwear_algorithm: Sequence[Literal["ggir", "cta", "detach"]] = ["ggir"],
 42    verbosity: int = logging.WARNING,
 43    output_filetype: Literal[".csv", ".parquet"] = ".csv",
 44) -> Union[writers.OrchestratorResults, Dict[str, writers.OrchestratorResults]]:
 45    """Runs main processing steps for wristpy on single files, or directories.
 46
 47    The run() function will execute the run_file() function on individual files, or
 48    run_directory() on entire directories. When the input path points to a file, the
 49    name of the save file will be taken from the given output path (if any). When the
 50    input path points to a directory the output path must be a valid directory as well.
 51    Output file names will be derived from original file names in the case of directory
 52    processing.
 53
 54    Args:
 55        input: Path to the input file or directory of files to be read. Currently,
 56            this supports .bin and .gt3x
 57        output: Path to directory data will be saved to. If processing a single file the
 58            path should end in the save file name in either .csv or .parquet formats.
 59        thresholds: The cut points for the light, moderate, and vigorous thresholds,
 60            given in that order. One threshold tuple must be provided for each activity
 61            metric, in the same order the metrics were specified. To use default values
 62            for all metrics, leave thresholds as None. Values must be ascending, unique,
 63            and greater than 0. Default values are optimized for subjects ages 7-11 [1]
 64            [3].
 65        calibrator: The calibrator to be used on the input data.
 66        epoch_length: The temporal resolution in seconds, the data will be down sampled
 67            to. It must be > 0.0.
 68        activity_metric: The metric(s) to be used for physical activity categorization.
 69            Multiple metrics can be specified as a sequence.
 70        nonwear_algorithm: The algorithm to be used for nonwear detection.
 71        verbosity: The logging level for the logger.
 72        output_filetype: Specifies the data format for the save files. Only used when
 73            processing directories.
 74
 75    Returns:
 76        All calculated data in a save ready format as a Results object or as a
 77        dictionary of OrchestratorResults objects.
 78
 79    Raises:
 80        ValueError: If the number of physical activity thresholds does not match the
 81            number of provided activity metrics.
 82        ValueError: If the physical activity thresholds are not unique or not in
 83            ascending order.
 84
 85    References:
 86        [1] Hildebrand, M., et al. (2014). Age group comparability of raw accelerometer
 87        output from wrist- and hip-worn monitors. Medicine and Science in Sports and
 88        Exercise, 46(9), 1816-1824.
 89        [2] Treuth MS, Schmitz K, Catellier DJ, McMurray RG, Murray DM, Almeida MJ,
 90        Going S, Norman JE, Pate R. Defining accelerometer thresholds for activity
 91        intensities in adolescent girls. Med Sci Sports Exerc. 2004 Jul;36(7):1259-66.
 92        PMID: 15235335; PMCID: PMC2423321.
 93        [3] Karas M, Muschelli J, Leroux A, Urbanek J, Wanigatunga A, Bai J,
 94        Crainiceanu C, Schrack J Comparison of Accelerometry-Based Measures of Physical
 95        Activity: Retrospective Observational Data Analysis Study JMIR Mhealth Uhealth
 96        2022;10(7):e38077 URL: https://mhealth.jmir.org/2022/7/e38077 DOI: 10.2196/38077
 97    """
 98    logger.setLevel(verbosity)
 99
100    input = pathlib.Path(input)
101    output = pathlib.Path(output) if output is not None else None
102
103    if thresholds is not None:
104        if len(activity_metric) != len(thresholds):
105            raise ValueError(
106                "Number of thresholds did not match the number of activity metrics. "
107                "Provide one threshold tuple per metric or use None for defaults."
108            )
109        metrics_dict = dict(zip(activity_metric, thresholds))
110    else:
111        metrics_dict = {
112            metric: DEFAULT_ACTIVITY_THRESHOLDS[metric] for metric in activity_metric
113        }
114
115    for metric, thresh in metrics_dict.items():
116        if not (0 <= thresh[0] < thresh[1] < thresh[2]):
117            message = (
118                f"Invalid thresholds for {metric}."
119                f" Threshold values must be >=0, unique, and in ascending order."
120            )
121            logger.error(message)
122            raise ValueError(message)
123
124    if input.is_file():
125        return _run_file(
126            input=input,
127            output=output,
128            thresholds=thresholds,
129            calibrator=calibrator,
130            epoch_length=epoch_length,
131            activity_metric=activity_metric,
132            verbosity=verbosity,
133            nonwear_algorithm=nonwear_algorithm,
134        )
135
136    return _run_directory(
137        input=input,
138        output=output,
139        thresholds=thresholds,
140        calibrator=calibrator,
141        epoch_length=epoch_length,
142        activity_metric=activity_metric,
143        verbosity=verbosity,
144        output_filetype=output_filetype,
145        nonwear_algorithm=nonwear_algorithm,
146    )
147
148
149def _run_directory(
150    input: pathlib.Path,
151    output: Optional[pathlib.Path] = None,
152    thresholds: Optional[Sequence[Tuple[float, float, float]]] = None,
153    calibrator: Union[
154        None,
155        Literal["ggir", "gradient"],
156    ] = "gradient",
157    epoch_length: float = 5,
158    nonwear_algorithm: Sequence[Literal["ggir", "cta", "detach"]] = ["ggir"],
159    verbosity: int = logging.WARNING,
160    output_filetype: Literal[".csv", ".parquet"] = ".csv",
161    activity_metric: Sequence[Literal["enmo", "mad", "ag_count", "mims"]] = ["enmo"],
162) -> Dict[str, writers.OrchestratorResults]:
163    """Runs main processing steps for wristpy on directories.
164
165    The _run_directory() function will execute the _run_file() function on entire
166    directories. The input and output (if any) paths must be directories. Output file
167    names will be derived from input file names.
168
169    Args:
170        input: Path to the input directory of files to be read. Currently,
171            this supports .bin and .gt3x
172        output: Path to directory data will be saved to.
173        thresholds: The cut points for the light, moderate, and vigorous thresholds,
174            given in that order. One threshold tuple must be provided for each activity
175            metric, in the same order the metrics were specified. To use default values
176            for all metrics, leave thresholds as None. Values must be ascending, unique,
177            and greater than 0. Default values are optimized for subjects ages 7-11
178            [1][2].
179        calibrator: The calibrator to be used on the input data.
180        epoch_length: The temporal resolution in seconds, the data will be down sampled
181            to. It must be > 0.0.
182        nonwear_algorithm: The algorithm to be used for nonwear detection.
183        verbosity: The logging level for the logger.
184        output_filetype: Specifies the data format for the save files.
185        activity_metric: The metric(s) to be used for physical activity categorization.
186            Multiple metrics can be specified as a sequence.
187
188    Returns:
189        All calculated data in a save ready format as a dictionary of
190        OrchestratorResults objects.
191
192    Raises:
193        ValueError: If the output given is not a directory.
194        ValueError: If the output_filetype is not a valid type.
195        FileNotFoundError: If the input directory contained no files of a valid type.
196
197    References:
198        [1] Hildebrand, M., et al. (2014). Age group comparability of raw accelerometer
199        output from wrist- and hip-worn monitors. Medicine and Science in Sports and
200        Exercise, 46(9), 1816-1824.
201        [2] Karas M, Muschelli J, Leroux A, Urbanek J, Wanigatunga A, Bai J,
202        Crainiceanu C, Schrack J Comparison of Accelerometry-Based Measures of Physical
203        Activity: Retrospective Observational Data Analysis Study JMIR Mhealth Uhealth
204        2022;10(7):e38077 URL: https://mhealth.jmir.org/2022/7/e38077 DOI: 10.2196/38077
205    """
206    if output is not None:
207        if output.is_file():
208            raise ValueError(
209                "Output is a file, but must be a directory when input is a directory."
210            )
211        if output_filetype not in VALID_FILE_TYPES:
212            raise ValueError(
213                "Invalid output_filetype: "
214                f"{output_filetype}. Valid options are: {VALID_FILE_TYPES}."
215            )
216
217    file_names = list(itertools.chain(input.glob("*.gt3x"), input.glob("*.bin")))
218
219    if not file_names:
220        raise exceptions.EmptyDirectoryError(
221            f"Directory {input} contains no .gt3x or .bin files."
222        )
223    results_dict = {}
224    for file in file_names:
225        output_file_path = (
226            output / pathlib.Path(file.stem).with_suffix(output_filetype)
227            if output
228            else None
229        )
230        logger.debug(
231            "Processing directory: %s, current file: %s, save path: %s",
232            input,
233            file,
234            output_file_path,
235        )
236        try:
237            results_dict[str(file)] = _run_file(
238                input=input / file,
239                output=output_file_path,
240                thresholds=thresholds,
241                calibrator=calibrator,
242                epoch_length=epoch_length,
243                verbosity=verbosity,
244                nonwear_algorithm=nonwear_algorithm,
245                activity_metric=activity_metric,
246            )
247        except Exception as e:
248            logger.error("Did not run file: %s, Error: %s", file, e)
249    logger.info("Processing for directory %s completed successfully.", output)
250    return results_dict
251
252
253def _run_file(
254    input: pathlib.Path,
255    output: Optional[pathlib.Path] = None,
256    thresholds: Optional[Sequence[Tuple[float, float, float]]] = None,
257    calibrator: Union[
258        None,
259        Literal["ggir", "gradient"],
260    ] = "gradient",
261    epoch_length: float = 5.0,
262    activity_metric: Sequence[Literal["enmo", "mad", "ag_count", "mims"]] = ["enmo"],
263    nonwear_algorithm: Sequence[Literal["ggir", "cta", "detach"]] = ["ggir"],
264    verbosity: int = logging.WARNING,
265) -> writers.OrchestratorResults:
266    """Runs main processing steps for wristpy and returns data for analysis.
267
268    The run_file() function will provide the user with the specified physical activity
269    metric(s), anglez, physical activity levels, detected sleep periods, and nonwear
270    data. All measures will be in the same temporal resolution. Users may choose from
271    'ggir' and 'gradient' calibration methods, or enter None to proceed without
272    calibration.
273
274    Args:
275        input: Path to the input file to be read. Currently, this supports .bin and
276            .gt3x
277        output: Path to save data to. The path should end in the save file name in
278            either .csv or .parquet formats.
279        thresholds: The cut points for the light, moderate, and vigorous thresholds,
280            given in that order. One threshold tuple must be provided for each activity
281            metric, in the same order the metrics were specified. To use default values
282            for all metrics, leave thresholds as None. Values must be ascending, unique,
283            and greater than 0. Default values are optimized for subjects ages 7-11
284            [1]-[3].
285        calibrator: The calibrator to be used on the input data.
286        epoch_length: The temporal resolution in seconds, the data will be down sampled
287            to. It must be > 0.0.
288        activity_metric: The metric(s) to be used for physical activity categorization.
289            Multiple metrics can be specified as a sequence.
290        nonwear_algorithm: The algorithm to be used for nonwear detection. A sequence of
291            algorithms can be provided. If so, a majority vote will be taken.
292        verbosity: The logging level for the logger.
293
294    Returns:
295        All calculated data in a save ready format as a OrchestratorResults object.
296
297    Raises:
298        ValueError: If the number of physical activity thresholds does not match the
299            number of provided activity metrics.
300        ValueError: If the physical activity thresholds are not unique or not in
301            ascending order.
302        ValueError: If an invalid Calibrator is chosen.
303        ValueError: If epoch_length is not greater than 0.
304
305    References:
306        [1] Hildebrand, M., et al. (2014). Age group comparability of raw accelerometer
307        output from wrist- and hip-worn monitors. Medicine and Science in Sports and
308        Exercise, 46(9), 1816-1824.
309        [2] Aittasalo, M., Vähä-Ypyä, H., Vasankari, T. et al. Mean amplitude deviation
310        calculated from raw acceleration data: a novel method for classifying the
311        intensity of adolescents' physical activity irrespective of accelerometer brand.
312        BMC Sports Sci Med Rehabil 7, 18 (2015). https://doi.org/10.1186/s13102-015-0010-0.
313        [3] Karas M, Muschelli J, Leroux A, Urbanek J, Wanigatunga A, Bai J,
314        Crainiceanu C, Schrack J Comparison of Accelerometry-Based Measures of Physical
315        Activity: Retrospective Observational Data Analysis Study JMIR Mhealth Uhealth
316        2022;10(7):e38077 URL: https://mhealth.jmir.org/2022/7/e38077 DOI: 10.2196/38077
317    """
318    logger.setLevel(verbosity)
319    if output is not None:
320        writers.OrchestratorResults.validate_output(output=output)
321
322    if calibrator is not None and calibrator not in ["ggir", "gradient"]:
323        msg = (
324            f"Invalid calibrator: {calibrator}. Choose: 'ggir', 'gradient'. "
325            "Enter None if no calibration is desired."
326        )
327        logger.error(msg)
328        raise ValueError(msg)
329
330    if epoch_length <= 0:
331        msg = "Epoch_length must be greater than 0."
332        logger.error(msg)
333        raise ValueError(msg)
334
335    watch_data = readers.read_watch_data(input)
336
337    if calibrator is None:
338        logger.debug("Running without calibration")
339        calibrated_acceleration = watch_data.acceleration
340    else:
341        calibrated_acceleration = calibration.CalibrationDispatcher(calibrator).run(
342            watch_data.acceleration, return_input_on_error=True
343        )
344
345    if watch_data.idle_sleep_mode_flag:
346        logger.debug("Imputing idle sleep mode gaps.")
347        calibrated_acceleration = (
348            idle_sleep_mode_imputation.impute_idle_sleep_mode_gaps(
349                calibrated_acceleration
350            )
351        )
352
353    anglez = metrics.angle_relative_to_horizontal(
354        calibrated_acceleration, epoch_length=epoch_length
355    )
356
357    metrics_dict: Dict[
358        Literal["enmo", "mad", "ag_count", "mims"], Tuple[float, float, float]
359    ]
360    if thresholds is not None:
361        metrics_dict = dict(zip(activity_metric, thresholds))
362    else:
363        metrics_dict = {
364            metric: DEFAULT_ACTIVITY_THRESHOLDS[metric] for metric in activity_metric
365        }
366
367    activity_measurements_list = []
368    physical_activity_levels_list = []
369    for metric_name, thresh in metrics_dict.items():
370        metric_measurement = _compute_activity(
371            calibrated_acceleration,
372            metric_name,
373            epoch_length,
374            dynamic_range=watch_data.dynamic_range,
375        )
376        activity_measurements_list.append(metric_measurement)
377        physical_activity_levels_list.append(
378            analytics.compute_physical_activty_categories(
379                metric_measurement,
380                thresh,
381                name=f"{metric_name} physical activity levels",
382            )
383        )
384
385    nonwear_array = processing_utils.get_nonwear_measurements(
386        calibrated_acceleration=calibrated_acceleration,
387        temperature=watch_data.temperature,
388        non_wear_algorithms=nonwear_algorithm,
389    )
390
391    nonwear_epoch = processing_utils.synchronize_measurements(
392        data_measurement=nonwear_array,
393        reference_measurement=anglez,
394        epoch_length=epoch_length,
395    )
396
397    sleep_detector = analytics.GgirSleepDetection(anglez)
398    sleep_parameters = sleep_detector.run_sleep_detection()
399    sleep_array = analytics.sleep_cleanup(
400        sleep_windows=sleep_parameters.sleep_windows, nonwear_measurement=nonwear_epoch
401    )
402    spt_windows = analytics.sleep_bouts_cleanup(
403        sleep_parameter=sleep_parameters.spt_windows,
404        nonwear_measurement=nonwear_epoch,
405        time_reference_measurement=anglez,
406        epoch_length=epoch_length,
407    )
408    sib_periods = analytics.sleep_bouts_cleanup(
409        sleep_parameter=sleep_parameters.sib_periods,
410        nonwear_measurement=nonwear_epoch,
411        time_reference_measurement=anglez,
412        epoch_length=epoch_length,
413    )
414
415    parameters_dictionary = {
416        "thresholds": list(thresholds) if thresholds is not None else None,
417        "calibrator": calibrator,
418        "epoch_length": epoch_length,
419        "activity_metric": activity_metric,
420        "nonwear_algorithm": list(nonwear_algorithm),
421        "input_file": str(input),
422    }
423
424    results = writers.OrchestratorResults(
425        physical_activity_metric=activity_measurements_list,
426        anglez=anglez,
427        physical_activity_levels=physical_activity_levels_list,
428        sleep_status=sleep_array,
429        sib_periods=sib_periods,
430        spt_periods=spt_windows,
431        nonwear_status=nonwear_epoch,
432        processing_params=parameters_dictionary,
433    )
434    if output is not None:
435        try:
436            results.save_results(output=output)
437        except (
438            exceptions.InvalidFileTypeError,
439            PermissionError,
440            FileExistsError,
441        ) as exc_info:
442            # Allowed to pass to recover in Jupyter Notebook scenarios.
443            logger.error(
444                (
445                    "Could not save output due to: %s. Call save_results "
446                    "on the output object with a correct filename to save these "
447                    "results.",
448                    exc_info,
449                )
450            )
451    logger.info("Processing for %s completed successfully.", input.stem)
452    return results
453
454
455def _compute_activity(
456    acceleration: models.Measurement,
457    activity_metric: Literal["ag_count", "mad", "enmo", "mims"],
458    epoch_length: float,
459    dynamic_range: Optional[tuple[float, float]],
460) -> models.Measurement:
461    """This is a helper function to organize the computation of the activity metric.
462
463    This function organizes the logic for computing the requested physical activity
464    metric at the desired temporal resolution.
465
466    Args:
467        acceleration: The acceleration data to compute the activity metric from.
468        activity_metric: The metric to be used for physical activity categorization.
469        epoch_length: The temporal resolution in seconds, the data will be down sampled
470            to.
471        dynamic_range: Tuple of the minimum and maximum accelerometer values. This
472            argument is only relevant to the mims metric. Values are taken from watch
473            metadata, if no metadata could be extracted, the default
474            values of (-8,8) are used.
475
476    Returns:
477        A Measurement object with the computed physical activity metric.
478    """
479    if activity_metric == "ag_count":
480        return metrics.actigraph_activity_counts(
481            acceleration, epoch_length=epoch_length, name=activity_metric
482        )
483    elif activity_metric == "mad":
484        return metrics.mean_amplitude_deviation(
485            acceleration, epoch_length=epoch_length, name=activity_metric
486        )
487    elif activity_metric == "mims":
488        if dynamic_range is None:
489            return metrics.monitor_independent_movement_summary_units(
490                acceleration, epoch=epoch_length, name=activity_metric
491            )
492        return metrics.monitor_independent_movement_summary_units(
493            acceleration,
494            epoch=epoch_length,
495            dynamic_range=dynamic_range,
496            name=activity_metric,
497        )
498
499    return metrics.euclidean_norm_minus_one(
500        acceleration, epoch_length=epoch_length, name=activity_metric
501    )