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