Source code for pycsamt.zonge.heads

# pycsamt/zonge/heads.py
# Author: LKouadio <etanoyau@gmail.com>
# License: LGPL-3.0-or-later
"""
Header facade that aggregates survey-level property containers
(Hardware, Annotation, Configuration, Tx, Rx, Skip) and exposes
a uniform read/write API based on $keyword=value lines.

This module is intentionally *key/value* oriented, not tabular.
Use AVGComponentBase elsewhere for row/column data blocks.
"""

from __future__ import annotations

import re
from collections.abc import Mapping, Sequence
from dataclasses import asdict
from datetime import datetime, timezone
from pathlib import Path
from typing import (
    Any,
)

from .property import (
    Hardware,
    Receiver,
    SkipFlag,
    SurveyAnnotation,
    SurveyConfiguration,
    Transmitter,
)

__all__ = ["Header", "Head"]

# ------------------------------------------------------------------ #
# hardware banner parsing (commented lines from legacy files)        #
# ------------------------------------------------------------------ #
_AMTAVG_RX = re.compile(
    r"""^\\\s*AMTAVG\s*(?P<ver>[\d.]+)\s*:\s*      # version
        "(?P<src>[^"]+)"\s*,\s*                   # source file
        Dated\s*(?P<dated>[^,]+)\s*,\s*           # dated
        Processed\s*(?P<proc>.+)$                 # processed
    """,
    re.IGNORECASE | re.VERBOSE,
)

_AST_RX = re.compile(
    r"""^\\\s*ASTATIC\s*
        (?P<ver>v?[\d.]+[a-z]?)\s*updated\s*data\s*on\s*
        (?P<updated>.+)$
    """,
    re.IGNORECASE | re.VERBOSE,
)

_TMA_RX = re.compile(
    r"""^\\\s*(?P<pts>\d+)[- ]point\s*TMA\s*Filter\s*at\s*
        (?P<freq>[\d.]+)\s*hertz
    """,
    re.IGNORECASE | re.VERBOSE,
)

_KV_RX = re.compile(r"^\s*\$(?P<key>[^=\s]+)\s*=\s*(?P<val>.*)\s*$")


class HeaderComponentBase:
    """
    Minimal base for *header* components that serialize to
    $keyword=value lines (no CSV tables).

    Subclasses implement:
      - update_from_keywords(meta: dict[str, Any]) -> None
      - to_keywords() -> dict[str, Any]

    This base provides a consistent read/from_avg/write surface
    and utility helpers but keeps the format very lightweight.
    """

    def read(
        self,
        df_unused: Any = None,
        meta: Mapping[str, Any] | None = None,
    ) -> None:
        """Populate from a keyword mapping (ignore df)."""
        if meta:
            self.update_from_keywords(dict(meta))

    def write(self) -> Sequence[str]:
        """Serialize to $keyword=value lines."""
        return _dict_to_kv_lines(self.to_keywords())

    # Subclasses should provide these two:
    def update_from_keywords(self, meta: dict[str, Any]) -> None:  # noqa: D401
        raise NotImplementedError

    def to_keywords(self) -> dict[str, Any]:  # noqa: D401
        raise NotImplementedError


def _kv_lines_to_dict(lines: Sequence[str]) -> dict[str, str]:
    """Parse $key=value lines into a dict (last key wins)."""
    out: dict[str, str] = {}
    for ln in lines:
        m = _KV_RX.match(ln)
        if not m:
            continue
        k = m.group("key").strip()
        v = m.group("val").strip().strip('"')
        out[k] = v
    return out


def _dict_to_kv_lines(d: Mapping[str, Any]) -> list[str]:
    """Format mapping as $key=value lines (stable key order)."""
    lines: list[str] = []
    for k in sorted(d.keys()):
        v = d[k]
        v_str = str(v).replace("\n", " ").strip()
        lines.append(f"${k}={v_str}")
    return lines


def _now_utc_stamp() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def _parse_hardware_banner(lines: Sequence[str]) -> dict[str, Any]:
    r"""
    Extract a few useful fields from legacy banner lines starting
    with backslash, e.g.:

      \ AMTAVG 7.76: "LCS01.fld", Dated 99-01-01, Processed 22 Jul 16
      \ ASTATIC v3.60d updated data on 22/07/16
      \ 5-point TMA Filter at 1024 hertz
    """
    hw: dict[str, Any] = {}
    for ln in lines:
        if not ln.strip().startswith("\\"):
            continue
        m1 = _AMTAVG_RX.match(ln)
        if m1:
            hw["version"] = m1.group("ver")
            hw["source_file"] = Path(m1.group("src"))
            hw["dated"] = m1.group("dated").strip()
            hw["processed"] = m1.group("proc").strip()
            continue
        m2 = _AST_RX.match(ln)
        if m2:
            hw["astatic_ver"] = m2.group("ver")
            hw["updated"] = m2.group("updated").strip()
            continue
        m3 = _TMA_RX.match(ln)
        if m3:
            hw["tma_points"] = int(m3.group("pts"))
            hw["tma_freq"] = float(m3.group("freq"))
            continue
    return hw


def _emit_hardware_banner(hw: Hardware) -> list[str]:
    """
    Emit a human-friendly banner (comment lines) at the top of
    the header. Values are best-effort; missing fields omitted.
    """
    out: list[str] = []
    # AMTAVG line
    parts: list[str] = []  # noqa
    ver = hw.get("version")
    src = hw.get("source_file")
    dat = hw.get("dated")
    pro = hw.get("processed")
    if ver or src or dat or pro:
        p = []
        if ver:
            p.append(f"AMTAVG {ver}:")
        if src:
            name = Path(src).name if isinstance(src, str) else src.name
            p.append(f'"{name}",')
        if dat:
            p.append(f"Dated {dat},")
        if pro:
            p.append(f"Processed {pro}")
        if p:
            out.append("\\ " + " ".join(p).rstrip(","))

    # ASTATIC line
    aver = hw.get("astatic_ver")
    upd = hw.get("updated")
    if aver or upd:
        p = ["ASTATIC"]
        if aver:
            p.append(str(aver))
        if upd:
            p.append("updated data on " + str(upd))
        out.append("\\ " + " ".join(p))

    # TMA filter line
    pts = hw.get("tma_points")
    frq = hw.get("tma_freq")
    if pts or frq:
        segs = []
        if pts:
            segs.append(f"{int(pts)}-point")
        segs.append("TMA Filter")
        if frq:
            segs.append(f"at {float(frq):g} hertz")
        out.append("\\ " + " ".join(segs))

    return out






# Backward-compatible alias
Head = Header