# Data Pipeline ```{note} This page documents the pipeline developed within the scope of the initial MVP. ``` ## Executive Summary ```{tip} :class: margin A data pipeline is an automated system that takes raw data, transforms it in some way, then moves it to a destination storage area. ``` ```{epigraph} The purpose of the Smart Health MVP data pipeline is to ingest a daily, compressed archive of `.json` data from Daiser, then clean, enrich, and pseudonymise the data. The processed data will form part of a "research-ready" database in S3 object storage. ``` Data will be ingressed to our RONIN Isolate environment on a daily basis from Daiser via Secure File Transfer Protocol (SFTP). This compressed archive will be stored in an S3 "Ingress" bucket. Each compressed archive will contain the previous 7 days' participant data for both GPS data and Fitbit data. The Fitbit data will be comprised of daily summary data, and intraday (many samples within one day) activity data. The ingressed data will be processed in two stages: In __Stage 1: Ingress to Intermediate__, the raw Device and GPS data is parsed, validated, and prepared for the Stage 2. The processed data from this stage is saved to the __"Intermediate"__ S3 bucket, and the raw data is re-compressed into separate archives for Device and GPS data, and then saved to the __"Raw"__ S3 bucket. __Note__: For data governance purposes, the highly-sensitive GPS data is processed on a separate machine to the other data. This processing includes deriving measures from the raw data before converting it to pseudonymised LSOA data. Further details about this processing are described below. In __Stage 2: Intermediate to Clean__, the pseudonymised LSOA data can be processed alongside other data steams. In this stage, the LSOA data is enriched with AHAH data. All processed data streams are then merged into the clean database, stored in the __"Clean"__ S3 bucket. This process is outlined in the diagram below. ```{figure} ../_static/pipeline_full.webp :width: 900px :alt: Fitbit and GPS data is ingressed from Daiser to an "Ingress" S3 bucket in a RONIN Isolate environment. GPS data is processed on a dedicated, secure machine to pseudonymise to LSOA value and derive mobility measures. Processed Fitbit and GPS (LSOA) data is moved to an "Intermediate" S3 bucket, before a final data cleaning and enrichment step. The fully-cleaned data is stored in a "Clean" S3 bucket. The raw data is recompressed and stored in a "Raw" S3 bucket A diagram of the Smart Health MVP pipeline. ``` --- ## Technical Specification The Smart Health pipeline is a __multi-stage, batch, ETL pipeline with medallion lakehouse architecture__. As data moves through the pipeline its quality is progressivley improved, from the "Raw" bucket (Bronze layer), to the validated, pseudonymised "Intermediate" bucket (Silver layer), through to the enriched, research-ready "Clean" bucket (Gold layer). ### Data Source __Ingestion method:__ SFTP into S3 object storage bucket in RONIN Isolate __Format (variety):__ `.json` within compressed `.7z` archive __Frequency (velocity):__ Daily. Each compressed `.7z` archive will contain the previous 7-days of data. __Volume:__ up to 10,000 participants, one `.json` file per participant, per day, per stream i.e. GPS, Fitbit Daily Summary, Fitbit Intraday activities. __Data Loading Strategy:__ Currently Full Load with a Change Data Capture (CDC) mechanism. The full archive of 7 previous days is processed and merged into the database. In future stages, this will move to an Incremental (delta) Load, where only data within the archive that has actually been changed will be synchronised. More details can be found in [Data Versioning](#data-versioning). ### Infrastructure and Configuration __Processing framework:__ Batch processing using Python scripts managed by a workflow manager. All scripts are available under an MIT licence in the [source repository](https://github.com/Data-Connect-UoS/BHF-SmartHealth). __Environment:__ RONIN Isolate AWS Secure Data Environment __Dependency Management:__ All Python dependencies are managed using the open-source UV package manager[^sn1] for fast, reliable, and reproducible virtual environment builds. [^sn1]: UV is an extremely fast Python package and project manager written in Rust. See [here](https://docs.astral.sh/uv/) for more information. __Resource requirements:__ Tested using EC2 VMs with 16Gb RAM, 4 vCPUs, and 50Gb SSD operating on Ubuntu 22.04. Note that a larger SSD, e.g. 100Gb, would provide a more sufficient margin for the expected data volumes. __Version requirements:__ Requires Python >=3.12 __Configuration:__ Database and Workflow manager configured seperately, see [Customising the Pipeline](#customising-the-pipeline). ### Transformation Logic __Validation:__ Validation of the raw `.json` data is applied using the `pydantic` Python library. For each raw data stream, a list of expected data fields and their expected data types are defined. Where possible, `pydantic` will coerce the data into the specified data type. Files containing unexpected fields are rejected and skipped for Information Governance purposes. __GPS data:__ - GPS data files will contain an array of intraday data points. Expects datetime, latitude, and longitude fields for each GPS data point in the following format: ```{code-block} python d: datetime long: float lat: float ``` - Unfeasible latitude / longitude values (outside of -90 to 90, and -180 to 180, respectively) are filtered from the data at this stage __Fitbit data:__ - Daily Summary data is expected to contain a number of fields in the following format: ```{code-block} python timestamp: date calories: int caloriesBMR: int distance: int elevation: int floors: int minutesSedentary: int minutesLightlyActive: int minutesFairlyActive: int minutesVeryActive: int steps: int breathingRate: int vo2Max: int dailyRmssd: int deepRmssd: int sleep_deep_minutes: int sleep_light_minutes: int sleep_rem_minutes: int sleep_wake_minutes: int sleep_minutesToFallAsleep: int sleep_timeInBed: int SP02_avg: int SP02_min: int SP02_max: int tempCore: int ``` - Intraday Activities data files will contain an array of intraday data points in the following format: ```{code-block} python d: datetime v: int ``` `d` is mapped to the field name `timestamp`. `v` is mapped to the field name `value`. __Cleaning:__ * Unrealistic speed of travel values (derived from GPS data) are filtered. By default this is any speed greater than 36 metres per second. * Where there is a large time jump between GPS data points (by default greater than 5 minutes, but configurable in `config.py`), the distance travelled and speed of travel will not be calculated. * The LSOA value at these invalid data points is also filtered __Pseudonymisation:__ * GPS data is highly sensitive information, and will therefore be pseudonymised by conversion to LSOA values. __Aggregation:__ * Raw intraday data points are aggregated to balance data size/ storage considerations with the granularity required for research purposes. By default, intraday measures (LSOA, mobility, Fitbit activites) are aggregated into 5-minute time periods. * The Environmental data is aggregated to 1-hour time periods. * Some Intraday Fitbit activity data is aggregated over a 24 hour period to produce additional daily summary metrics, such as the average (mean) heart rate per day. The specific (default) aggregate calculations used to produce each table, and any filters applied during calculation, are as follows: _Intraday LSOA:_ - Aggregated into 5-minute periods. The 'latest' LSOA value is taken within each 5-minute period, defined as the most-recent LSOA at a valid data point (i.e. note after a large time jump.) _Intraday Mobility:_ - Aggregated into 5-minute periods, including: * The average (mean) speed of travel within each time period in metres per second * The total (sum) distance travelled within each time period in metres * The number (count) of samples in each time period * The number (count) of unique LSOAs in each time period _Intraday Environmental:_ - Aggregated into 1-hour periods. These aggregates are calculated from the aggregated Intraday LSOA data points. The modal LSOA is calculated for each time period. Where a tie-break exists, the latest (most recent) LSOA will be used. _Fitbit Intraday Activities:_ - Aggregated into 5-minute periods __and__ 24 hour period: * Heart rate: * Average (mean) heart rate for each 5-minute and 24 hour period * Minimum heart rate for each 5-minute and 24 hour period * Maximum heart rate for each 5-minute and 24 hour period * The number (count) of samples in each 5-minute time period _Note that due to expected artefacts in device-measured heart rate data, a configurable filter can be used to perform aggregate calculations on a filtered subset of the heart rate data. Data is not explicitly removed from the database such that researchers can filter data at their discretion. Note that currently the only supported filtered aggregate calculations are `size`, `mean`, `max`, and `min`._ _The filter (configured in `config.py`) by default filters heart rate values below 0 or above 320, i.e. incompatible with life._ _For completion, the above list of aggregated measures (average, min, mean, and number of samples) is also reported for the filtered heart rate data._ * Steps: * The total number of steps taken for each 5-minute and 24 hour period * The number (count) of samples in each 5-minute time period _Fitbit Daily Summary:_ Naturally a single, daily metric. No aggregation is applied. __Enrichment:__ * All tables except Fitbit Daily Summary will have a `date`, `time`, and `datetime` field * AHAH data used to derive Environmental data table from LSOA values * A `sheffield_id`, unique to each participant, is attached to each record ### Destination __Target system:__ AWS S3 data lakehouse in a RONIN Isolate Secure Data Environment __Output tables:__ * Intraday LSOA * Intraday Environmental * Intraday Mobility * Fitbit Daily Summary * Fitbit Intraday Activities __File Format:__ Apache Parquet __Update strategy:__ Upsert/merge __Schema enforcement:__ Flexible, as parquet file contains its own schema metadata __Partioning:__ Tables will be partitioned based on time (year, month, and day) __Structure:__ `s3://////clean__YYYY-MM-DD.parquet`. See [Directory Structure](../database_design/database_design.md#directory-structure). ### Orchestration [^sn2]: [snakemake.github.io](snakemake.github.io) __Workflow management:__ The pipeline's processing framework is managed and controlled by Snakemake[^sn2], a specialized workflow management system based on Python. Snakemake defines the pipeline's entire control flow as a Directed Acyclic Graph (DAG), where individual transformation steps are represented as rules. This ensures that the complex multi-stage process (`Ingress` --> `Intermediate` --> `Clean`) is executed reliably and reproducibly. Execution of the Snakemake workflows will be automated using `cron` scheduled jobs from a `shell` script. __Retry policy:__ The Snakemake workflow can be configured to retry failing jobs. ### Monitoring, Logging, and Testing: __Error handling:__ * Where dates specified for processing are not available in the daily, compressed archive, Snakemake will log a warning and continue with the available dates. * Where files contain additional fields to those expected to be ingressed, a file will be rejected on Data Governance grounds. * Files that fail data validation / quality checks will be skipped. __Logging:__ The Smart Health pipeline extends Snakemake's native logging to provide detailed and persistent logs for each workflow execution. Logging is managed using Python's native `logging` module. The path for the log files can be configured separately to the path for data outputs. This facilitates the use of storage-optimised, ephemeral drives to fast I/O data operations, whilst ensuring logs persist in the case of machine failure. Each log filename contains a timestamp to prevent overwriting. Two types of log are output for each workflow execution: * A global Snakemake workflow log. This file contains the date that the workflow is executed on. * Rule-specific logs for each rule that executes a Python script. The logging level can be configured from the `config.py` file. This file is named with the date of the processed day within the compressed archive. Log files are stored under a top-level `logs/` directory. Under this directory is a folder labelled with the date of the daily archive being processed, with the name of the workflow appended. When the same date is processed in multiple daily archives, the logs for each will be in separate subdirectories (labelled after the archive date). e.g. for the archive dated 2025-11-11, processing day -1 and -2, the logs from the GPS workflow run (with one script running rule, `process_GPS`) would output the following: ```{code-block} shell ├── logs │ └── 2025-11-11_GPS │ ├── 2025-11-09_08:34:20_process_GPS.log │ ├── 2025-11-10_08:34:20_process_GPS.log │ └── 2025-12-09_08:34:25_snakemake_log.log ``` [^sn3]: [Logging documentation](https://docs.python.org/3/library/logging.html#formatter-objects) By default, the format of the log files is `"%(asctime)s [%(levelname)s] | %(name)s: %(message)s"`[^sn3]. This format outputs the logs as (if the logging level is configured to `INFO`): ```{code-block} shell 'YYYY-MM-DD HH:MM:SS,ms [INFO] | bhf_smarthealth..: ' ``` __Unit testing:__ Each unit function of the pipeline codebase is unit tested using the Python testing framework `pytest`. __End-to-end testing:__ The Snakemake-orchestrated pipeline has been end-to-end tested for 7 days of 10,000 participants' data to mimic the maximum expected daily, compressed archive size. --- ## Snakemake Pipeline Workflow Implementation ```{epigraph} A Snakemake workflow is defined by specifying rules in a Snakefile. Rules decompose the workflow into small steps (for example, the application of a single tool) by specifying how to create sets of output files from sets of input files. Snakemake automatically determines the dependencies between the rules by matching file names. -- [Snakemake documentation](https://snakemake.readthedocs.io/en/stable/tutorial/basics.html) ``` By default, the first rule in a Snakefile is used to define build targets. In the Smart Health Snakefiles, this is named `rule all`. Snakemake automatically builds a Directed Acyclic Graph (DAG) of the series of rules in the Snakefile to determine the execution order that created the build targets. Each Snakemake rule essentially says "To create `[OUTPUTS]`, I need these `[INPUTS]`, and I will execute `[ACTIONS]` to do it." The actions can be command-line code, or custom Python scripts. This section details the rules / steps that Snakemake manages for both stages of the Smart Health pipeline. #### Data Versioning The compressed daily archive recieved from Daiser will contain the 7 previous days' worth of data. The reason for including the previous 7 days is to capture any data that may have been delayed, corrected, or updated since the last data file. This would be as a result of a participant syncing their wearable device beyond the date that it was collected. This means that a participants' data for a particular day will be "seen" up to 7 times by the Smart Health pipeline. There are two main approaches to re-processing this sliding window of data: __1. Full Load (a.k.a. process everything)__ This is the current Data Loading strategy implemented. All records within the archive are processed by the Smart Health pipeline, irrespective of whether the file contains differences to previous versions. Any changes are upserted into the "Clean" tables. This is simple to implement, but has high resource usage. This has been minimised by processing a file for only __3 timepoints in the 7-day sliding window__. This is demonstrated in the diagram below. ```{figure} ../_static/versioning.webp :align: centre :width: 900px :alt: Data associated with Day 8, for example, would be received 7 times from Day 9 onwards. Instead of processing Day 8's data 7 times, it will be processed 3 times; on the first, fourth, and seventh day of receipt A diagram showing that for each day/date of data collected (columns), the data is received from Daiser multiple times (rows). Each day's data is received 7 times, but processed 3 times. This balances resource costs (for processing) with processing the data suffienctly early in the 7-day window to identify, and rectify, any data quality issues. ``` To implement the Full Loading strategy, the `workflow/config/config.yaml` file contains a `XX_DAY_DELTAS` key, where `XX` is the workflow being executed (either `GPS` or `Device`). This list of integers indicates which days prior to the receipt date of the archive that should be processed. See [Customising the pipeline](#customising-the-pipeline) for more information. This list of deltas is used to dynamically generate the output files expected by the `rule all` in each Stage 1 workflow. At this initial workflow stage, a Snakemake `checkpoint` is run to ensure that the defined deltas are present in the daily archive file. A message is logged to warn if any are not present, and to list the dates that will be processed. __2. Incremental (delta) Load (a.k.a. process new data only)__ In a future development phase, the Smart Health will instead implement an incremental load strategy. The pipeline will implement a filtering strategy that will check each file for changes compared to its previous versions. This will be achieved by comparing the file `diff`, or by generating a hashed string from each file to check its identity. Using this approach, only files that have actually changed would be processed by the pipeline, with the changes upserted to the "Clean" tables. ### Stage 1: Ingress to Intermediate ```{admonition} Click here to see the Stage 1 pipeline diagram :class: dropdown ```{figure} ../_static/pipeline_stage1.webp :width: 900px :alt: Stage 1 of the pipeline shows the ingress of data from Daiser to the "Ingress" bucket, followed by GPS and Device data processing to the "Intermediate" bucket stage. In this stage, the raw data is also re-compressed and stored in the "Raw" bucket Stage 1 of the Smart Health pipeline ``` ```{figure} ../_static/stage1.webp :align: centre :width: 900px :alt: For GPS processing in Stage 1, 4 Snakemake rules move and decompress the raw data onto a dedicated GPS-processing virtual machine, process the data, move the processed data to the intermediate bucket, and recompress the raw data for storage. For Device processing in Stage 1, 6 Snakemake rules exist. Again the raw data is moved and decompressed onto a virtual machine, the raw data is recompressed for storage, and the final processed data is moved to the intermediate bucket. The Daily Summary and Intraday data is processed by one rule each. The final rule combines the Daily Summary data with summary metrics derived from the intraday data, before the processed data is moved. A diagram outlining the Snakemake rules, input and output files, and data flow of Stage 1 of the Smart Health pipeline ``` The GPS and Device data processing workflows are seperated into two Snakefiles. This both satisfies Separation of Concerns, and facilities the isolated (and parallel) processing of GPS data on a separate machine. #### GPS processing The GPS Snakefile runs the following steps: 1. Download and extract GPS data from the compressed daily archive in the __"Ingress"__ bucket using a `shell` command 2. Reference LSOA data is downloaded from S3 storage using a `shell` command. This step only runs if the LSOA data is not already available in the output location 3. Process the downloaded GPS data by running custom Python scripts 4. Upload the processed GPS `.parquet` files to the __"Intermediate"__ bucket using a `shell` command 5. Recompress the daily archive data and upload it to the __"Raw"__ bucket using a `shell` command Within step 3, a workflow Python script (`workflow/scripts/GPS_workflow.py`) is executed to process the decompressed raw `.json` data. This script in turn calls a GPS processing script from the Smart Health `process` sub-module. This Python script iterates over each individual file and processes it. This process involves: 1. Loading the data 2. Validating the data using `pydantic` 3. Converting to a `pandas` dataframe 4. Adding geometry information to facilitate distance calculations 5. Calculation of distance and speed measures between data points 6. Data cleaning of unfeasible distances or speeds. These filters can be configured. 7. Joining the reference LSOAs to the GPS locations 8. Removing columns that should not be included in the output tables (e.g. non-pseudonymised GPS data) 9. The LSOA is converted to `NULL` at invalid time points. This is specified as after a large time jump (defaulting to 5 minutes). This prevents the inclusion of these LSOA values in aggregate calculations. 10. Aggregation of the data at the configured time resolution (default 5 minutes) and calculation of the specified metrics 11. Finally, columns are renamed to match expected outputs All of the processed data per day is then saved into one storage-efficient binary format (as Apache Parquet files). The pipeline logs the number of files successfully processed (and whether this differs from the total number of available files). In essence, the Snakefile takes as input the GPS data `.json` files from the compressed daily archive, for the configured day deltas, and returns as output one `.parquet` file per day for each of the pseudonymised LSOA and derived mobility data. #### Device Processing The Device Snakefile runs the following steps: 1. Download and extract Device data from the compressed daily archive in the __"Ingress"__ bucket using a `shell` command 2. Process the downloaded Device Daily Summary data by running custom Python scripts 3. Process the downloaded Device Intraday Activities data by running custom Python scripts 4. Combine the Daily Summary data with derived daily summary metrics from the Intraday Activities files by running custom Python scripts 5. Upload the processed Device `.parquet` files to the __"Intermediate"__ bucket using a `shell` command 6. Recompress the daily archive data and upload it to the __"Raw"__ bucket using a `shell` command Within steps 2,3, and 4, a workflow Python script (`workflow/scripts/XX_workflow.py`) is executed to process the decompressed raw `.json` data. Here, `XX` is either `fitbit_intraday_`, `fitbit_summary`, or `combine_summaries`. The `fitbit_intraday_` and `fitbit_summary` scripts in turn call processing scripts from the Smart Health `process` sub-module. These Python script iterate over each individual file and processes it by: 1. Loading the data 2. Validating the data using `pydantic` 3. Creating a `pandas` dataframe For the intraday data processing, an additional step exists to: 4. Aggregate the data into both: * Intraday measures (default 5 minute intervals) * A daily summary All of the processed data per day is then saved into one storage-efficient binary format (as Apache Parquet files). At this stage, the Daily Summary data is an 'Interim' file. The pipeline logs the number of files successfully processed (and whether this differs from the total number of available files). The `combine_summaries` workflow then joins the daily aggregated metrics from the raw intraday data into the Interim Daily Summary dataframe, before saving the combined output into a final Daily Summary `.parquet` file. The Snakefile takes as input the Device data Daily Summary and Intraday Activities `.json` files from the compressed daily archive, for the configured day deltas, and returns as output one `.parquet` file per day for the Daily Summary data and one file for each of the Intraday Activities. ### Stage 2: Intermediate to Clean ```{admonition} Click here to see the Stage 2 pipeline diagram :class: dropdown ```{figure} ../_static/pipeline_stage2.webp :width: 900px :alt: Stage 2 encompasses the semi-processed data from the "Intermediate" bucket being cleaned and synchronised with the final database in the "Clean" bucket Stage 2 of the Smart Health pipeline ``` ```{figure} ../_static/stage2.webp :align: centre :width: 900px :alt: The Snakemake rules in Stage 2 move the Intermediate data files onto a virtual machine, before combining the intraday data streams to one file, and enriching the pseudonymised LSOA data with environmental measures. In a final cleaning step, the field names are prepared for synchronisation with the database. The relevant partitioned data files are moved from the "Clean" bucket onto the same machine, the data is synchronised, and the updated clean files are re-uploaded to the "Clean" bucket A diagram outlining the Snakemake rules, input and output files, and data flow of Stage 2 of the Smart Health pipeline ``` As it is pseudonymised data, the LSOA data can now be processed on the same machine as other data streams. Therefore, one Snakefile co-ordinates Stage 2 of the Smart Health pipeline. The `clean` Snakemake worklow broadly covers two operations; enriching the intermediate data, then merging it into the existing database. #### Data Enrichment In the data enrichment stage: * Data stored in the "Intermediate" bucket is first downloaded * If not already available, the referance AHAH data is downloaded * The AHAH data is combined with the `lsoa` table to generate the `environmental` table * Each Fitbit intraday activity data table is combined into one `fitbit_intraday` table The resulting `.parquet` files are stored as `temp` Snakemake files ready for merging into the Smart Health database. #### Data Synchronisation Before database synchronisation, there is first a Snakemake `checkpoint` step to check whether the data is being inserted into the database for the first time, or merged with existing data: * In the S3 datalake, the Smart Health database tables are stored in daily partitions for each table. The Snakemake `clean` workflow looks for a daily file corresponding to the date being processed. This file may already be present if the date has been processed before (due to receiving a given day's data for 7 days), or it will be missing if this is the first day of receipt of valid data for this date. A temporary file is generated to indicate the outocme of searching for the given file; if the corresponding file is found, the file will contain `FOUND`, else it will contain `MISSING`. In the next step, the contents of this temporary file are inspected. If the file is found, file download is triggered. If not, an empty list is returned. If the file state cannot be determined, an error is raised. A log file is generated to indicate the successful completion of this step. Next, either the existing file or empty list is used as an input for the 'upsert' step. The data to be added to the database is first enriched with additional `date` and `time` columns generated from the `timestamp` column. * If no existing file is present, this enriched data file is simply copied to the S3 bucket * If an existing file is present, the `execute_upsert` function is run. This function upserts the daily data into an in-memory DuckDB database created from the downloaded, pre-exisitng data file. The number of rows modified is logged. The resulting, modified `.parquet` file is re-uploaded to the "Clean" S3 bucket. A 'receipt' is generated to indicate the step has proceeded successfully both locally (under `/logs`) and in the S3 bucket next to the daily data file. The local receipt flags to Snakemake that the workflow has run successfully for the given data table. A third, timestamped log file is saved to the local log path directory containing the logs from the upsert stage. The updated Smart Health database is now read for querying! --- ## Customising the Pipeline Please see [Getting Started](../getting_started/getting_started.md#getting-started) for a quickstart guide to running the Smart Health pipeline. #### Snakemake configuration Snakemake can be configured using the keys available in `workflow/config/config.yaml`. Snakemake can also be configured further by using command line arguments. For a full list of available options, see the [Snakemake documentation](https://snakemake.readthedocs.io/en/stable/executing/cli.html#all-options) ##### Day Deltas Two settings are available: * `GPS_DAY_DELTAS` * `DEVICE_DAY_DELTAS` Both keys expect a list of integers which indicate the days prior to when the archive was recieved that should be processed. Note that this should always be a value greater than 1. Values less than 1, or repeated values, will be ignored. ##### Output paths `LOCAL_OUTPUT_PATH` defines the local machine path for storing intermediate files. `LOCAL_LOG_PATH` defines the local machine path for storing log files. A subdirectory called `logs/` will be created unter this directory. #### Database configuration The Smart Health database can be configured using `src/bhf_smarthealth/conf.py`. ##### Logging level The logging configuration can be set using `DEFAULT_LOG_LEVEL`, set to `INFO` by default. See [logging levels](https://docs.python.org/3/library/logging.html#logging-levels) for available levels. ##### Intraday Aggregation Period The time resolution that intraday data (activities, LSOA, and mobility) is aggregated to is specified by `DEFAULT_TIME_PERIOD`, which defaults to 5 minutes. ##### Data Output Settings A number of settings exist to customise the data ouputs of the Smart Health pipeline. This includes the aggregation measures (e.g. mean, median, min, max etc.), which columns are aggregated (this can be more than one aggregate measure), and the name of the columns in the output files. ###### GPS data specific settings `DEFAULT_GPS_FILE_NAME`: specifies the file name of the raw `.json` GPS data file to expect. Defaults to `gps_data`. `DEFAULT_GPS_CRS`: The co-ordinate reference system (CRS) to map GPS latitude and longitude values to. Defaults to "EPSG:4326". `DEFAULT_PLANAR_CRS`: The planar CRS to map GPS datapoints to. Necessary for distance calculations. This defaults to the British National Grid. `DEFAULT_DECIMAL_PLACES`: The default decimal places to round to. Defaults to 3. `DEFAULT_MAX_SPEED_MS`: The maximum allowed speed value, in metres per second. Defaults to 36 m/s, or ~130 kph. Values above this value are filtered from the data set. `DEFAULT_MAX_TIME_JUMP_MINS`: The maximum allowed time jump, in minutes, between data points. Defaults to 5 minutes. If a time jump between data points is greater than this value, the distance and speed of travel between the adjacent points is omitted from the dataset. The LSOA values corresponding to these data points will also not be included in aggregate calculations. `DEFAULT_REMOVE_GPS_COLUMNS`: The default columns to remove from the processed GPS `pandas` dataframe. Defaults to `["lat", "long", "geometry", "index_right"]`. The first three must be removed to remove the highly-sensitive GPS data, whilst the `index_right` is a superfluous field generated during data joining. `DEFAULT_GPS_AGG_DICT`: The Python dictionary of columns to aggregate, and the calculations to perform for each column. Defaults to ```{code-block} python { "distance": ["size", "sum"], "speed": ["mean"], "LSOA21NM": ["last"], "LSOA21CD": ["last", "nunique"], } ``` `DEFAULT_RENAME_GPS_COLUMNS`: Defines the column names in the output tables. Defaults to ```{code-block} python [ "mobility_n_samples", "dist_travelled", "mean_speed", "latest_LSOA21NM", "latest_LSOA21CD", "n_unique_LSOAs", ] ``` `DEFAULT_GPS_OUTPUT_COLUMNS`: Defines the output files and their columns. Defaults to ```{code-block} python { "LSOA_out": ["ID", "latest_LSOA21CD", "latest_LSOA21NM"], "mobility_out": ["ID", "mobility_n_samples", "dist_travelled", "mean_speed", "n_unique_LSOAs"], } ``` ###### Device data specific settings `DEFAULT_FB_INTRADAY_AGG_DICT`: The Python dictionary of columns to aggregate, and the calculations to perform for each column. Defaults to ```{code-block} python { "heart": {"value": ["mean", "min", "max", "size"]}, "steps": {"value": ["sum", "size"]}, } ``` `DEFAULT_FB_INTRADAY_AGG_SETTINGS`: The time periods to aggregate the Fitbit intraday activities to. Facilitates the generation of additional Daily Summary data derived from the intraday raw data. Defaults to: ```{code-block} python { "intraday": {"time_period": DEFAULT_TIME_PERIOD}, "summary": {"time_period": "1d", "label": "left"}, } ``` ###### Environmental data specific settings `DEFAULT_AHAH_TIME_PERIOD`: The time period to aggregate LSOA data into to join with environmental measures. Defaults to 1 hour. `DEFAULT_AHAH_AGG_DICT`: The Python dictionary of columns to aggregate, and the calculations to perform for each column. Defaults to ```{code-block} python { "LSOA_n_samples": ["sum"], "latest_LSOA21CD": [aggregators.mode_with_later_tiebreak], "latest_LSOA21NM": [aggregators.mode_with_later_tiebreak], } ``` `DEFAULT_RENAME_AHAH_COLUMNS`: Defines the column names in the output tables. ```{code-block} python [ "env_n_samples", "modal_LSOA21CD", "modal_LSOA21NM", ] ``` `DEFAULT_AHAH_MERGE_COL`: The column used to merge reference AHAH data with the aggregated LSOA data. Defaults to ["modal_LSOA21CD"]. `DEFAULT_REMOVE_AHAH_COLUMNS`: The default columns to remove from the Environmental data table. Defaults to `["LSOA21CD", "modal_LSOA21CD", "modal_LSOA21NM"]`. `DEFAULT_AHAH_FIELDS`: Defines the AHAH data fields available in the output Environmental data table.