Hook up initial audits & CLI functionality

README.md

@@ -1,9 +1,14 @@
 # Dropmate-py
-[![pre-commit](https://img.shields.io/badge/pre--commit-enabled-brightgreen?logo=pre-commit&logoColor=white)](https://github.com/pre-commit/pre-commit)
+[![PyPI - Python Version](https://img.shields.io/pypi/pyversions/dropmate-py/0.1.0?logo=python&logoColor=FFD43B)](https://pypi.org/project/dropmate-py/)
+[![PyPI](https://img.shields.io/pypi/v/dropmate-py?logo=Python&logoColor=FFD43B)](https://pypi.org/project/dropmate-py/)
+[![PyPI - License](https://img.shields.io/pypi/l/dropmate-py?color=magenta)](https://github.com/sco1/dropmate-py/blob/main/LICENSE)
+[![pre-commit.ci status](https://results.pre-commit.ci/badge/github/sco1/dropmate-py/main.svg)](https://results.pre-commit.ci/latest/github/sco1/dropmate-py/main)
 [![Code style: black](https://img.shields.io/badge/code%20style-black-black)](https://github.com/psf/black)
 
 Python helpers for the [EDC Dropmate](https://earthlydynamics.com/dropmate/).
 
+🚨 This is an beta project. User-facing functionality is still under development 🚨
+
 ## Installation
 Install by cloning this repository and installing into a virtual environment:
 
@@ -34,6 +39,7 @@ Commands:
 <!-- [[[end]]] -->
 
 ## Usage
+**NOTE:** All functionality assumes that log records have been provided by Dropmate app version 1.5.16 or newer. Prior versions may not contain all the necessary data columns to conduct the data audit, and there may also be column naming discrepancies between the iOS and Android apps.
 ### Environment Variables
 The following environment variables are provided to help customize pipeline behaviors.
 
@@ -47,7 +53,7 @@ Process a consolidated Dropmate log CSV.
 | Parameter              | Description                                   | Type         | Default    |
 |------------------------|-----------------------------------------------|--------------|------------|
 | `--log-filepath`       | Path to Dropmate log CSV to parse.            | `Path\|None` | GUI Prompt |
-| `--min-alt-loss`       | Threshold altitude delta.                     | `int`        | `200`      |
+| `--min-alt-loss-ft`       | Threshold altitude delta, feet.            | `int`        | `200`      |
 | `--min-firmware`       | Threshold firmware version.                   | `int\|float` | `5`        |
 | `--time-delta-minutes` | Dropmate internal clock delta from real-time. | `int`        | `60`       |
 
@@ -58,7 +64,7 @@ Batch process a directory of consolidated Dropmate log CSVs.
 |------------------------|-----------------------------------------------|--------------|------------|
 | `--log-dir`            | Path to Dropmate log directory to parse.      | `Path\|None` | GUI Prompt |
 | `--log-pattern`        | Dropmate log file glob pattern.<sup>1,2</sup> | `str`        | `"*.csv"`  |
-| `--min-alt-loss`       | Threshold altitude delta.                     | `int`        | `200`      |
+| `--min-alt-loss-ft`       | Threshold altitude delta, feet.            | `int`        | `200`      |
 | `--min-firmware`       | Threshold firmware version.                   | `int\|float` | `5`        |
 | `--time-delta-minutes` | Dropmate internal clock delta from real-time. | `int`        | `60`       |
 

dropmate_py/audits.py

@@ -0,0 +1,65 @@
+from collections import abc
+
+from dropmate_py.parser import Dropmate
+
+
+def _audit_drops(dropmate: Dropmate, min_alt_loss_ft: int) -> list[str]:
+    """Audit for missing drop records and for issues with total altitude loss."""
+    found_issues = []
+
+    if len(dropmate.drops) == 0:
+        found_issues.append(f"UID {dropmate.uid} contains no drop records.")
+
+    for drop_record in dropmate.drops:
+        altitude_loss = (
+            drop_record.start_barometric_altitude_msl_ft
+            - drop_record.end_barometric_altitude_msl_ft
+        )
+        if altitude_loss < min_alt_loss_ft:
+            found_issues.append(
+                f"UID {dropmate.uid} drop #{drop_record.flight_index} below threshold altitude loss: {altitude_loss} feet."  # noqa: E501
+            )
+
+    return found_issues
+
+
+def _audit_dropmate(
+    dropmate: Dropmate, min_firmware: float, max_scanned_time_delta_sec: int
+) -> list[str]:
+    """Audit for issues with firmware version and delta between internal and external clocks."""
+    found_issues = []
+
+    if dropmate.firmware_version < min_firmware:
+        found_issues.append(
+            f"UID {dropmate.uid} firmware below threshold: {dropmate.firmware_version}"
+        )
+
+    internal_timedelta = dropmate.last_scanned_time_utc - dropmate.dropmate_internal_time_utc
+    if abs(internal_timedelta.total_seconds()) > max_scanned_time_delta_sec:
+        found_issues.append(
+            f"UID {dropmate.uid} internal time delta from scanned time exceeds threshold: {internal_timedelta.total_seconds()} seconds"  # noqa: E501
+        )
+
+    return found_issues
+
+
+def audit_pipeline(
+    consolidated_log: abc.Iterable[Dropmate],
+    min_alt_loss_ft: int,
+    min_firmware: float,
+    max_scanned_time_delta_sec: int,
+) -> list[str]:
+    """Run the desired audits over all Dropmate devices and their respective drop records."""
+    found_issues = []
+
+    for dropmate in consolidated_log:
+        found_issues.extend(
+            _audit_dropmate(
+                dropmate,
+                min_firmware=min_firmware,
+                max_scanned_time_delta_sec=max_scanned_time_delta_sec,
+            )
+        )
+        found_issues.extend(_audit_drops(dropmate, min_alt_loss_ft=min_alt_loss_ft))
+
+    return found_issues

dropmate_py/cli.py

@@ -1,11 +1,15 @@
 import os
 from pathlib import Path
 
+import click
 import typer
 from dotenv import load_dotenv
 from sco1_misc.prompts import prompt_for_dir, prompt_for_file
 
-MIN_ALT_LOSS = 200
+from dropmate_py.audits import audit_pipeline
+from dropmate_py.parser import log_parse_pipeline
+
+MIN_ALT_LOSS = 200  # feet
 MIN_FIRMWARE = 5
 MIN_TIME_DELTA_MINUTES = 60
 
@@ -18,40 +22,75 @@
 
 @dropmate_cli.command()
 def audit(
-    log_filepath: Path = typer.Option(exists=True, file_okay=True, dir_okay=False),
-    min_alt_loss: int = typer.Option(default=MIN_ALT_LOSS),
+    log_filepath: Path = typer.Option(None, exists=True, file_okay=True, dir_okay=False),
+    min_alt_loss_ft: int = typer.Option(default=MIN_ALT_LOSS),
     min_firmware: float = typer.Option(default=MIN_FIRMWARE),
     time_delta_minutes: int = typer.Option(default=MIN_TIME_DELTA_MINUTES),
 ) -> None:
     """Audit a consolidated Dropmate log."""
     if log_filepath is None:
-        log_filepath = prompt_for_file(
-            title="Select Flight Log",
-            start_dir=PROMPT_START_DIR,
-            filetypes=[
-                ("Compiled Dropmate Logs", "*.csv"),
-                ("All Files", "*.*"),
-            ],
-        )
+        try:
+            log_filepath = prompt_for_file(
+                title="Select Flight Log",
+                start_dir=PROMPT_START_DIR,
+                filetypes=[
+                    ("Compiled Dropmate Logs", ("*.csv", ".txt")),
+                    ("All Files", "*.*"),
+                ],
+            )
+        except ValueError:
+            raise click.ClickException("No file selected for processing, aborting.")
+
+    conslidated_log = log_parse_pipeline(log_filepath)
+    found_errs = audit_pipeline(
+        consolidated_log=conslidated_log,
+        min_alt_loss_ft=min_alt_loss_ft,
+        min_firmware=min_firmware,
+        max_scanned_time_delta_sec=time_delta_minutes * 60,
+    )
 
-    raise NotImplementedError
+    print(f"Found {len(found_errs)} errors.")
+    if found_errs:
+        for err in found_errs:
+            print(err)
 
 
 @dropmate_cli.command()
 def audit_bulk(
-    log_dir: Path = typer.Option(exists=True, file_okay=False, dir_okay=True),
+    log_dir: Path = typer.Option(None, exists=True, file_okay=False, dir_okay=True),
     log_pattern: str = typer.Option("*.csv"),
-    min_alt_loss: int = typer.Option(default=MIN_ALT_LOSS),
+    min_alt_loss_ft: int = typer.Option(default=MIN_ALT_LOSS),
     min_firmware: float = typer.Option(default=MIN_FIRMWARE),
     time_delta_minutes: int = typer.Option(default=MIN_TIME_DELTA_MINUTES),
 ) -> None:
     """Audit a directory of consolidated Dropmate logs."""
     if log_dir is None:
-        log_dir = prompt_for_dir(
-            title="Select directory for batch processing", start_dir=PROMPT_START_DIR
+        try:
+            log_dir = prompt_for_dir(
+                title="Select directory for batch processing", start_dir=PROMPT_START_DIR
+            )
+        except ValueError:
+            raise click.ClickException("No directory selected for processing, aborting.")
+
+    log_files = list(log_dir.glob(log_pattern))
+    print(f"Found {len(log_files)} log files to process.")
+
+    found_errs = []
+    for log_filepath in log_files:
+        conslidated_log = log_parse_pipeline(log_filepath)
+        found_errs.extend(
+            audit_pipeline(
+                consolidated_log=conslidated_log,
+                min_alt_loss_ft=min_alt_loss_ft,
+                min_firmware=min_firmware,
+                max_scanned_time_delta_sec=time_delta_minutes * 60,
+            )
         )
 
-    raise NotImplementedError
+    print(f"Found {len(found_errs)} errors.")
+    if found_errs:
+        for err in found_errs:
+            print(err)
 
 
 if __name__ == "__main__":

dropmate_py/parser.py

@@ -108,6 +108,7 @@ class DropRecord:
     battery: Health
     device_health: Health
     firmware_version: float
+    flight_index: int
     start_time_utc: dt.datetime
     end_time_utc: dt.datetime
     start_barometric_altitude_msl_ft: int
@@ -140,6 +141,7 @@ def from_raw(cls, log_line: str, indices: ColumnIndices) -> DropRecord:
             battery=Health(df["battery"].lower()),
             device_health=Health(df["device_health"].lower()),
             firmware_version=float(df["firmware_version"]),
+            flight_index=int(df["flight_index"]),
             start_time_utc=dt.datetime.fromisoformat(df["start_time_utc"]),
             end_time_utc=dt.datetime.fromisoformat(df["end_time_utc"]),
             start_barometric_altitude_msl_ft=int(df["start_barometric_altitude_msl_ft"]),
@@ -154,6 +156,7 @@ class Dropmate:  # noqa: D101
     uid: str
     drops: list[DropRecord]
     firmware_version: float
+    dropmate_internal_time_utc: dt.datetime
     last_scanned_time_utc: dt.datetime
 
     def __len__(self) -> int:  # pragma: no cover
@@ -175,6 +178,7 @@ def _group_by_uid(drop_logs: list[DropRecord]) -> list[Dropmate]:
                 # It should be a safe assumption that these values are consistent across logs from
                 # the same device
                 firmware_version=logs[0].firmware_version,
+                dropmate_internal_time_utc=logs[0].dropmate_internal_time_utc,
                 last_scanned_time_utc=logs[0].last_scanned_time_utc,
             )
         )

tests/test_log_objects.py

@@ -14,6 +14,7 @@
     battery=parser.Health.GOOD,
     device_health=parser.Health.GOOD,
     firmware_version=5.1,
+    flight_index=1,
     end_time_utc=dt.datetime(
         year=2023, month=4, day=20, hour=11, minute=30, second=0, tzinfo=dt.timezone.utc
     ),