#!/usr/bin/env python3 """Фазовая калибровка MIMO-радара по точечному отражателю (шару) и применение её к данным. Расчёт тот же, что в MIMO_calibration.ipynb, без графиков и диагностики. Формат файлов калибровки (cal_*.npy, cal_meta.json) и папки postprocessed — тот же. Запуск целиком: python mimo_calibration.py (работает по параметрам из блока ниже) Из другого кода: import mimo_calibration as mc cal = mc.compute_calibration(cal_sets, empty_scene_path, mc.Settings(...)) mc.save_calibration(cal, cal_dir) cal = mc.load_calibration(cal_dir) s21_corr = mc.apply_calibration(s21, freq, cal) # массив (TX, RX, Nf) в памяти mc.apply_to_measurement(meas_dir, cal) # preprocessed -> postprocessed """ import datetime import json import re import shutil from dataclasses import dataclass from pathlib import Path import numpy as np from scipy.optimize import minimize # ╔══════════════════════════════════════════════════════════════════════════════════════╗ # ║ ЗАДАВАЕМЫЕ ПАРАМЕТРЫ ║ # ╚══════════════════════════════════════════════════════════════════════════════════════╝ # --- что делать ------------------------------------------------------------------------- RUN_CALIBRATION = True # True -> посчитать калибровку по CAL_SETS и записать в CAL_DIR; # False -> взять готовую калибровку из CAL_DIR APPLY_TO_DATA = False # True -> записать исправленные данные в <измерение>/postprocessed # --- калибровочные измерения ------------------------------------------------------------ # Каждая запись — одно положение отражателя. snapshots: список префиксов снимков ('0003') # или None (все снимки папки). Несколько снимков одной записи усредняются. # target_scene_m = (X, Y, Z) в осях сцены, метры; для шара — координаты его ЦЕНТРА, # измеренные физически (а не взятые с некалиброванного изображения): # X — поперёк, вправо, от начала координат антенн (config_profile, x = 0); # Y — высота центра отражателя над землёй; # Z — расстояние вдоль земли от точки под началом координат антенн до отражателя. CAL_ROOT = Path('/Users/ivan_root/Downloads/Telegram_dwnld/2026-09-31 balls calib/' 'librevna_multi_0.5-6GHz_1031pts_-3dBm_big_ball_h_0cm_z_5m_x_0cm/preprocessed') CAL_SETS = [ dict(path=CAL_ROOT, snapshots=['0000'], target_scene_m=(0.0, 0.16, 5.20)), # Здесь нужно указывать координаты, измернные dict(path=CAL_ROOT, snapshots=['0001'], target_scene_m=(0.0, 0.16, 5.20)), # в системе координат земли (x,y,z) dict(path=CAL_ROOT, snapshots=['0002'], target_scene_m=(0.0, 0.16, 5.20)), # X - поеперчная координата от центра радара dict(path=CAL_ROOT, snapshots=['0003'], target_scene_m=(0.0, 0.16, 5.20)), # y - выота от поверхности земли dict(path=CAL_ROOT, snapshots=['0004'], target_scene_m=(0.0, 0.16, 5.20)), # z - длина вдоль земли от вертикальной плоскости, про ] # проходящей через нижние антенны до центра шара (до переднего края + радиус) SPHERE_RADIUS_M = 0.16 # радиус шара-отражателя [м]; 0 -> точечная цель EMPTY_SCENE_PATH = Path('/Users/ivan_root/Downloads/Telegram_dwnld/2026-09-31 balls calib/' 'librevna_multi_0.5-6GHz_1031pts_-3dBm_bkgd_h_0cm/preprocessed') # папка снимков той же сцены БЕЗ отражателя (вычитается их среднее); # None -> без вычитания # --- геометрия (как в софте построения изображений) --------------------------------------- INPUT_IDX = [0, 1, 2, 3] # физические входы (Rx), input_pos OUTPUT_IDX = [0, 1, 2, 3] # физические выходы (Tx), output_pos ANTENNA_HEIGHT_M = 0.92 # высота начала координат антенн (нижний ряд) над землёй [м] PITCH_DEG = 10.0 # наклон оси радара ниже горизонта [град] # --- оценка ----------------------------------------------------------------------------- CAL_F_MIN = 2.0e9 # полоса, по которой оценивается калибровка [Гц] CAL_F_MAX = 6.0e9 GATE_HALF_M = 0.15 # окно вокруг эха цели, ± по бистатическому пути [м] GATE_TAPER_M = 0.03 # плавные края окна [м] COARSE_RANGE_SEARCH_M = 0.6 # грубый поиск эха по дальности перед точной подгонкой, ± по пути [м]; # 0 -> выключен REFINE_AXES = ('x', 'y', 'z') # какие координаты цели уточнять по данным; если X и Y измерены # точно — ('z',) REFINE_RANGE_M = 0.10 # уточнение не дальше этого от найденной точки [м] APPLY_AMPLITUDE = False # True -> поправка выравнивает и амплитуды антенн PHASE_MODEL = 'linear' # модель фазы антенны по частоте: 'const' | 'linear' | 'poly3' | 'raw' # --- куда писать калибровку --------------------------------------------------------------- CAL_NAME = 'cal_20260931_big_ball_h0_script' CAL_DIR = Path(__file__).resolve().parent / 'results' / 'calibration' / CAL_NAME OVERWRITE = True # True -> перезаписать файлы калибровки в CAL_DIR # --- к каким измерениям применять --------------------------------------------------------- # Список папок измерений (тех, где лежат preprocessed, raw, results). APPLY_TO_MEASUREMENTS = [ # '/путь/к/измерению', ] OVERWRITE_POSTPROCESSED = False # True -> пересоздать postprocessed (только созданный этой калибровкой: # папка без calibration_applied.json не удаляется никогда) # ╔══════════════════════════════════════════════════════════════════════════════════════╗ # ║ КОНЕЦ ЗАДАВАЕМЫХ ПАРАМЕТРОВ ║ # ╚══════════════════════════════════════════════════════════════════════════════════════╝ C_LIGHT = 299_792_458.0 @dataclass class Settings: """Настройки расчёта; значения по умолчанию — из блока параметров выше.""" sphere_radius_m: float = SPHERE_RADIUS_M input_idx: tuple = tuple(INPUT_IDX) output_idx: tuple = tuple(OUTPUT_IDX) antenna_height_m: float = ANTENNA_HEIGHT_M pitch_deg: float = PITCH_DEG f_min: float = CAL_F_MIN f_max: float = CAL_F_MAX gate_half_m: float = GATE_HALF_M gate_taper_m: float = GATE_TAPER_M coarse_range_search_m: float = COARSE_RANGE_SEARCH_M refine_axes: tuple = tuple(REFINE_AXES) refine_range_m: float = REFINE_RANGE_M apply_amplitude: bool = APPLY_AMPLITUDE phase_model: str = PHASE_MODEL # ══════════════════════════════════════════════════════ # ЗАГРУЗКА # ══════════════════════════════════════════════════════ def find_config_profile(start, max_levels=8): p = Path(start) for _ in range(max_levels): if (p / 'config_profile.json').exists(): return p / 'config_profile.json' p = p.parent raise FileNotFoundError(f'config_profile.json не найден выше {start}') def load_geometry(config_path, input_idx, output_idx): """Координаты антенн в осях радара [м], массивы (N,3) в порядке сортировки физических портов.""" gpr = json.load(open(config_path, encoding='utf-8'))['gpr'] tx = {int(e['output_pos']): (e['x_m'], e.get('y_m', 0.0), e.get('z_m', 0.0)) for e in gpr['tx_geometry']} rx = {int(e['input_pos']): (e['x_m'], e.get('y_m', 0.0), e.get('z_m', 0.0)) for e in gpr['rx_geometry']} return (np.array([tx[k] for k in sorted(output_idx)], float), np.array([rx[k] for k in sorted(input_idx)], float)) def list_snapshots(root, prefixes=None): dirs = sorted(p for p in Path(root).iterdir() if p.is_dir() and (p / 'meta.json').exists()) if prefixes is None: return dirs out = [d for d in dirs if any(d.name.startswith(str(pr)) for pr in prefixes)] if len(out) != len(prefixes): raise FileNotFoundError(f'В {root} найдены не все снимки {prefixes}') return out def load_s21(snapshot_dir, input_idx, output_idx): """S21 (TX, RX, Nf) и частоты. Индексы — порядок сортировки физических портов.""" s21, freq = None, None for a, o in enumerate(sorted(output_idx)): for b, i in enumerate(sorted(input_idx)): s = np.load(Path(snapshot_dir) / f'i{i}_o{o}_s21.npy') f = np.load(Path(snapshot_dir) / f'i{i}_o{o}_freq.npy') if s21 is None: s21 = np.zeros((len(output_idx), len(input_idx), s.size), complex); freq = f elif not np.allclose(f, freq): raise ValueError(f'Разные частотные сетки у пар в {snapshot_dir}') s21[a, b] = s return s21, freq # ══════════════════════════════════════════════════════ # РАСЧЁТ КАЛИБРОВКИ # ══════════════════════════════════════════════════════ def best_phases(Z): """max |Σ Z_ij·e^{-j(t_i+r_j)}| по фазам антенн, t_0 = r_0 = 0.""" t = np.zeros(Z.shape[0]); r = np.zeros(Z.shape[1]) for _ in range(100): t = np.angle((Z * np.exp(-1j * r)[None, :]).sum(axis=1)) r = np.angle((Z * np.exp(-1j * t)[:, None]).sum(axis=0)) t = t - t[0]; r = r - r[0] return t, r def coherence(Z, t=None, r=None): """Сфазированность вкладов пар: |Σ| / Σ|·| (1 — все в фазе).""" if t is not None: Z = Z * np.exp(-1j * (t[:, None] + r[None, :])) return float(np.abs(Z.sum()) / (np.abs(Z).sum() + 1e-30)) def circular_median(phases): """Круговая медиана по первой оси: значение с минимальной суммой угловых расстояний до остальных.""" d = np.abs(np.angle(np.exp(1j * (phases[:, None] - phases[None, :])))) best = np.argmin(d.sum(axis=1), axis=0) return np.take_along_axis(phases, best[None], axis=0)[0] def combine(sets): """Добавки антенн по всем калибровочным точкам: круговая медиана фаз и медиана усиления.""" out = [] for key in ('T', 'R'): Z = np.array([s[key] for s in sets]) # (точки, антенны, частоты) out.append(np.exp(np.median(np.log(np.abs(Z)), axis=0)) * np.exp(1j * circular_median(np.angle(Z)))) return tuple(out) class _Estimator: """Геометрия, полоса и окно одной калибровки.""" def __init__(self, st, tx, rx, freq): self.st, self.tx, self.rx, self.freq = st, tx, rx, freq self.n_tx, self.n_rx = len(tx), len(rx) self.band = (freq >= st.f_min) & (freq <= st.f_max) self.fb = freq[self.band] self.win = np.hanning(self.fb.size) self.ok = self.win > 0.2 # края полосы после деления на окно ненадёжны self.nfft = 1 << int(np.ceil(np.log2(self.fb.size * 8))) self.dpath = np.fft.fftfreq(self.nfft, self.fb[1] - self.fb[0]) * C_LIGHT # ось «лишний путь», м self.gate = np.clip(np.minimum((self.dpath + st.gate_half_m) / st.gate_taper_m, (st.gate_half_m - self.dpath) / st.gate_taper_m) + 0.5, 0, 1) def scene_to_radar(self, q): """(X, Y над землёй, Z вдоль земли) -> оси радара (x, y, z).""" p = np.radians(self.st.pitch_deg); y0 = q[1] - self.st.antenna_height_m return np.array([q[0], y0 * np.cos(p) + q[2] * np.sin(p), -y0 * np.sin(p) + q[2] * np.cos(p)]) def pair_paths(self, q_scene): """Бистатический путь каждой пары до цели (TX, RX). Для шара радиуса a (q_scene — центр) путь короче на 2a·cos(beta/2), beta — угол между направлениями из центра на Tx и на Rx.""" a = self.st.sphere_radius_m pr = self.scene_to_radar(np.asarray(q_scene, float)) vt = self.tx - pr; vr = self.rx - pr dt = np.linalg.norm(vt, axis=1); dr = np.linalg.norm(vr, axis=1) paths = dt[:, None] + dr[None, :] if a: cos_beta = (vt @ vr.T) / (dt[:, None] * dr[None, :]) paths = paths - 2 * a * np.sqrt((1 + cos_beta) / 2) return paths def _unwound(self, s21_band, paths): return s21_band * np.exp(2j * np.pi * self.fb[None, None, :] * paths[:, :, None] / C_LIGHT) def target_echo(self, s21_band, paths): """Эхо цели E (TX, RX, Nb) со снятым путём: окно ±gate_half_m вокруг заданной точки.""" h = np.fft.ifft(self._unwound(s21_band, paths) * self.win, self.nfft, axis=2) return np.fft.fft(h * self.gate, axis=2)[:, :, :self.fb.size] / np.maximum(self.win, 1e-3) def broadband(self, E): """Вклад пары в изображение в точке: сумма эха по надёжной части полосы.""" return E[:, :, self.ok].sum(axis=2) def coarse_range(self, s21, q_nominal): """Грубая привязка по дальности: максимум суммарной огибающей всех пар в пределах ±coarse_range_search_m по пути; заданная дальность Z сдвигается под него.""" q = q_nominal.copy() if not self.st.coarse_range_search_m: return q s = np.mean(s21, axis=0)[:, :, self.band]; sel = np.abs(self.dpath) <= self.st.coarse_range_search_m for _ in range(3): env = np.abs(np.fft.ifft(self._unwound(s, self.pair_paths(q)) * self.win, self.nfft, axis=2)).sum(axis=(0, 1)) q[2] += self.dpath[sel][np.argmax(env[sel])] / 2 return q def refine_target(self, s21, q0): """Положение цели и широкополосные фазы антенн, максимизирующие сфазированность.""" P0 = self.pair_paths(q0) E0 = np.mean([self.target_echo(s[:, :, self.band], P0) for s in s21], axis=0) axes = [('xyz').index(a) for a in self.st.refine_axes] def E_at(v): q = q0.copy(); q[axes] += v dP = self.pair_paths(q) - P0 return E0 * np.exp(2j * np.pi * self.fb[None, None, :] * dP[:, :, None] / C_LIGHT), q def score(v, t, r): if np.any(np.abs(v) > self.st.refine_range_m): return 0.0 return coherence(self.broadband(E_at(v)[0]), t, r) v = np.zeros(len(axes)) if axes: for _ in range(6): t, r = best_phases(self.broadband(E_at(v)[0])) simplex = np.vstack([v] + [v + 0.03 * np.eye(len(axes))[k] for k in range(len(axes))]) v = minimize(lambda w: -score(w, t, r), v, method='Nelder-Mead', options=dict(xatol=5e-4, fatol=1e-7, initial_simplex=simplex)).x return E_at(v) def factorize(self, E): """E_ij(f) ≈ g(f)·T_i(f)·R_j(f) на каждой частоте: фазы — best_phases, амплитуды — МНК по log|E|.""" nb = E.shape[2] T = np.ones((self.n_tx, nb), complex); R = np.ones((self.n_rx, nb), complex) L = np.log(np.abs(E) + 1e-30) for k in range(nb): t, r = best_phases(E[:, :, k]) tt = L[:, :, k].mean(axis=1) - L[:, :, k].mean(); rr = L[:, :, k].mean(axis=0) - L[:, :, k].mean() T[:, k] = np.exp(tt + 1j * t); R[:, k] = np.exp(rr + 1j * r) return T, R def antenna_params(self, Z): """Фаза, продолженная к 0 Гц, задержка (эквивалентный лишний путь) и усиление антенны.""" ph = np.unwrap(np.angle(Z[self.ok])); f = self.fb[self.ok] slope, ph0 = np.polyfit(f, ph, 1) return dict(phase0_deg=float(np.degrees(ph0)), delay_path_m=float(-slope * C_LIGHT / (2 * np.pi)), gain_db=float(20 * np.log10(np.mean(np.abs(Z[self.ok])))), phase_rms_after_linear_deg=float(np.degrees(np.std(ph - (slope * f + ph0))))) def phase_model(self, Z): """Фаза добавки антенны по модели phase_model (подгонка по надёжной части полосы).""" model = self.st.phase_model if model == 'raw': return Z ph = np.unwrap(np.angle(Z)); f = (self.fb - self.fb.mean()) / 1e9 if model == 'const': u = np.exp(1j * np.angle(Z[self.ok])).mean() return np.abs(Z) * np.exp(1j * np.angle(u)) deg = {'linear': 1, 'poly3': 3}.get(model) if deg is None: raise ValueError(f"phase_model должен быть 'const', 'linear', 'poly3' или 'raw', а не {model!r}") return np.abs(Z) * np.exp(1j * np.polyval(np.polyfit(f[self.ok], ph[self.ok], deg), f)) def pair_correction(self, T, R): """Поправка пары C_ij(f) = 1/(T_i·R_j); без apply_amplitude — только фаза.""" Z = T[:, None, :] * R[None, :, :] return np.conj(Z) / np.abs(Z) if not self.st.apply_amplitude else 1.0 / Z def to_full_grid(self, Z_band, params): """На полную сетку частот: внутри надёжной полосы — оценка, вне её — продолжение с наклоном линейной модели фазы антенны, пристыкованное к краю полосы без скачка.""" idx = np.flatnonzero(self.band)[self.ok] ph_in = np.unwrap(np.angle(Z_band[self.ok])); mag_in = np.abs(Z_band[self.ok]) slope = -2 * np.pi * params['delay_path_m'] / C_LIGHT # рад/Гц ph = np.empty(self.freq.size); mag = np.empty(self.freq.size) ph[idx] = ph_in; mag[idx] = mag_in lo, hi = idx[0], idx[-1] ph[:lo] = ph_in[0] + slope * (self.freq[:lo] - self.freq[lo]); mag[:lo] = mag_in[0] ph[hi + 1:] = ph_in[-1] + slope * (self.freq[hi + 1:] - self.freq[hi]); mag[hi + 1:] = mag_in[-1] return mag * np.exp(1j * ph) def compute_calibration(cal_sets, empty_scene_path=None, settings=None): """Калибровка по набору измерений отражателя. Возвращает словарь в том же виде, что load_calibration.""" st = settings or Settings() cal_sets = list(cal_sets) config_path = find_config_profile(cal_sets[0]['path']) tx, rx = load_geometry(config_path, st.input_idx, st.output_idx) freq = None empty = None if empty_scene_path is not None: loaded = [load_s21(d, st.input_idx, st.output_idx) for d in list_snapshots(empty_scene_path)] empty = np.mean([s for s, _ in loaded], axis=0); freq = loaded[0][1] sets = [] for n, cs in enumerate(cal_sets): snaps = list_snapshots(cs['path'], cs.get('snapshots')) data = [] for d in snaps: s21, f = load_s21(d, st.input_idx, st.output_idx) if freq is None: freq = f elif f.shape != freq.shape or not np.allclose(f, freq): raise ValueError(f'Частотная сетка в {d} отличается от остальных калибровочных данных') cfg = find_config_profile(d) if cfg != config_path: tx_d, rx_d = load_geometry(cfg, st.input_idx, st.output_idx) if not (np.allclose(tx_d, tx) and np.allclose(rx_d, rx)): raise ValueError(f'Геометрия в {d} отличается от {config_path}') data.append(s21 - empty if empty is not None else s21) sets.append(dict(n=n, snaps=[d.name for d in snaps], s21=np.array(data), q_nominal=np.array(cs['target_scene_m'], float))) est = _Estimator(st, tx, rx, freq) for s in sets: s['E'], s['q'] = est.refine_target(s['s21'], est.coarse_range(s['s21'], s['q_nominal'])) s['T'], s['R'] = est.factorize(s['E']) T_all, R_all = combine(sets) T_all = np.array([est.phase_model(z) for z in T_all]); R_all = np.array([est.phase_model(z) for z in R_all]) ant = {f'Tx{k}': est.antenna_params(T_all[k]) for k in range(est.n_tx)} ant.update({f'Rx{k}': est.antenna_params(R_all[k]) for k in range(est.n_rx)}) # качество: сфазированность пар в точке цели без поправки и с ней coh_with = lambda s, C: coherence(est.broadband(s['E'] * C)) quality = dict(coherence_before=[], coherence_own=[], coherence_leave_one_out=[], coherence_final=[]) for s in sets: others = [o for o in sets if o is not s] quality['coherence_before'].append(coh_with(s, 1.0)) quality['coherence_own'].append(coh_with(s, est.pair_correction(s['T'], s['R']))) quality['coherence_leave_one_out'].append(coh_with(s, est.pair_correction(*combine(others))) if others else float('nan')) quality['coherence_final'].append(coh_with(s, est.pair_correction(T_all, R_all))) T_full = np.array([est.to_full_grid(T_all[k], ant[f'Tx{k}']) for k in range(est.n_tx)]) R_full = np.array([est.to_full_grid(R_all[k], ant[f'Rx{k}']) for k in range(est.n_rx)]) Z_full = T_full[:, None, :] * R_full[None, :, :] unit = (lambda Z: np.conj(Z) / np.abs(Z)) if not st.apply_amplitude else (lambda Z: 1.0 / Z) valid = np.zeros(freq.size, bool); valid[np.flatnonzero(est.band)[est.ok]] = True meta = dict( created=datetime.datetime.now().isoformat(timespec='seconds'), name=None, convention=dict(apply='S21_corrected[tx, rx, f] = S21[tx, rx, f] * cal_pair[tx, rx, f]', index_order='tx = номер в cal_tx_output_pos (output_pos по возрастанию), rx = cal_rx_input_pos', files='S21 пары в данных: i{input_pos}_o{output_pos}_s21.npy', phase_reference='фазы относительно Tx0 и Rx0; общая фаза/задержка не входит', amplitude='выравнивается' if st.apply_amplitude else 'не меняется (только фаза)', outside_band='вне cal_valid: продолжение линейной моделью фазы антенны'), geometry=dict(config=str(config_path), tx_xyz_m=tx.tolist(), rx_xyz_m=rx.tolist(), antenna_height_m=st.antenna_height_m, pitch_deg=st.pitch_deg), estimation=dict(band_hz=[st.f_min, st.f_max], gate_half_m=st.gate_half_m, refine_axes=list(st.refine_axes), phase_model=st.phase_model, sphere_radius_m=st.sphere_radius_m, empty_scene=str(empty_scene_path)), targets=[dict(path=str(cal_sets[s['n']]['path']), snapshots=s['snaps'], nominal_scene_m=s['q_nominal'].tolist(), refined_scene_m=s['q'].tolist()) for s in sets], antennas=ant, quality=quality) return dict( freq_hz=freq.astype(np.float64), pair=unit(Z_full).astype(np.complex64), # (TX, RX, Nf): S21_corr = S21 * pair tx=unit(T_full).astype(np.complex64), # (TX, Nf): pair = tx[:,None] * rx[None,:] rx=unit(R_full).astype(np.complex64), # (RX, Nf) valid=valid, # (Nf,) True — внутри полосы калибровки tx_output_pos=np.array(sorted(st.output_idx), np.int32), # индекс -> физический выход rx_input_pos=np.array(sorted(st.input_idx), np.int32), # индекс -> физический вход antenna_params=np.array([[np.radians(a['phase0_deg']), a['delay_path_m'], a['gain_db']] for a in ant.values()], np.float64), # (TX+RX, 3): фаза к 0 Гц [рад], путь [м], усиление [дБ] meta=meta, dir=None) # ══════════════════════════════════════════════════════ # ФАЙЛЫ КАЛИБРОВКИ # ══════════════════════════════════════════════════════ _CAL_ARRAYS = ('freq_hz', 'pair', 'tx', 'rx', 'valid', 'tx_output_pos', 'rx_input_pos', 'antenna_params') def save_calibration(cal, cal_dir, overwrite=False): cal_dir = Path(cal_dir) cal_dir.mkdir(parents=True, exist_ok=True) existing = [p.name for p in cal_dir.glob('cal_*')] if existing and not overwrite: raise FileExistsError(f'В {cal_dir} уже есть файлы калибровки ({existing[:3]}...). ' f'Задайте новое имя или overwrite=True.') for k in _CAL_ARRAYS: np.save(cal_dir / f'cal_{k}.npy', cal[k]) cal['meta']['name'] = cal_dir.name; cal['dir'] = cal_dir json.dump(cal['meta'], open(cal_dir / 'cal_meta.json', 'w', encoding='utf-8'), ensure_ascii=False, indent=2) return cal_dir def load_calibration(cal_dir): cal_dir = Path(cal_dir) cal = {k: np.load(cal_dir / f'cal_{k}.npy') for k in _CAL_ARRAYS} cal['meta'] = json.load(open(cal_dir / 'cal_meta.json', encoding='utf-8')) cal['dir'] = cal_dir return cal # ══════════════════════════════════════════════════════ # ПРИМЕНЕНИЕ # ══════════════════════════════════════════════════════ def calibration_on_grid(cal, freq): """Поправка пар (TX, RX, Nf) на частотной сетке freq. Совпадает с сеткой калибровки — берётся как есть; иначе фаза и модуль поправок антенн интерполируются, вне сетки калибровки — линейная модель фазы.""" fc = cal['freq_hz'] if freq.shape == fc.shape and np.allclose(freq, fc): return cal['pair'] def interp(Z, par): ph = np.unwrap(np.angle(Z)); mag = np.abs(Z) out_ph = np.interp(freq, fc, ph); out_mag = np.interp(freq, fc, mag) lin = -(par[0] - 2 * np.pi * freq * par[1] / C_LIGHT) # поправка = −добавка outside = (freq < fc[0]) | (freq > fc[-1]) out_ph[outside] = lin[outside] + np.round((out_ph[outside] - lin[outside]) / (2 * np.pi)) * 2 * np.pi return out_mag * np.exp(1j * out_ph) n_tx = cal['tx'].shape[0] tx = np.array([interp(cal['tx'][k], cal['antenna_params'][k]) for k in range(n_tx)]) rx = np.array([interp(cal['rx'][k], cal['antenna_params'][n_tx + k]) for k in range(cal['rx'].shape[0])]) return (tx[:, None, :] * rx[None, :, :]).astype(cal['pair'].dtype) def apply_calibration(s21, freq, cal): """s21 (TX, RX, Nf) в порядке cal['tx_output_pos'] x cal['rx_input_pos'] -> исправленный массив.""" return s21 * calibration_on_grid(cal, np.asarray(freq)) def apply_to_measurement(meas_dir, cal, overwrite=False): """<измерение>/preprocessed -> <измерение>/postprocessed: S21 пар умножаются на поправку, остальные файлы копируются. Возвращает (папка, снимков, исправлено S21, скопировано файлов).""" meas_dir = Path(meas_dir); pre = meas_dir / 'preprocessed'; post = meas_dir / 'postprocessed' if not pre.is_dir(): raise FileNotFoundError(f'Нет {pre}') # калибровка годится только для той же сборки антенн tx_m, rx_m = load_geometry(find_config_profile(pre), cal['rx_input_pos'].tolist(), cal['tx_output_pos'].tolist()) if not (np.allclose(tx_m, cal['meta']['geometry']['tx_xyz_m']) and np.allclose(rx_m, cal['meta']['geometry']['rx_xyz_m'])): raise ValueError(f'Геометрия антенн в {meas_dir.name} отличается от геометрии калибровки — применять нельзя') if post.exists(): if not overwrite: raise FileExistsError(f'{post} уже существует. overwrite=True, чтобы пересоздать.') if not (post / 'calibration_applied.json').exists(): raise FileExistsError(f'{post} существует, но создан не калибровкой (нет calibration_applied.json) — ' f'удалять не буду, разберитесь вручную.') shutil.rmtree(post) tx_of = {int(p): k for k, p in enumerate(cal['tx_output_pos'])} rx_of = {int(p): k for k, p in enumerate(cal['rx_input_pos'])} pat = re.compile(r'^i(\d+)_o(\d+)_s21\.npy$') n_snap = n_fix = n_copy = 0 grid_cache = {} post.mkdir() for item in sorted(pre.iterdir()): if item.is_file(): shutil.copy2(item, post / item.name); n_copy += 1 continue out = post / item.name; out.mkdir(); n_snap += 1 for f in sorted(item.iterdir()): m = pat.match(f.name) if f.is_dir() or m is None or int(m.group(2)) not in tx_of or int(m.group(1)) not in rx_of: if f.is_dir(): shutil.copytree(f, out / f.name) else: shutil.copy2(f, out / f.name) n_copy += 1 continue s21 = np.load(f); freq = np.load(item / f'i{m.group(1)}_o{m.group(2)}_freq.npy') key = (freq.size, float(freq[0]), float(freq[-1])) if key not in grid_cache: grid_cache[key] = calibration_on_grid(cal, freq) corr = grid_cache[key][tx_of[int(m.group(2))], rx_of[int(m.group(1))]] np.save(out / f.name, (s21 * corr).astype(s21.dtype)); n_fix += 1 json.dump(dict(calibration=cal['meta']['name'], calibration_dir=str(Path(cal['dir']).resolve()), created=datetime.datetime.now().isoformat(timespec='seconds'), source=str(pre), rule='postprocessed i*_o*_s21 = preprocessed i*_o*_s21 * cal_pair[tx(o), rx(i)]; остальные файлы скопированы', frequency_grids=[list(k) for k in grid_cache]), open(post / 'calibration_applied.json', 'w', encoding='utf-8'), ensure_ascii=False, indent=2) return post, n_snap, n_fix, n_copy def main(): if RUN_CALIBRATION: cal = compute_calibration(CAL_SETS, EMPTY_SCENE_PATH, Settings()) save_calibration(cal, CAL_DIR, overwrite=OVERWRITE) q = cal['meta']['quality'] print(f'Калибровка записана в {CAL_DIR}: сфазированность пар на отражателе ' f'{np.mean(q["coherence_before"]):.2f} -> {np.mean(q["coherence_final"]):.2f}') if APPLY_TO_DATA: cal = load_calibration(CAL_DIR) targets = APPLY_TO_MEASUREMENTS if isinstance(targets, (str, Path)): # один путь вместо списка targets = [targets] for meas in targets: post, n_snap, n_fix, _ = apply_to_measurement(meas, cal, OVERWRITE_POSTPROCESSED) print(f'{Path(meas).name}: {n_snap} снимков, исправлено S21 {n_fix} -> {post}') if __name__ == '__main__': main()