feat: add copy_temporal_columns function to preserve acquisition timestamps and update relevant parsers

jirhiker · jirhiker · commit 653c9bbf9e6b · 2026-04-21T15:22:12.000-06:00
diff --git a/services/aem_parsers/agf.py b/services/aem_parsers/agf.py
@@ -23,7 +23,7 @@
 
 import pandas as pd
 
-from services.aem_parsers.common import finalize_parsed_dataframe
+from services.aem_parsers.common import copy_temporal_columns, finalize_parsed_dataframe
 
 logger = logging.getLogger(__name__)
 
@@ -78,6 +78,7 @@ def parse_agf_lci(filepath: str, system: str) -> pd.DataFrame:
                 raise ValueError(f"Expected column '{col}' not found in AGF file.")
 
         layer_df = df[["record_id", "Line", "E_UTM13Nm", "N_UTM13Nm", "DEM_m"]].copy()
+        layer_df = copy_temporal_columns(df, layer_df)
 
         # Sounding-level columns (constant across layers)
         if "ALTITUDE__m_" in df.columns:
diff --git a/services/aem_parsers/common.py b/services/aem_parsers/common.py
@@ -20,6 +20,21 @@
 # to 26913 in-memory before deriving the stored WGS84 geometry.
 TARGET_EPSG = 26913
 
+TEMPORAL_DATETIME_COLUMNS = [
+    "acquisition_datetime",
+    "datetime_acquired",
+    "acquired_at",
+    "timestamp",
+    "datetime_utc",
+    "utc_datetime",
+]
+TEMPORAL_TIME_COLUMNS = [
+    "acquisition_time",
+    "time_acquired",
+    "time_utc",
+    "utc_time",
+]
+
 # Canonical column names for the parser/Parquet contract.
 # The database only persists the non-coordinate fields plus a WGS84 geometry.
 CANONICAL_COLUMNS = [
@@ -66,6 +81,17 @@ def ensure_canonical_columns(df: pd.DataFrame) -> pd.DataFrame:
     return df
 
 
+def copy_temporal_columns(
+    source_df: pd.DataFrame,
+    target_df: pd.DataFrame,
+) -> pd.DataFrame:
+    """Preserve likely acquisition timestamp columns through parser reshaping."""
+    for col in TEMPORAL_DATETIME_COLUMNS + TEMPORAL_TIME_COLUMNS:
+        if col in source_df.columns and col not in target_df.columns:
+            target_df[col] = source_df[col]
+    return target_df
+
+
 def finalize_parsed_dataframe(
     df: pd.DataFrame,
     source_label: str,
diff --git a/services/aem_parsers/seogi.py b/services/aem_parsers/seogi.py
@@ -25,7 +25,7 @@
 
 import pandas as pd
 
-from services.aem_parsers.common import finalize_parsed_dataframe
+from services.aem_parsers.common import copy_temporal_columns, finalize_parsed_dataframe
 from services.aem_parsers.detect import extract_flight_id
 
 logger = logging.getLogger(__name__)
@@ -89,6 +89,7 @@ def parse_seogi_rho(filepath: str, flight_id: Optional[str] = None) -> pd.DataFr
                 )
 
         layer_df = df[["record_id", "line_no", "utmx", "utmy", "elevation"]].copy()
+        layer_df = copy_temporal_columns(df, layer_df)
 
         # Include plm if present (some Seogi outputs include it)
         if "plm" in df.columns:
diff --git a/services/aem_stac.py b/services/aem_stac.py
@@ -9,12 +9,16 @@
 import pandas as pd
 import re
 from collections.abc import Callable, Iterable
-from datetime import date, datetime, timezone
+from datetime import date, datetime, time, timezone
 from google.cloud import storage
 from schemas.aem import SURVEY_METADATA, IngestConfig
+from services.aem_parsers.common import (
+    TEMPORAL_DATETIME_COLUMNS,
+    TEMPORAL_TIME_COLUMNS,
+)
 from services.util import transform_srid
 from shapely.geometry import box, mapping
-from urllib.parse import quote_plus, urlencode
+from urllib.parse import quote, quote_plus, urlencode
 
 logger = logging.getLogger(__name__)
 
@@ -42,8 +46,75 @@ def _stac_datetime_or_none(value) -> str | None:
     return f"{text_value}Z"
 
 
+def _combine_date_and_time(date_value, time_value):
+    if date_value is None or pd.isna(date_value):
+        return None
+    if time_value is None or pd.isna(time_value):
+        return date_value
+
+    if isinstance(date_value, pd.Timestamp):
+        date_value = date_value.to_pydatetime()
+    if isinstance(time_value, pd.Timestamp):
+        time_value = time_value.to_pydatetime()
+
+    if isinstance(date_value, datetime):
+        if date_value.time() != datetime.min.time():
+            return date_value
+        date_part = date_value.date()
+    elif isinstance(date_value, date):
+        date_part = date_value
+    else:
+        parsed_date = pd.to_datetime(date_value, errors="coerce")
+        if pd.isna(parsed_date):
+            return date_value
+        date_part = parsed_date.date()
+
+    if isinstance(time_value, datetime):
+        time_part = time_value.timetz()
+    elif isinstance(time_value, time):
+        time_part = time_value
+    else:
+        parsed_time = pd.to_datetime(time_value, errors="coerce")
+        if pd.isna(parsed_time):
+            return date_value
+        time_part = parsed_time.timetz()
+
+    return datetime.combine(date_part, time_part)
+
+
+def _stac_datetimes_from_frame(df: pd.DataFrame) -> list[str]:
+    for col in TEMPORAL_DATETIME_COLUMNS:
+        if col not in df.columns:
+            continue
+        values = [_stac_datetime_or_none(value) for value in df[col]]
+        cleaned = sorted({value for value in values if value is not None})
+        if cleaned:
+            return cleaned
+
+    if "date_acquired" not in df.columns:
+        return []
+
+    for time_col in TEMPORAL_TIME_COLUMNS:
+        if time_col not in df.columns:
+            continue
+        values = [
+            _stac_datetime_or_none(_combine_date_and_time(date_value, time_value))
+            for date_value, time_value in zip(
+                df["date_acquired"], df[time_col], strict=False
+            )
+        ]
+        cleaned = sorted({value for value in values if value is not None})
+        if cleaned:
+            return cleaned
+
+    values = [_stac_datetime_or_none(value) for value in df["date_acquired"]]
+    return sorted({value for value in values if value is not None})
+
+
 def _gcs_href(bucket: str, path: str) -> str:
-    return f"gs://{bucket}/{path}"
+    return (
+        f"https://storage.googleapis.com/{bucket}/{quote(path.lstrip('/'), safe='/')}"
+    )
 
 
 def _get_env_or_none(name: str) -> str | None:
@@ -56,12 +127,30 @@ def _build_geoserver_endpoint(
     default_path: str,
     override_env_name: str,
 ) -> str:
+    def _join_url_path(base_url: str, path: str) -> str:
+        normalized_base = base_url.rstrip("/")
+        normalized_path = f"/{path.lstrip('/')}"
+        if normalized_base.endswith(normalized_path):
+            return normalized_base
+        base_parts = normalized_base.split("/")
+        path_parts = normalized_path.lstrip("/").split("/")
+        overlap = 0
+        max_overlap = min(len(base_parts), len(path_parts))
+        for size in range(max_overlap, 0, -1):
+            if base_parts[-size:] == path_parts[:size]:
+                overlap = size
+                break
+        if overlap:
+            suffix = "/".join(path_parts[overlap:])
+            return normalized_base if not suffix else f"{normalized_base}/{suffix}"
+        return f"{normalized_base}{normalized_path}"
+
     override = _get_env_or_none(override_env_name)
     if override is None:
-        return f"{public_url.rstrip('/')}{default_path}"
+        return _join_url_path(public_url, default_path)
     if override.startswith("http://") or override.startswith("https://"):
         return override
-    return f"{public_url.rstrip('/')}/{override.lstrip('/')}"
+    return _join_url_path(public_url, override)
 
 
 def _geoserver_layer_name(collection_id: str, workspace: str) -> str:
@@ -158,17 +247,7 @@ def _stac_temporal_extent(
     df: pd.DataFrame,
     config: IngestConfig,
 ) -> tuple[str, str]:
-    temporal_values = (
-        sorted(
-            {
-                _stac_datetime_or_none(value)
-                for value in df.get("date_acquired", pd.Series())
-            }
-        )
-        if "date_acquired" in df.columns
-        else []
-    )
-    temporal_values = [value for value in temporal_values if value is not None]
+    temporal_values = _stac_datetimes_from_frame(df)
     if not temporal_values:
         fallback = _fallback_stac_datetime(config)
         return fallback, fallback
@@ -316,11 +395,7 @@ def build_stac_items(
         )
         geometry = mapping(geometry_point)
         bbox = [round(float(value), 6) for value in geometry_point.bounds]
-        datetimes = [
-            _stac_datetime_or_none(value)
-            for value in group.get("date_acquired", pd.Series())
-        ]
-        datetimes = sorted({value for value in datetimes if value is not None})
+        datetimes = _stac_datetimes_from_frame(group)
         if datetimes:
             datetime_value = datetimes[0] if len(datetimes) == 1 else None
             start_datetime = datetimes[0]
diff --git a/tests/test_aem.py b/tests/test_aem.py
@@ -1017,7 +1017,8 @@ def test_build_stac_payloads_are_deterministic():
     ]
     assert (
         collection["assets"]["parquet"]["href"]
-        == "gs://example-bucket/surveys/gila_animas_2025/out.parquet"
+        == "https://storage.googleapis.com/example-bucket/"
+        "surveys/gila_animas_2025/out.parquet"
     )
     assert [item["id"] for item in items] == [
         "aem-gila_animas_2025-preliminary_inversion-L1-R1",
@@ -1030,6 +1031,65 @@ def test_build_stac_payloads_are_deterministic():
     assert "wcs" not in collection["assets"]
 
 
+def test_build_stac_payloads_preserve_acquisition_time():
+    df = pd.DataFrame(
+        [
+            {
+                "line_id": "L1",
+                "record_id": "R1",
+                "easting": 500000,
+                "northing": 3800000,
+                "source_epsg": 32613,
+                "acquisition_datetime": datetime.datetime(
+                    2025, 3, 1, 14, 15, tzinfo=datetime.timezone.utc
+                ),
+            },
+            {
+                "line_id": "L2",
+                "record_id": "R2",
+                "easting": 500100,
+                "northing": 3800100,
+                "source_epsg": 32613,
+                "acquisition_datetime": datetime.datetime(
+                    2025, 3, 2, 9, 45, tzinfo=datetime.timezone.utc
+                ),
+            },
+        ]
+    )
+    config = IngestConfig(
+        filepath="/tmp/input.csv",
+        survey_id="gila_animas_2025",
+        processing_stage=ProcessingStage.PRELIMINARY,
+        inversion_code=InversionCode.SEOGI_PYTHON,
+        contractor="GeoTech/Seogi",
+        gcs_bucket="example-bucket",
+        source_gcs_path="surveys/gila_animas_2025/source.csv",
+    )
+
+    collection = aem_stac_service.build_stac_collection(
+        df=df,
+        config=config,
+        parquet_gcs_path="surveys/gila_animas_2025/out.parquet",
+        raw_manifest_gcs_path="surveys/gila_animas_2025/raw_files.json",
+    )
+    items = aem_stac_service.build_stac_items(
+        df=df,
+        config=config,
+        parquet_gcs_path="surveys/gila_animas_2025/out.parquet",
+        raw_manifest_gcs_path="surveys/gila_animas_2025/raw_files.json",
+    )
+
+    assert collection["extent"]["temporal"]["interval"] == [
+        [
+            "2025-03-01T14:15:00Z",
+            "2025-03-02T09:45:00Z",
+        ]
+    ]
+    assert items[0]["properties"]["datetime"] == "2025-03-01T14:15:00Z"
+    assert items[0]["properties"]["start_datetime"] == "2025-03-01T14:15:00Z"
+    assert items[0]["properties"]["end_datetime"] == "2025-03-01T14:15:00Z"
+
+
 def test_build_stac_collection_includes_survey_level_geoserver_assets(monkeypatch):
     monkeypatch.setenv("GEOSERVER_PUBLIC_URL", "https://maps.example.com")
     monkeypatch.setenv("GEOSERVER_WORKSPACE", "aem")
@@ -1080,6 +1140,45 @@ def test_build_stac_collection_includes_survey_level_geoserver_assets(monkeypatc
     assert collection["assets"]["wms"]["geoserver:layer"] == "aem:aem-gila_animas_2025"
 
 
+def test_build_stac_collection_deduplicates_geoserver_base_path(monkeypatch):
+    monkeypatch.setenv("GEOSERVER_PUBLIC_URL", "https://maps.example.com/geoserver")
+    monkeypatch.setenv("GEOSERVER_WORKSPACE", "aem")
+
+    df = pd.DataFrame(
+        [
+            {
+                "line_id": "L1",
+                "record_id": "R1",
+                "easting": 500000,
+                "northing": 3800000,
+                "source_epsg": 32613,
+            }
+        ]
+    )
+    config = IngestConfig(
+        filepath="/tmp/input.csv",
+        survey_id="gila_animas_2025",
+        processing_stage=ProcessingStage.PRELIMINARY,
+        inversion_code=InversionCode.SEOGI_PYTHON,
+        contractor="GeoTech/Seogi",
+        gcs_bucket="example-bucket",
+        source_gcs_path="surveys/gila_animas_2025/source.csv",
+    )
+
+    collection = aem_stac_service.build_stac_collection(
+        df=df,
+        config=config,
+        parquet_gcs_path="surveys/gila_animas_2025/out.parquet",
+        raw_manifest_gcs_path="surveys/gila_animas_2025/raw_files.json",
+    )
+
+    assert collection["assets"]["wcs"]["href"] == (
+        "https://maps.example.com/geoserver/ows"
+        "?service=WCS&version=2.0.1&request=DescribeCoverage"
+        "&coverageId=aem%3Aaem-gila_animas_2025"
+    )
+
+
 def test_load_stac_to_pgstac_uses_upsert(monkeypatch):
     calls = []
 
