Cette page a été traduite automatiquement de l’anglais et peut contenir des erreurs. Original en anglais
Code source de axqua.flowtracker
"""FlowTracker2 -> calibration-target hydraulics tab.
Reads SonTek FlowTracker2 exports and writes one row per measurement point into
the ``hydraulics`` tab of a ``calibration-target-data.xlsx`` template, keyed by
each point's own ID (which then joins the DGPS point layer in
``user-sources/geodata/``, see :mod:`axqua.targets`).
Per point it extracts the three velocity components and their turbulent RMS
fluctuations:
* ``u_x, u_y, u_z`` <- the reported mean components ``VelX/VelY/VelZ`` (aka
``v(x)/v(y)/v(z)``).
* ``u_x', u_y', u_z'`` (RMS fluctuations) <- the sample **standard deviation**
``u std`` / ``Std Dev v'(x)`` / ``sigma_x``. **These are NOT the ``VxErr``
columns:** ``VxErr`` is the FlowTracker *standard error of the mean* (the
velocity measurement uncertainty), i.e. ~``std / sqrt(Npts)``. When only a
plain ``.ft.sum`` summary is available (``VxErr`` + ``Npts`` but no std-dev
column), the RMS is reconstructed as ``u' ~ VxErr * sqrt(Npts)``.
``TKE`` is taken from a measured column when present, else left to the
template's live formula ``0.5*(u_x'^2 + u_y'^2 + u_z'^2)``; the water depth is
the local total/final depth. ``U_h`` and ``U_h'`` recompute from the components.
Three export layouts are auto-detected by their headers:
1. **SonTek ``.ft.sum``** (``ID, MeasD, Npts, VelX/VelY/VelZ, VxErr/VyErr/VzErr,
FinalD``): only stderr -> RMS reconstructed.
2. **TKE-stats export** (``VelX...`` plus ``u std/v std/w std``): real RMS.
3. **FT_TKE_Summary** (``Point, v(x)..., Std Dev v'(x)..., TKE [m2/s2]``): real
RMS + measured TKE; multiple worksheets scanned.
"""
from __future__ import annotations
import logging
import re
import zipfile
from pathlib import Path
import numpy as np
import pandas as pd
from axqua.ground_truth import read_xlsx_sheet
from axqua.targets import _norm_header
log = logging.getLogger("axqua")
# --------------------------------------------------------------------------- #
# Header classification (robust to the three FlowTracker layouts)
# --------------------------------------------------------------------------- #
def _norm(h: object) -> str:
"""Lower-case a header and drop trailing unit annotations.
Strips ``[m/s]``-style brackets and known unit parentheses (``(m/s)``,
``(deg)``, ...) but keeps a component parenthesis like ``(x)`` so that
``v(x)``/``v(y)``/``v(z)`` stay distinct.
"""
s = str(h).strip().lower()
s = re.sub(r"\[[^\]]*\]", "", s)
s = re.sub(
r"\((?:m/s|m2/s2|m²/s²|m|deg|°c|db|-|°|s|"
r"ellipsoidal[^)]*|wgs[^)]*|dgps[^)]*)\)",
"", s)
return re.sub(r"\s+", " ", s).strip()
# canonical field -> the normalized headers that map to it, in priority order
_FIELD_HEADERS: dict[str, tuple[str, ...]] = {
"u": ("velx", "v(x)", "vx", "u average", "average v(x)"),
"v": ("vely", "v(y)", "vy", "v average", "average v(y)"),
"w": ("velz", "v(z)", "vz", "w average", "average v(z)"),
"u_std": ("u std", "std dev v’(x)", "std dev v'(x)", "σx", "sigmax"),
"v_std": ("v std", "std dev v’(y)", "std dev v'(y)", "σy", "sigmay"),
"w_std": ("w std", "std dev v’(z)", "std dev v'(z)", "σz", "sigmaz"),
"u_err": ("vxerr", "v_err(x)", "u stderr"),
"v_err": ("vyerr", "v_err(y)", "v stderr"),
"w_err": ("vzerr", "v_err(z)", "w stderr"),
"npts": ("npts",),
"tke": ("tke",), # measured 3D TKE in m2/s2 (exact)
"meas_depth": ("measd", "meas. depth", "meas depth"),
"depth": ("finald", "total depth"),
"id": ("id", "point", "#", "st", "st (dgps point id)"),
}
_VEL_HEADERS = {h for f in ("u", "v", "w") for h in _FIELD_HEADERS[f]}
def _build_column_map(header_cells: list) -> dict[str, int]:
"""Map canonical fields to column indices for one header row."""
normed = {i: _norm(c) for i, c in enumerate(header_cells)}
colmap: dict[str, int] = {}
for field, candidates in _FIELD_HEADERS.items():
for cand in candidates:
hit = next((i for i, n in normed.items() if n == cand), None)
if hit is not None:
colmap[field] = hit
break
return colmap
def _to_float(value: object) -> float:
try:
f = float(value)
except (TypeError, ValueError):
return np.nan
return f
def _clean_id(value: object) -> object | None:
"""Integral floats -> int (so 3.0 joins '3'); text IDs kept verbatim."""
if value is None:
return None
s = str(value).strip()
try:
f = float(s)
except ValueError:
return s or None
if np.isnan(f):
return None
return int(f) if f == int(f) else f
# --------------------------------------------------------------------------- #
# Reading
# --------------------------------------------------------------------------- #
def _sheet_count(path: Path) -> int:
with zipfile.ZipFile(path) as z:
return sum(1 for n in z.namelist()
if re.fullmatch(r"xl/worksheets/sheet\d+\.xml", n))
def _read_sheet_points(raw: pd.DataFrame, reconstruct_rms: bool) -> pd.DataFrame | None:
"""Extract per-point velocities/fluctuations from one raw worksheet frame."""
header_idx = next(
(i for i in range(len(raw))
if any(_norm(c) in _VEL_HEADERS for c in raw.iloc[i])),
None)
if header_idx is None:
return None
colmap = _build_column_map(list(raw.iloc[header_idx]))
if not all(f in colmap for f in ("u", "v", "w")):
return None
# a genuine summary/point table always carries a per-point error, std-dev or
# TKE column; this rejects raw per-sample time-series sheets (which expose
# only u/v/w) that a multi-sheet workbook may also contain.
if not any(f in colmap for f in ("u_err", "v_err", "w_err",
"u_std", "v_std", "w_std", "tke")):
return None
# data starts after the header, skipping a units row (non-numeric velocity)
start = header_idx + 1
if start < len(raw) and np.isnan(_to_float(raw.iloc[start, colmap["u"]])):
start += 1
records: list[dict] = []
for i in range(start, len(raw)):
row = raw.iloc[i]
u = _to_float(row[colmap["u"]])
if np.isnan(u):
continue # blank / trailing row
rec: dict[str, object] = {"u": u,
"v": _to_float(row[colmap["v"]]),
"w": _to_float(row[colmap["w"]])}
rid = _clean_id(row[colmap["id"]]) if "id" in colmap else None
rec["ID"] = rid if rid is not None else len(records)
for f in ("u_std", "v_std", "w_std", "u_err", "v_err", "w_err",
"npts", "tke", "meas_depth", "depth"):
if f in colmap:
rec[f] = _to_float(row[colmap[f]])
records.append(rec)
if not records:
return None
df = pd.DataFrame(records)
# RMS fluctuations: prefer the measured std dev; else reconstruct from the
# standard error of the mean and the sample count (u' ~ stderr * sqrt(Npts)).
npts = df["npts"] if "npts" in df else pd.Series(np.nan, index=df.index)
for comp in ("u", "v", "w"):
std_col, err_col = f"{comp}_std", f"{comp}_err"
std = df[std_col] if std_col in df else pd.Series(np.nan, index=df.index)
if reconstruct_rms and err_col in df:
reconstructed = df[err_col] * np.sqrt(npts)
std = std.where(std.notna(), reconstructed)
df[std_col] = std
keep = ["ID", "u", "v", "w", "u_std", "v_std", "w_std", "tke", "depth"]
for c in keep:
if c not in df:
df[c] = np.nan
return df[keep]
[docs]
def read_flowtracker(path: Path, *, sheet: str | int | None = None,
reconstruct_rms: bool = True) -> pd.DataFrame:
"""Read one FlowTracker2 export -> per-point velocity/fluctuation table.
Returns columns ``ID, u, v, w, u_std, v_std, w_std, tke, depth`` (one row
per measurement point). All worksheets that look like a FlowTracker table
are scanned and concatenated unless ``sheet`` restricts to one (name or
0-based index). With ``reconstruct_rms`` the RMS fluctuations are filled
from ``VxErr * sqrt(Npts)`` where a direct std-dev column is absent.
"""
path = Path(path)
if sheet is not None:
raw = read_xlsx_sheet(path, sheet=sheet)
df = _read_sheet_points(raw, reconstruct_rms)
if df is None:
raise ValueError(f"no FlowTracker velocity table found in {path.name} "
f"sheet {sheet!r}")
parts = [df]
else:
parts = []
for i in range(_sheet_count(path)):
df = _read_sheet_points(read_xlsx_sheet(path, sheet=i), reconstruct_rms)
if df is not None:
log.info(" %s sheet %d: %d FlowTracker point(s)", path.name, i, len(df))
parts.append(df)
if not parts:
raise ValueError(f"no FlowTracker velocity table found in {path.name}")
return pd.concat(parts, ignore_index=True)
[docs]
def read_flowtrackers(paths, *, reconstruct_rms: bool = True) -> pd.DataFrame:
"""Read and concatenate several FlowTracker2 exports (e.g. day1 + day2)."""
frames = [read_flowtracker(p, reconstruct_rms=reconstruct_rms) for p in paths]
return pd.concat(frames, ignore_index=True) if frames else pd.DataFrame()
# --------------------------------------------------------------------------- #
# Writing into the calibration-target hydraulics tab
# --------------------------------------------------------------------------- #
[docs]
def fill_template_hydraulics(template: Path, sources, *, sheet_name: str = "hydraulics",
start_row: int = 2, reconstruct_rms: bool = True) -> int:
"""Write FlowTracker points into the template's hydraulics tab, keyed by ID.
``sources`` is one path or a list of FlowTracker2 exports. Existing rows in
the tab are cleared first (so re-running is idempotent). ``u_x/u_y/u_z`` and
``u_x'/u_y'/u_z'`` are written; ``U_h``, ``U_h'`` and (unless a measured TKE
is present) ``TKE`` keep their live formulas; the water depth is filled.
Returns the number of points written.
"""
from openpyxl import load_workbook
from openpyxl.utils import get_column_letter
template = Path(template)
if not template.exists():
raise FileNotFoundError(
f"template not found: {template}. Run `axqua targets <config>` first.")
paths = [Path(sources)] if isinstance(sources, (str, Path)) else [Path(p) for p in sources]
points = read_flowtrackers(paths, reconstruct_rms=reconstruct_rms)
if points.empty:
log.warning("no FlowTracker points found in %s", [p.name for p in paths])
return 0
wb = load_workbook(template)
if sheet_name not in wb.sheetnames:
raise ValueError(f"template {template.name} has no {sheet_name!r} sheet")
ws = wb[sheet_name]
# locate columns from the header row (robust to any future reordering)
header = {(_norm_header(ws.cell(row=1, column=c).value)): c
for c in range(1, ws.max_column + 1)}
need = {"id": "id", "u": "u_x", "v": "u_y", "w": "u_z",
"u_std": "u_x'", "v_std": "u_y'", "w_std": "u_z'",
"u_h": "u_h", "u_h_std": "u_h'", "tke": "tke", "depth": "water depth"}
missing = [h for h in need.values() if h not in header]
if missing:
raise ValueError(f"{template.name} {sheet_name!r} tab is missing columns {missing}")
col = {k: header[v] for k, v in need.items()}
letter = {k: get_column_letter(c) for k, c in col.items()}
# clear any previously-written rows (data columns only; keep the formulas)
data_cols = [col[k] for k in ("id", "u", "v", "w", "u_std", "v_std", "w_std", "depth")]
last = start_row - 1
for r in range(start_row, ws.max_row + 1):
if ws.cell(row=r, column=col["id"]).value not in (None, ""):
last = r
for r in range(start_row, last + 1):
for c in data_cols:
ws.cell(row=r, column=c).value = None
def _cell(r: int, key: str, value) -> None:
if value is not None and not (isinstance(value, float) and np.isnan(value)):
ws.cell(row=r, column=col[key]).value = round(float(value), 6) \
if key != "id" else value
for i, (_, p) in enumerate(points.iterrows()):
r = start_row + i
_cell(r, "id", p["ID"])
for k in ("u", "v", "w", "u_std", "v_std", "w_std", "depth"):
_cell(r, k, p[k])
# (re)write the derived-quantity formulas so they are correct regardless
# of the tab's prior state
ws.cell(row=r, column=col["u_h"]).value = (
f'=IF(OR({letter["u"]}{r}="",{letter["v"]}{r}=""),"",'
f'SQRT({letter["u"]}{r}^2+{letter["v"]}{r}^2))')
ws.cell(row=r, column=col["u_h_std"]).value = (
f'=IF(OR({letter["u_std"]}{r}="",{letter["v_std"]}{r}=""),"",'
f'SQRT({letter["u_std"]}{r}^2+{letter["v_std"]}{r}^2))')
if pd.notna(p["tke"]):
_cell(r, "tke", p["tke"]) # measured TKE wins over the proxy
else:
ws.cell(row=r, column=col["tke"]).value = (
f'=IF(OR({letter["u_std"]}{r}="",{letter["v_std"]}{r}="",'
f'{letter["w_std"]}{r}=""),"",0.5*({letter["u_std"]}{r}^2+'
f'{letter["v_std"]}{r}^2+{letter["w_std"]}{r}^2))')
wb.save(template)
n_rms = int(points[["u_std", "v_std", "w_std"]].notna().all(axis=1).sum())
log.info("wrote %d FlowTracker point(s) into %s (%r tab); %d with full RMS "
"fluctuations", len(points), template.name, sheet_name, n_rms)
return len(points)
# thin runnable driver dropped next to the generated template (see
# axqua.targets.write_target_template) so users get a co-located script.
DRIVER_SCRIPT = '''#!/usr/bin/env python
"""Extract FlowTracker2 velocities + RMS fluctuations into the hydraulics tab of
calibration-target-data.xlsx (this folder), keyed by each point's ID.
Run in the axqua env, from this folder:
mamba run -n axqua-env python extract_flowtracker.py FILE1.xlsx [FILE2.xlsx ...]
With no arguments it picks up *FlowTracker*/*ft*sum* .xlsx files in this folder.
Each point becomes one row keyed by its ID (must match the ID field of the
ground_truth.targets.hydraulics_positions layer). u_x/u_y/u_z come from
VelX/VelY/VelZ; the RMS fluctuations u_x'/u_y'/u_z' come from the sample standard
deviation (u std / Std Dev v'(x) / sigma) - NOT VxErr, which is the standard
error of the mean (reconstructed as VxErr*sqrt(Npts) only when no std-dev column
exists). U_h, U_h' and TKE recompute from these in the sheet.
"""
import sys
from pathlib import Path
from axqua.flowtracker import fill_template_hydraulics
HERE = Path(__file__).resolve().parent
TEMPLATE = HERE / "calibration-target-data.xlsx"
def main(argv):
files = [Path(a) for a in argv]
if not files:
files = sorted(HERE.glob("*[Ff]low[Tt]racker*.xlsx")) + sorted(HERE.glob("*ft*sum*"))
if not files:
print("usage: python extract_flowtracker.py <flowtracker2 xlsx> [more ...]")
return 1
n = fill_template_hydraulics(TEMPLATE, files)
print(f"wrote {n} FlowTracker point(s) into {TEMPLATE.name}")
return 0
if __name__ == "__main__":
raise SystemExit(main(sys.argv[1:]))
'''
[docs]
def write_driver_script(path: Path, force: bool = False) -> Path | None:
"""Drop the ``extract_flowtracker.py`` driver next to the template."""
path = Path(path)
if path.exists() and not force:
return None
path.write_text(DRIVER_SCRIPT)
return path