Add heading parsing & column indexing

Not being able to use pandas or polars makes things complicated D:

README.md

@@ -28,8 +28,8 @@ Options:
   --help  Show this message and exit.
 
 Commands:
-  audit
-  audit-bulk
+  audit       Audit a consolidated Dropmate log.
+  audit-bulk  Audit a directory of consolidated Dropmate logs.
 ```
 <!-- [[[end]]] -->
 
@@ -44,21 +44,23 @@ The following environment variables are provided to help customize pipeline beha
 ### `dropmate audit`
 Process a consolidated Dropmate log CSV.
 #### Input Parameters
-| 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-firmware` | Threshold firmware version.        | `int\|float` | `5`        |
+| 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-firmware`       | Threshold firmware version.                   | `int\|float` | `5`        |
+| `--time-delta-minutes` | Dropmate internal clock delta from real-time. | `int`        | `60`       |
 
 ### `dropmate audit-bulk`
 Batch process a directory of consolidated Dropmate log CSVs.
 #### Input Parameters
-| Parameter        | Description                                   | Type         | Default    |
-|------------------|-----------------------------------------------|--------------|------------|
-| `--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-firmware` | Threshold firmware version.                   | `int\|float` | `5`        |
+| Parameter              | Description                                   | Type         | Default    |
+|------------------------|-----------------------------------------------|--------------|------------|
+| `--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-firmware`       | Threshold firmware version.                   | `int\|float` | `5`        |
+| `--time-delta-minutes` | Dropmate internal clock delta from real-time. | `int`        | `60`       |
 
 1. Case sensitivity is deferred to the host OS
 2. Recursive globbing requires manual specification (e.g. `**/*.csv`)

dropmate_py/cli.py

@@ -7,6 +7,7 @@
 
 MIN_ALT_LOSS = 200
 MIN_FIRMWARE = 5
+MIN_TIME_DELTA_MINUTES = 60
 
 load_dotenv()
 start_dir = os.environ.get("PROMPT_START_DIR", ".")
@@ -20,6 +21,7 @@ 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),
     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:
@@ -41,6 +43,7 @@ def audit_bulk(
     log_pattern: str = typer.Option("*.csv"),
     min_alt_loss: 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:

dropmate_py/parser.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+import typing as t
+from dataclasses import dataclass, fields
+
+
+@dataclass
+class ColumnIndices:
+    """
+    Column index in the parsed Dropmate compiled log CSV.
+
+    Attribute names are assumed to correspond to the Dropmate log header name. Columns contained may
+    vary by Dropmate app version, so if they are not present in the compiled log they will remain at
+    a value of `-1`, indicating they were not found.
+    """
+
+    serial_number: int = -1
+    uid: int = -1
+    battery: int = -1
+    device_health: int = -1
+    firmware_version: int = -1
+    start_time_utc: int = -1
+    end_time_utc: int = -1
+    start_barometric_altitude_msl_ft: int = -1
+    end_barometric_altitude_msl_ft: int = -1
+    dropmate_internal_time: int = -1
+    last_scanned_time: int = -1
+
+    def __iter__(self) -> t.Generator[str, None, None]:
+        for f in fields(self):
+            yield f.name
+
+    @classmethod
+    def from_header(cls, header: str) -> ColumnIndices:
+        """Attempt to match attribute names to their corresponding data columns."""
+        indices = ColumnIndices()
+
+        # Some column names may have stray spaces in them, listify so we can iterate over repeatedly
+        col_names = [c.strip().lower() for c in header.split(",")]
+        for query_col in indices:
+            for idx, col in enumerate(col_names):
+                if col == query_col:
+                    setattr(indices, query_col, idx)
+
+        return indices
+
+    def __str__(self) -> str:  # pragma: no cover
+        return ", ".join(f"({f}, {getattr(self, f)})" for f in self)

pyproject.toml

@@ -49,7 +49,6 @@ mypy = "^1.0"
 pep8-naming = "^0.13"
 pre-commit = "^3.0"
 pytest = "^7.2"
-pytest-check = "^2.1"
 pytest-cov = "^4.0"
 pytest-randomly = "^3.12"
 tox = "^4.4"

tests/test_heading_parse.py

@@ -0,0 +1,44 @@
+from dropmate_py.parser import ColumnIndices
+
+SAMPLE_FULL_HEADER = "serial_number,uid,battery,device_health, firmware_version,log_timestamp,log_altitude,total_flights,flights_over_18kft,recorded_flights,flight_index,start_time_utc,end_time_utc,start_barometric_altitude_msl_ft,end_barometric_altitude_msl_ft,dropmate_internal_time,last_scanned_time"
+
+
+def test_parse_indices_full_header() -> None:
+    TRUTH_INDICES = ColumnIndices(
+        serial_number=0,
+        uid=1,
+        battery=2,
+        device_health=3,
+        firmware_version=4,
+        start_time_utc=11,
+        end_time_utc=12,
+        start_barometric_altitude_msl_ft=13,
+        end_barometric_altitude_msl_ft=14,
+        dropmate_internal_time=15,
+        last_scanned_time=16,
+    )
+
+    indices = ColumnIndices.from_header(SAMPLE_FULL_HEADER)
+    assert indices == TRUTH_INDICES
+
+
+SAMPLE_OLD_HEADER = "serial_number,uid,battery,log_timestamp,log_altitude,total_flights,prior_flights,flights_over_18kft,recorded_flights,flight_index,start_time_utc,end_time_utc,start_barometric_altitude_msl_ft,end_barometric_altitude_msl_ft"
+
+
+def test_parse_indices_old_header() -> None:
+    TRUTH_INDICES = ColumnIndices(
+        serial_number=0,
+        uid=1,
+        battery=2,
+        device_health=-1,
+        firmware_version=-1,
+        start_time_utc=10,
+        end_time_utc=11,
+        start_barometric_altitude_msl_ft=12,
+        end_barometric_altitude_msl_ft=13,
+        dropmate_internal_time=-1,
+        last_scanned_time=-1,
+    )
+
+    indices = ColumnIndices.from_header(SAMPLE_OLD_HEADER)
+    assert indices == TRUTH_INDICES