add calibration file

rfg_adc_plotter/state/app_state.py

@@ -1,14 +1,21 @@
 """Состояние приложения: текущие свипы и настройки калибровки/нормировки."""
 
+import os
 from queue import Empty, Queue
 from typing import Any, Dict, Mapping, Optional
 
 import numpy as np
 
-from rfg_adc_plotter.processing.normalizer import normalize_by_calib
+from rfg_adc_plotter.processing.normalizer import (
+    build_calib_envelopes,
+    normalize_by_calib,
+    normalize_by_envelope,
+)
 from rfg_adc_plotter.state.ring_buffer import RingBuffer
 from rfg_adc_plotter.types import SweepInfo, SweepPacket
 
+CALIB_ENVELOPE_PATH = "calib_envelope.npy"
+
 
 def format_status(data: Mapping[str, Any]) -> str:
     """Преобразовать словарь метрик в одну строку 'k:v'."""
@@ -44,21 +51,65 @@ class AppState:
         self.current_info: Optional[SweepInfo] = None
         self.calib_enabled: bool = False
         self.norm_type: str = norm_type
+        # "live" — нормировка по текущему ch0-свипу; "file" — по огибающей из файла
+        self.calib_mode: str = "live"
+        self.calib_file_envelope: Optional[np.ndarray] = None
 
     def _normalize(self, raw: np.ndarray, calib: np.ndarray) -> np.ndarray:
+        if self.calib_mode == "file" and self.calib_file_envelope is not None:
+            return normalize_by_envelope(raw, self.calib_file_envelope)
         return normalize_by_calib(raw, calib, self.norm_type)
 
+    def save_calib_envelope(self, path: str = CALIB_ENVELOPE_PATH) -> bool:
+        """Вычислить огибающую из last_calib_sweep и сохранить в файл.
+
+        Возвращает True при успехе.
+        """
+        if self.last_calib_sweep is None:
+            return False
+        try:
+            _lower, upper = build_calib_envelopes(self.last_calib_sweep)
+            np.save(path, upper)
+            return True
+        except Exception as exc:
+            import sys
+            sys.stderr.write(f"[warn] Не удалось сохранить огибающую: {exc}\n")
+            return False
+
+    def load_calib_envelope(self, path: str = CALIB_ENVELOPE_PATH) -> bool:
+        """Загрузить огибающую из файла.
+
+        Возвращает True при успехе.
+        """
+        if not os.path.isfile(path):
+            return False
+        try:
+            env = np.load(path)
+            self.calib_file_envelope = np.asarray(env, dtype=np.float32)
+            return True
+        except Exception as exc:
+            import sys
+            sys.stderr.write(f"[warn] Не удалось загрузить огибающую: {exc}\n")
+            return False
+
+    def set_calib_mode(self, mode: str):
+        """Переключить режим калибровки: 'live' или 'file'."""
+        self.calib_mode = mode
+
     def set_calib_enabled(self, enabled: bool):
         """Включить/выключить режим калибровки, пересчитать norm-свип."""
         self.calib_enabled = enabled
-        if (
-            self.calib_enabled
-            and self.current_sweep_raw is not None
-            and self.last_calib_sweep is not None
-        ):
-            self.current_sweep_norm = self._normalize(
-                self.current_sweep_raw, self.last_calib_sweep
-            )
+        if self.calib_enabled and self.current_sweep_raw is not None:
+            if self.calib_mode == "file" and self.calib_file_envelope is not None:
+                self.current_sweep_norm = normalize_by_envelope(
+                    self.current_sweep_raw, self.calib_file_envelope
+                )
+            elif self.calib_mode == "live" and self.last_calib_sweep is not None:
+                self.current_sweep_norm = self._normalize(
+                    self.current_sweep_raw, self.last_calib_sweep
+                )
+            else:
+                self.current_sweep_norm = None
         else:
             self.current_sweep_norm = None
 
@@ -86,11 +137,17 @@ class AppState:
             # Канал 0 — опорный (калибровочный) свип
             if ch == 0:
                 self.last_calib_sweep = s
+                self.save_calib_envelope()
                 self.current_sweep_norm = None
                 sweep_for_ring = s
             else:
-                if self.calib_enabled and self.last_calib_sweep is not None:
-                    self.current_sweep_norm = self._normalize(s, self.last_calib_sweep)
+                can_normalize = self.calib_enabled and (
+                    (self.calib_mode == "file" and self.calib_file_envelope is not None)
+                    or (self.calib_mode == "live" and self.last_calib_sweep is not None)
+                )
+                if can_normalize:
+                    calib_ref = self.last_calib_sweep if self.last_calib_sweep is not None else s
+                    self.current_sweep_norm = self._normalize(s, calib_ref)
                     sweep_for_ring = self.current_sweep_norm
                 else:
                     self.current_sweep_norm = None