"""Потоковый поиск эталонной лестницы Lab064/Lab065 в большом cs8-файле. Утилита не обращается к SDR, не использует пакетный приёмник и не меняет исходные IQ-файлы. Полная запись просматривается блоками с перекрытием. """ from __future__ import annotations import argparse import ctypes from dataclasses import asdict, dataclass import hashlib import json import math import os from pathlib import Path import threading import time from typing import Callable, Iterable import numpy as np from scipy import signal EXPECTED_REFERENCE_SHA256 = ( "fa260719659b2984fe29588238415622cd75c6daefec6cc9130ba1bf851bd3f9" ) ANALYZER_VERSION = "2026.09.28.1" @dataclass(frozen=True) class ReferenceLayout: sample_rate_hz: int = 8_000_000 symbol_rate: int = 20_000 carrier_hz: float = 50_000.0 leading_samples: int = 800_000 segment_samples: int = 800_000 gap_samples: int = 400_000 trailing_samples: int = 800_000 levels_db: tuple[float, ...] = (-18.0, -12.0, -6.0, 0.0) @property def total_samples(self) -> int: return ( self.leading_samples + len(self.levels_db) * self.segment_samples + (len(self.levels_db) - 1) * self.gap_samples + self.trailing_samples ) def active_windows(self) -> tuple[tuple[int, int], ...]: cursor = self.leading_samples windows = [] for index in range(len(self.levels_db)): windows.append((cursor, cursor + self.segment_samples)) cursor += self.segment_samples if index + 1 < len(self.levels_db): cursor += self.gap_samples return tuple(windows) def background_windows(self) -> tuple[tuple[int, int], ...]: width = min(self.gap_samples, self.leading_samples, self.trailing_samples) // 2 cursor = (self.leading_samples - width) // 2 windows = [(cursor, cursor + width)] cursor = self.leading_samples + self.segment_samples for index in range(len(self.levels_db) - 1): start = cursor + (self.gap_samples - width) // 2 windows.append((start, start + width)) cursor += self.gap_samples + self.segment_samples final_start = self.total_samples - self.trailing_samples start = final_start + (self.trailing_samples - width) // 2 windows.append((start, start + width)) return tuple(windows) @dataclass(frozen=True) class SearchCriteria: decimation: int = 32 input_block_samples: int = 4_000_000 anti_alias_passband_hz: float = 75_000.0 anti_alias_stopband_hz: float = 120_000.0 anti_alias_attenuation_db: float = 80.0 feature_period_seconds: float = 0.001 refinement_radius_seconds: float = 0.003 cfo_min_hz: float = -3.0 cfo_max_hz: float = 3.0 cfo_step_hz: float = 0.25 normalized_correlation_minimum: float = 0.05 peak_to_background_minimum_db: float = 6.0 maximum_step_error_db: float = 3.0 l3_matched_to_residual_minimum_db: float = -18.0 l3_spectral_contrast_minimum_db: float = 3.0 histogram_stride: int = 1024 maximum_reported_candidates: int = 5 DEFAULT_LAYOUT = ReferenceLayout() DEFAULT_CRITERIA = SearchCriteria() def current_process_rss_bytes() -> int: if not hasattr(ctypes, "windll"): return 0 class ProcessMemoryCounters(ctypes.Structure): _fields_ = [ ("cb", ctypes.c_ulong), ("page_fault_count", ctypes.c_ulong), ("peak_working_set_size", ctypes.c_size_t), ("working_set_size", ctypes.c_size_t), ("quota_peak_paged_pool_usage", ctypes.c_size_t), ("quota_paged_pool_usage", ctypes.c_size_t), ("quota_peak_non_paged_pool_usage", ctypes.c_size_t), ("quota_non_paged_pool_usage", ctypes.c_size_t), ("pagefile_usage", ctypes.c_size_t), ("peak_pagefile_usage", ctypes.c_size_t), ] counters = ProcessMemoryCounters() counters.cb = ctypes.sizeof(counters) get_current_process = ctypes.windll.kernel32.GetCurrentProcess # type: ignore[attr-defined] get_current_process.restype = ctypes.c_void_p get_process_memory_info = ctypes.windll.psapi.GetProcessMemoryInfo # type: ignore[attr-defined] get_process_memory_info.argtypes = [ ctypes.c_void_p, ctypes.POINTER(ProcessMemoryCounters), ctypes.c_ulong, ] get_process_memory_info.restype = ctypes.c_int succeeded = get_process_memory_info( get_current_process(), ctypes.byref(counters), counters.cb, ) return int(counters.working_set_size) if succeeded else 0 class MemoryPeakSampler: def __init__(self, poll_seconds: float = 0.02) -> None: self.poll_seconds = poll_seconds self.baseline_bytes = current_process_rss_bytes() self.peak_bytes = self.baseline_bytes self._stop = threading.Event() self._thread = threading.Thread(target=self._sample, daemon=True) def _sample(self) -> None: while not self._stop.wait(self.poll_seconds): self.peak_bytes = max(self.peak_bytes, current_process_rss_bytes()) def __enter__(self) -> "MemoryPeakSampler": self._thread.start() return self def __exit__(self, *_exc: object) -> None: self.peak_bytes = max(self.peak_bytes, current_process_rss_bytes()) self._stop.set() self._thread.join(timeout=1.0) @property def additional_peak_bytes(self) -> int: return max(0, self.peak_bytes - self.baseline_bytes) def _write_progress( path: Path, *, phase: str, processed_samples: int, total_samples: int, ) -> None: value = { "schema_version": 1, "phase": phase, "processed_samples": processed_samples, "total_samples": total_samples, "coverage_complete": ( phase == "COMPLETE" and processed_samples == total_samples ), "updated_at_epoch": time.time(), "hardware_accessed": False, "rf_tx": 0, } path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_text( json.dumps(value, separators=(",", ":"), sort_keys=True) + "\n", encoding="utf-8", ) os.replace(temporary, path) def sha256_file( path: Path, progress: Callable[[int, int], None] | None = None, ) -> str: digest = hashlib.sha256() total = path.stat().st_size processed = 0 with path.open("rb") as stream: while block := stream.read(4 * 1024 * 1024): digest.update(block) processed += len(block) if progress is not None: progress(processed, total) return digest.hexdigest() def cs8_sample_count(path: Path) -> int: size = path.stat().st_size if size <= 0 or size % 2: raise ValueError("cs8-файл должен содержать целое ненулевое число пар I/Q") return size // 2 def design_antialias_filter( layout: ReferenceLayout, criteria: SearchCriteria, ) -> np.ndarray: """Build the fixed linear-phase FIR used before every decimation.""" if criteria.decimation <= 0: raise ValueError("decimation must be positive") if criteria.decimation == 1: return np.ones(1, dtype=np.float64) nyquist = layout.sample_rate_hz / 2 decimated_nyquist = layout.sample_rate_hz / (2 * criteria.decimation) if not ( 0 < criteria.anti_alias_passband_hz < criteria.anti_alias_stopband_hz < decimated_nyquist ): raise ValueError("anti-alias passband and stopband are invalid") transition_width = ( criteria.anti_alias_stopband_hz - criteria.anti_alias_passband_hz ) / nyquist taps_count, beta = signal.kaiserord( criteria.anti_alias_attenuation_db, transition_width, ) taps_count = max(3, taps_count | 1) cutoff_hz = ( criteria.anti_alias_passband_hz + criteria.anti_alias_stopband_hz ) / 2 return signal.firwin( taps_count, cutoff_hz, window=("kaiser", beta), fs=layout.sample_rate_hz, ).astype(np.float64) def _read_cs8_complex(path: Path, start: int, stop: int) -> np.ndarray: """Read one bounded source interval; out-of-file samples are zero.""" if start >= stop: raise ValueError("invalid source interval") total = cs8_sample_count(path) values = np.zeros(stop - start, dtype=np.complex64) clipped_start = max(0, start) clipped_stop = min(total, stop) if clipped_start >= clipped_stop: return values raw = np.memmap(path, dtype=np.int8, mode="r") interleaved = np.asarray( raw[2 * clipped_start:2 * clipped_stop], dtype=np.float32, ) destination = slice(clipped_start - start, clipped_stop - start) values[destination] = interleaved[0::2] + 1j * interleaved[1::2] return values def read_filtered_decimated_window( path: Path, start: int, stop: int, layout: ReferenceLayout, criteria: SearchCriteria, taps: np.ndarray, ) -> np.ndarray: """Read a decimated window with FIR overlap and delay correction. Output index ``k`` is centred on input sample ``k * decimation``. The function reads half the FIR on both sides, so adjacent blocks have exactly the overlap required to match one full-file convolution. Zeros are used only outside the physical file. """ total = cs8_sample_count(path) total_decimated = math.ceil(total / criteria.decimation) if not 0 <= start < stop <= total_decimated: raise ValueError("invalid filtered-decimated window") delay = (taps.size - 1) // 2 first_center = start * criteria.decimation last_center = (stop - 1) * criteria.decimation source_start = first_center - delay source_stop = last_center + delay + 1 values = _read_cs8_complex(path, source_start, source_stop) if taps.size == 1: return values[::criteria.decimation].astype(np.complex64, copy=False) filtered = signal.fftconvolve(values, taps, mode="valid") result = filtered[::criteria.decimation] if result.size != stop - start: raise RuntimeError("filtered decimator produced an unexpected length") return result.astype(np.complex64, copy=False) def iter_filtered_decimated( path: Path, layout: ReferenceLayout, criteria: SearchCriteria, taps: np.ndarray, ) -> Iterable[tuple[int, np.ndarray]]: """Yield the full file in bounded, boundary-correct decimated blocks.""" if criteria.input_block_samples <= 0: raise ValueError("input block size must be positive") total = cs8_sample_count(path) total_decimated = math.ceil(total / criteria.decimation) block = max(1, criteria.input_block_samples // criteria.decimation) for start in range(0, total_decimated, block): stop = min(total_decimated, start + block) yield start, read_filtered_decimated_window( path, start, stop, layout, criteria, taps, ) def load_reference( path: Path, layout: ReferenceLayout, criteria: SearchCriteria, *, expected_sha256: str | None, ) -> tuple[np.ndarray, str]: if cs8_sample_count(path) != layout.total_samples: raise ValueError("размер эталона не соответствует фиксированной структуре") digest = sha256_file(path) if expected_sha256 is not None and digest != expected_sha256: raise ValueError("SHA256 эталона не совпадает с разрешённым значением") stop = math.ceil(layout.total_samples / criteria.decimation) taps = design_antialias_filter(layout, criteria) reference = read_filtered_decimated_window( path, 0, stop, layout, criteria, taps, ) reference -= np.mean(reference, dtype=np.complex128) return reference, digest def _frequency_grid(criteria: SearchCriteria) -> np.ndarray: count = round( (criteria.cfo_max_hz - criteria.cfo_min_hz) / criteria.cfo_step_hz ) return criteria.cfo_min_hz + criteria.cfo_step_hz * np.arange(count + 1) def _histogram_quantile(histogram: np.ndarray, quantile: float) -> float: total = int(histogram.sum()) if total == 0: return 0.0 target = math.ceil(total * quantile) index = int(np.searchsorted(np.cumsum(histogram), target, side="left")) return min(1.0, (index + 0.5) / histogram.size) def _deduplicate_candidates( candidates: Iterable[dict[str, float | int]], minimum_separation: int, limit: int, ) -> list[dict[str, float | int]]: selected: list[dict[str, float | int]] = [] for candidate in sorted(candidates, key=lambda row: float(row["score"]), reverse=True): start = int(candidate["start_decimated_sample"]) if any( abs(start - int(existing["start_decimated_sample"])) < minimum_separation for existing in selected ): continue selected.append(candidate) if len(selected) == limit: break return selected def _power_features(values: np.ndarray, stride: int) -> np.ndarray: complete = values.size // stride if complete == 0: return np.empty(0, dtype=np.float64) reshaped = values[:complete * stride].reshape(complete, stride) return np.mean(np.abs(reshaped) ** 2, axis=1, dtype=np.float64) def _normalized_sliding_scores( values: np.ndarray, template: np.ndarray, ) -> np.ndarray: centred = template.astype(np.float64) - float(np.mean(template)) template_energy = float(np.vdot(centred, centred).real) count = centred.size prefix = np.concatenate(([0.0], np.cumsum(values, dtype=np.float64))) prefix_power = np.concatenate( ([0.0], np.cumsum(values * values, dtype=np.float64)) ) window_sum = prefix[count:] - prefix[:-count] window_power = prefix_power[count:] - prefix_power[:-count] ac_power = window_power - (window_sum * window_sum) / count denominator = np.sqrt(np.maximum(template_energy * ac_power, 1e-30)) correlation = signal.fftconvolve(values, centred[::-1], mode="valid") return np.abs(correlation) / denominator def _refine_candidate( capture_path: Path, reference: np.ndarray, approximate_start: int, layout: ReferenceLayout, criteria: SearchCriteria, taps: np.ndarray, ) -> dict[str, float | int]: decimated_rate = layout.sample_rate_hz / criteria.decimation radius = max(1, round(criteria.refinement_radius_seconds * decimated_rate)) total_decimated = math.ceil( cs8_sample_count(capture_path) / criteria.decimation ) first = max(0, approximate_start - radius) last = min(total_decimated - reference.size, approximate_start + radius) received = read_filtered_decimated_window( capture_path, first, last + reference.size, layout, criteria, taps, ) count = reference.size prefix_power = np.concatenate( ([0.0], np.cumsum(np.abs(received) ** 2, dtype=np.float64)) ) prefix_sum = np.concatenate( ([0.0 + 0.0j], np.cumsum(received, dtype=np.complex128)) ) window_power = prefix_power[count:] - prefix_power[:-count] window_sum = prefix_sum[count:] - prefix_sum[:-count] window_ac_power = window_power - (np.abs(window_sum) ** 2) / count reference_energy = float(np.vdot(reference, reference).real) denominator = np.sqrt(np.maximum(reference_energy * window_ac_power, 1e-30)) time_axis = np.arange(reference.size, dtype=np.float64) / decimated_rate best_score = -1.0 best_index = 0 best_cfo = 0.0 for cfo_hz in _frequency_grid(criteria): rotated = reference * np.exp(2j * np.pi * cfo_hz * time_axis) correlation = signal.fftconvolve( received, np.conj(rotated[::-1]), mode="valid", ) scores = np.abs(correlation) / denominator index = int(np.argmax(scores)) score = float(scores[index]) if score > best_score: best_score = score best_index = index best_cfo = float(cfo_hz) start_decimated = first + best_index return { "score": best_score, "start_decimated_sample": start_decimated, "start_sample": start_decimated * criteria.decimation, "start_seconds": ( start_decimated * criteria.decimation / layout.sample_rate_hz ), "cfo_hz": best_cfo, } def locate_reference_streaming( capture_path: Path, reference: np.ndarray, layout: ReferenceLayout, criteria: SearchCriteria, progress: Callable[[int, int], None] | None = None, ) -> dict[str, object]: """Scan every source sample through a bounded anti-aliased feature stream.""" total_samples = cs8_sample_count(capture_path) total_decimated = math.ceil(total_samples / criteria.decimation) if total_decimated < reference.size: raise ValueError("RX короче эталонной лестницы") decimated_rate = layout.sample_rate_hz / criteria.decimation feature_stride = round(criteria.feature_period_seconds * decimated_rate) if feature_stride <= 0 or criteria.histogram_stride <= 0: raise ValueError("feature period and histogram stride must be positive") taps = design_antialias_filter(layout, criteria) feature_parts: list[np.ndarray] = [] pending = np.empty(0, dtype=np.complex64) maximum_loaded = 0 streamed_decimated = 0 for block_start, block in iter_filtered_decimated( capture_path, layout, criteria, taps, ): if block_start != streamed_decimated: raise RuntimeError("decimated stream is not contiguous") streamed_decimated += block.size if progress is not None: progress( min(total_samples, streamed_decimated * criteria.decimation), total_samples, ) combined = np.concatenate((pending, block)) if pending.size else block complete = combined.size // feature_stride used = complete * feature_stride if used: feature_parts.append(_power_features(combined[:used], feature_stride)) pending = combined[used:].copy() maximum_loaded = max(maximum_loaded, int(combined.size)) if pending.size: feature_parts.append( np.array([np.mean(np.abs(pending) ** 2, dtype=np.float64)]) ) features = np.concatenate(feature_parts) reference_features = _power_features(reference, feature_stride) if features.size < reference_features.size: raise ValueError("RX feature stream is shorter than the reference") scores = _normalized_sliding_scores(features, reference_features) sampled = np.clip(scores[::criteria.histogram_stride], 0.0, 1.0) histogram = np.histogram(sampled, bins=2000, range=(0.0, 1.0))[0] background_p99 = _histogram_quantile(histogram, 0.99) candidate_count = min(max(20, criteria.maximum_reported_candidates * 8), scores.size) indexes = np.argpartition(scores, -candidate_count)[-candidate_count:] coarse = [ { "score": float(scores[index]), "start_decimated_sample": int(index * feature_stride), "start_feature": int(index), } for index in indexes ] coarse = _deduplicate_candidates( coarse, max(1, reference.size // 3), criteria.maximum_reported_candidates, ) refined = [] for row in coarse: candidate = _refine_candidate( capture_path, reference, int(row["start_decimated_sample"]), layout, criteria, taps, ) temporal_score = float(row["score"]) candidate["temporal_power_score"] = temporal_score candidate["score_to_background_p99_db"] = ( 20.0 * math.log10(temporal_score / background_p99) if temporal_score > 0 and background_p99 > 0 else None ) refined.append(candidate) independent = _deduplicate_candidates( refined, max(1, reference.size // 3), criteria.maximum_reported_candidates, ) return { "coverage_start_sample": 0, "coverage_stop_sample": total_samples, "coverage_seconds": total_samples / layout.sample_rate_hz, "coverage_complete": streamed_decimated == total_decimated, "source_samples_processed": total_samples, "tested_feature_start_positions": int(scores.size), "feature_stride_decimated_samples": feature_stride, "feature_stride_source_samples": feature_stride * criteria.decimation, "frequency_grid_hz": [ float(value) for value in _frequency_grid(criteria) ], "background_score_p99": background_p99, "candidates": independent, "maximum_loaded_decimated_samples": maximum_loaded, "maximum_loaded_iq_bytes": maximum_loaded * np.dtype(np.complex64).itemsize, "anti_alias_filter": { "kind": "linear_phase_fir_kaiser", "tap_count": int(taps.size), "passband_hz": criteria.anti_alias_passband_hz, "stopband_hz": criteria.anti_alias_stopband_hz, "design_attenuation_db": criteria.anti_alias_attenuation_db, "group_delay_source_samples": (int(taps.size) - 1) // 2, "position_delay_correction_source_samples": ( -((int(taps.size) - 1) // 2) ), "boundary_handling": "symmetric_overlap_with_zero_extension_at_file_edges", }, } def _read_candidate( capture_path: Path, start_decimated: int, count: int, layout: ReferenceLayout, criteria: SearchCriteria, ) -> np.ndarray: taps = design_antialias_filter(layout, criteria) return read_filtered_decimated_window( capture_path, start_decimated, start_decimated + count, layout, criteria, taps, ) def _decimated_window(window: tuple[int, int], decimation: int) -> tuple[int, int]: return round(window[0] / decimation), round(window[1] / decimation) def _band_power(values: np.ndarray, sample_rate_hz: float, low: float, high: float) -> float: nperseg = min(4096, values.size) frequencies, density = signal.welch( values, fs=sample_rate_hz, nperseg=nperseg, noverlap=nperseg // 2, return_onesided=False, scaling="density", ) mask = (frequencies >= low) & (frequencies <= high) if not np.any(mask): return 0.0 return float(np.trapezoid(density[mask], frequencies[mask])) def verify_candidate( capture_path: Path, reference: np.ndarray, candidate: dict[str, float | int], layout: ReferenceLayout, criteria: SearchCriteria, ) -> dict[str, object]: start_decimated = int(candidate["start_decimated_sample"]) received = _read_candidate( capture_path, start_decimated, reference.size, layout, criteria, ) decimated_rate = layout.sample_rate_hz / criteria.decimation cfo_hz = float(candidate["cfo_hz"]) time_axis = np.arange(reference.size, dtype=np.float64) / decimated_rate rotated = reference * np.exp(2j * np.pi * cfo_hz * time_axis) segment_decimated = round(layout.segment_samples / criteria.decimation) trim = min( max(1, round(21 * layout.sample_rate_hz / layout.symbol_rate / criteria.decimation)), max(1, segment_decimated // 10), ) rows = [] for index, window in enumerate(layout.active_windows()): start, stop = _decimated_window(window, criteria.decimation) start += trim stop -= trim expected = rotated[start:stop] observed = received[start:stop] expected = expected - np.mean(expected, dtype=np.complex128) observed = observed - np.mean(observed, dtype=np.complex128) expected_energy = float(np.vdot(expected, expected).real) observed_energy = float(np.vdot(observed, observed).real) gain = np.vdot(expected, observed) / max(expected_energy, 1e-30) matched = gain * expected residual = observed - matched matched_power = float(np.mean(np.abs(matched) ** 2)) residual_power = float(np.mean(np.abs(residual) ** 2)) coherence = float( abs(np.vdot(expected, observed)) / math.sqrt(max(expected_energy * observed_energy, 1e-30)) ) rows.append( { "label": f"L{index}", "expected_level_db": layout.levels_db[index], "matched_power": matched_power, "residual_power": residual_power, "matched_to_residual_db": ( 10.0 * math.log10(matched_power / residual_power) if matched_power > 0 and residual_power > 0 else None ), "normalized_coherence": coherence, "gain_abs": float(abs(gain)), } ) step_rows = [] for lower, upper in zip(rows, rows[1:]): expected_step = float(upper["expected_level_db"]) - float(lower["expected_level_db"]) lower_power = float(lower["matched_power"]) upper_power = float(upper["matched_power"]) observed_step = ( 10.0 * math.log10(upper_power / lower_power) if lower_power > 0 and upper_power > 0 else None ) step_rows.append( { "comparison": f"{upper['label']}/{lower['label']}", "expected_step_db": expected_step, "observed_step_db": observed_step, "error_db": observed_step - expected_step if observed_step is not None else None, } ) finite_errors = [ abs(float(row["error_db"])) for row in step_rows if row["error_db"] is not None ] maximum_step_error = max(finite_errors) if len(finite_errors) == len(step_rows) else None l3_start, l3_stop = _decimated_window(layout.active_windows()[-1], criteria.decimation) background_start, background_stop = _decimated_window( layout.background_windows()[-2], criteria.decimation ) band_half_width = 0.675 * layout.symbol_rate low = layout.carrier_hz - band_half_width high = layout.carrier_hz + band_half_width l3_band_power = _band_power(received[l3_start:l3_stop], decimated_rate, low, high) background_band_power = _band_power( received[background_start:background_stop], decimated_rate, low, high ) spectral_contrast = ( 10.0 * math.log10(l3_band_power / background_band_power) if l3_band_power > 0 and background_band_power > 0 else None ) l3_ratio = rows[-1]["matched_to_residual_db"] matched_powers = [float(row["matched_power"]) for row in rows] monotonic = all(upper > lower for lower, upper in zip(matched_powers, matched_powers[1:])) step_pass = ( maximum_step_error is not None and maximum_step_error <= criteria.maximum_step_error_db and monotonic ) coherence_pass = ( l3_ratio is not None and float(l3_ratio) >= criteria.l3_matched_to_residual_minimum_db ) spectral_pass = ( spectral_contrast is not None and spectral_contrast >= criteria.l3_spectral_contrast_minimum_db ) return { "segments": rows, "steps": step_rows, "maximum_absolute_step_error_db": maximum_step_error, "monotonic_matched_power": monotonic, "step_pattern_pass": step_pass, "l3_matched_to_residual_pass": coherence_pass, "l3_band_power": l3_band_power, "background_band_power": background_band_power, "l3_spectral_contrast_db": spectral_contrast, "spectral_contrast_pass": spectral_pass, "qualified": step_pass and coherence_pass and spectral_pass, } def classify_search( search: dict[str, object], verifications: list[dict[str, object]], criteria: SearchCriteria, ) -> tuple[str, list[str]]: if search.get("coverage_complete") is not True: return "ANALYSIS_INCOMPLETE", ["FULL_CAPTURE_COVERAGE_NOT_CONFIRMED"] candidates = list(search["candidates"]) if not candidates: return "NOT_DETECTED", ["NO_CANDIDATES"] peak = candidates[0] score = float(peak["score"]) contrast = peak.get("score_to_background_p99_db") correlation_pass = score >= criteria.normalized_correlation_minimum contrast_pass = ( contrast is not None and float(contrast) >= criteria.peak_to_background_minimum_db ) qualified = [row for row in verifications if row.get("qualified") is True] reasons = [] if not correlation_pass: reasons.append("NORMALIZED_CORRELATION_BELOW_THRESHOLD") if not contrast_pass: reasons.append("PEAK_TO_BACKGROUND_CONTRAST_BELOW_THRESHOLD") if not qualified: reasons.append("STEP_OR_SPECTRAL_CONFIRMATION_FAILED") if correlation_pass and contrast_pass and len(qualified) == 1: return "DETECTED", [] if (correlation_pass and contrast_pass) or qualified: if len(qualified) > 1: reasons.append("MULTIPLE_QUALIFIED_CANDIDATES") return "AMBIGUOUS", reasons return "NOT_DETECTED", reasons def analyze( reference_path: Path, capture_path: Path, report_path: Path, *, expected_capture_sha256: str, layout: ReferenceLayout = DEFAULT_LAYOUT, criteria: SearchCriteria = DEFAULT_CRITERIA, expected_reference_sha256: str = EXPECTED_REFERENCE_SHA256, progress_path: Path | None = None, ) -> dict[str, object]: if report_path.exists(): raise FileExistsError(f"отчёт уже существует: {report_path}") total_samples = cs8_sample_count(capture_path) if progress_path is not None: _write_progress( progress_path, phase="VERIFYING_REFERENCE", processed_samples=0, total_samples=total_samples, ) reference, reference_sha256 = load_reference( reference_path, layout, criteria, expected_sha256=expected_reference_sha256, ) capture_progress: Callable[[int, int], None] | None = None if progress_path is not None: capture_progress = lambda processed, total: _write_progress( progress_path, phase="VERIFYING_RX_SHA256", processed_samples=processed // 2, total_samples=total // 2, ) capture_sha256 = sha256_file(capture_path, progress=capture_progress) if capture_sha256 != expected_capture_sha256: raise ValueError("SHA256 RX-файла не совпадает с ожидаемым значением") analysis_started = time.perf_counter() with MemoryPeakSampler() as memory: search_progress: Callable[[int, int], None] | None = None if progress_path is not None: search_progress = lambda processed, total: _write_progress( progress_path, phase="SEARCHING", processed_samples=processed, total_samples=total, ) search = locate_reference_streaming( capture_path, reference, layout, criteria, progress=search_progress, ) verifications = [] for candidate in search["candidates"]: verification = verify_candidate( capture_path, reference, candidate, layout, criteria, ) verifications.append({"candidate": candidate, **verification}) verdict, reasons = classify_search(search, verifications, criteria) analysis_seconds = time.perf_counter() - analysis_started result = { "schema_version": 1, "analyzer_version": ANALYZER_VERSION, "mode": "OFFLINE_REFERENCE_STAIRCASE_SEARCH", "reference_profile": "LAB065_REFERENCE_STAIRCASE", "reference_path": str(reference_path), "reference_sha256": reference_sha256, "capture_path": str(capture_path), "capture_sha256": capture_sha256, "capture_bytes": capture_path.stat().st_size, "capture_complex_samples": cs8_sample_count(capture_path), "layout": asdict(layout), "criteria_fixed_before_capture_analysis": asdict(criteria), "search": search, "verifications": verifications, "verdict": verdict, "verdict_reasons": reasons, "peak_process_memory_bytes": memory.peak_bytes, "peak_additional_memory_bytes": memory.additional_peak_bytes, "analysis_seconds": analysis_seconds, "hardware_accessed": False, "rf_tx": 0, } report_path.parent.mkdir(parents=True, exist_ok=True) serialized = json.dumps(result, ensure_ascii=False, indent=2, allow_nan=False) + "\n" with report_path.open("x", encoding="utf-8") as stream: stream.write(serialized) if progress_path is not None: _write_progress( progress_path, phase="COMPLETE", processed_samples=total_samples, total_samples=total_samples, ) return result def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--reference", required=True, type=Path) parser.add_argument("--capture", required=True, type=Path) parser.add_argument("--capture-sha256", required=True) parser.add_argument("--report", required=True, type=Path) parser.add_argument("--progress", type=Path) arguments = parser.parse_args() result = analyze( arguments.reference, arguments.capture, arguments.report, expected_capture_sha256=arguments.capture_sha256, progress_path=arguments.progress, ) print(f"REFERENCE STAIRCASE = {result['verdict']}; RF TX = 0") if __name__ == "__main__": main()