Add avg power and NP to activity summary; NP uses Coggan 30s rolling-average method

bincio/extract/metrics.py

@@ -53,6 +53,7 @@ class ComputedMetrics:
     max_hr_bpm: Optional[int]
     avg_cadence_rpm: Optional[int]
     avg_power_w: Optional[int]
+    np_power_w: Optional[int]
     max_power_w: Optional[int]
     bbox: Optional[tuple[float, float, float, float]]   # min_lon, min_lat, max_lon, max_lat
     start_latlng: Optional[tuple[float, float]]
@@ -74,6 +75,7 @@ def compute(activity: ParsedActivity) -> ComputedMetrics:
     avg_hr, max_hr = _hr_stats(pts)
     avg_cad = _avg_nonnull([p.cadence_rpm for p in pts])
     avg_pow = _avg_nonnull([p.power_w for p in pts])
+    np_pow  = _np_power(pts, activity.started_at)
     max_pow = _max_nonnull([p.power_w for p in pts])
     bbox = _bbox(pts)
     start_ll, end_ll = _endpoints(pts)
@@ -92,6 +94,7 @@ def compute(activity: ParsedActivity) -> ComputedMetrics:
         max_hr_bpm=max_hr,
         avg_cadence_rpm=avg_cad,
         avg_power_w=avg_pow,
+        np_power_w=np_pow,
         max_power_w=max_pow,
         bbox=bbox,
         start_latlng=start_ll,
@@ -415,6 +418,50 @@ def _max_nonnull(values: list) -> Optional[int]:
     return max(v) if v else None
 
 
+def _np_power(pts: list[DataPoint], started_at: datetime) -> Optional[int]:
+    """Normalized power (Coggan method): 30 s rolling average → 4th power → mean → 4th root.
+
+    Uses a dense 1 Hz series (gaps zero-filled) identical to the MMP pipeline.
+    Returns None when the activity has no power data or is shorter than 30 s.
+    """
+    sparse: dict[int, int] = {}
+    last_t = -1
+    for p in pts:
+        t = int((p.timestamp - started_at).total_seconds())
+        if t < 0 or t == last_t:
+            continue
+        last_t = t
+        if p.power_w is not None:
+            sparse[t] = p.power_w
+
+    if len(sparse) < 2:
+        return None
+
+    t_min, t_max = min(sparse), max(sparse)
+    if t_max - t_min > 7 * 24 * 3600:
+        return None
+
+    power_1hz = [sparse.get(t, 0) for t in range(t_min, t_max + 1)]
+    n = len(power_1hz)
+    win = 30
+    if n < win:
+        return None
+
+    # 30 s centred rolling mean, then raise to 4th power
+    half = win // 2
+    total = sum(power_1hz[:win])
+    fourth_powers: list[float] = []
+    for i in range(half, n - half):
+        avg = total / win
+        fourth_powers.append(avg ** 4)
+        if i + half + 1 < n:
+            total += power_1hz[i + half + 1] - power_1hz[i - half]
+
+    if not fourth_powers:
+        return None
+    return int(round((sum(fourth_powers) / len(fourth_powers)) ** 0.25))
+
+
 def _bbox(pts: list[DataPoint]) -> Optional[tuple[float, float, float, float]]:
     lats = [p.lat for p in pts if p.lat is not None]
     lons = [p.lon for p in pts if p.lon is not None]
@@ -438,7 +485,7 @@ def _empty() -> ComputedMetrics:
         elevation_gain_m=None, elevation_loss_m=None,
         avg_speed_kmh=None, max_speed_kmh=None,
         avg_hr_bpm=None, max_hr_bpm=None,
-        avg_cadence_rpm=None, avg_power_w=None, max_power_w=None,
+        avg_cadence_rpm=None, avg_power_w=None, np_power_w=None, max_power_w=None,
         bbox=None, start_latlng=None, end_latlng=None,
         mmp=None, best_efforts=None, best_climb_m=None,
     )