Lab051: accelerate 500 ksymbol/s streaming receiver

experiments/lab051_receiver_performance.py

@@ -0,0 +1,320 @@
+"""Lab051: измеренный блочный RX и возраст JPEG в шестисекундном потоке.
+
+Источник синтетический, из реального фото. RF не выполняется. Виртуальное
+время работника использует фактическое время каждого push/сборки/JPEG decode.
+Это офлайн-воспроизведение; отставание явно означает требуемый внешний IQ backlog.
+"""
+
+from __future__ import annotations
+
+import argparse
+from collections import Counter
+from dataclasses import asdict
+import hashlib
+from io import BytesIO
+import json
+import math
+from pathlib import Path
+import statistics
+import sys
+import time
+
+if __package__ in (None, ""):
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+import numpy as np
+from PIL import Image
+
+from experiments import lab050_video_stream as video
+from experiments import lab051_streaming_receiver as fast
+from experiments.lab051_receiver_profile import OUTPUT, ROOT, StageProfiler, array_observation, process_memory
+from protocol.video_frame_scheduler import FramePolicy, schedule_video_frames
+
+
+def write_result(name: str, result: dict) -> None:
+    OUTPUT.mkdir(parents=True, exist_ok=True)
+    (OUTPUT / name).write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+
+
+class ImageSink:
+    """Только текущий объект; никаких ожидаемых ID, размеров JPEG или TX bytes."""
+
+    def __init__(self):
+        self.object_id = -1
+        self.last_completed_id = -1
+        self.assembler = None
+        self.first_rf_sample = None
+        self.first_rx_time = None
+        self.retained_payload_peak = 0
+        self.assembly_seconds = self.jpeg_decode_seconds = 0.0
+        self.rejected = Counter()
+
+    def offer(self, packet, received_time: float) -> dict | None:
+        started = time.perf_counter()
+        fragment = video.transport.recover_fragment_from_payload(packet.payload)
+        if fragment is None or fragment.original_size > 65536 or fragment.fragment_count > 269:
+            self.rejected["invalid_fragment"] += 1
+            self.assembly_seconds += time.perf_counter() - started
+            return None
+        if fragment.image_id < self.object_id or fragment.image_id <= self.last_completed_id:
+            self.rejected["older_object"] += 1
+            self.assembly_seconds += time.perf_counter() - started
+            return None
+        if fragment.image_id > self.object_id:
+            self.object_id = fragment.image_id
+            self.assembler = video.transport.ObjectAssembler()
+            self.first_rf_sample = packet.start_sample
+            self.first_rx_time = received_time
+        self.assembler.offer(fragment)
+        self.retained_payload_peak = max(self.retained_payload_peak, fragment.original_size)
+        jpeg = self.assembler.recovered_object()
+        self.assembly_seconds += time.perf_counter() - started
+        if jpeg is None:
+            return None
+        decode_started = time.perf_counter()
+        try:
+            with Image.open(BytesIO(jpeg)) as image:
+                image.load()
+                dimensions = image.size
+        except (OSError, ValueError):
+            self.rejected["jpeg_decode"] += 1
+            return None
+        finally:
+            self.jpeg_decode_seconds += time.perf_counter() - decode_started
+        self.last_completed_id = self.object_id
+        self.assembler = None
+        return {"image_id": self.object_id, "jpeg": jpeg, "dimensions": dimensions,
+                "first_rf_sample": self.first_rf_sample, "last_rf_sample": packet.end_sample,
+                "first_rx_time": self.first_rx_time}
+
+
+def run_stream(sequence: list[video.EncodedImage]) -> dict:
+    """Вход RX — только последовательные IQ-блоки; TX-диагностика после JPEG."""
+    mode = video.phy.PhyMode(500000)
+    receiver = fast.StreamingReceiver()
+    sink = ImageSink()
+    random_generator = np.random.default_rng(50050)
+    frames = video.make_frame_groups(sequence, 5)
+    packet_time = video.FRAME_INTERVALS / mode.symbol_rate
+    bitrate = (256 + video.HEADER_SIZE) * 8 / packet_time
+    # TX использует известную нижнюю задержку радио/буфера, без предсказания
+    # будущего CPU и без знания RX verdict. Перегрузка RX измеряется ниже.
+    buffer_delay = (receiver.lookahead + 2*receiver.core_samples) / mode.sample_rate_hz
+    schedule = schedule_video_frames(frames, (), FramePolicy.LATEST_ONLY, bitrate,
+                                     max_video_age_seconds=.2, completion_latency_seconds=buffer_delay)
+    starts = {}
+    for sent in schedule.transmitted:
+        starts.setdefault(sent.item.composite_frame_id, sent.start_seconds)
+    horizon = len(sequence) / 5
+    encoded_lookup = {image.image_id: image for image in sequence}
+    capture_lookup = {frame.composite_frame_id: frame.generation_time_seconds for frame in frames}
+    pending = np.empty(0, np.complex64)
+    total_samples = tx_packets = 0
+    worker_time = rx_processing = max_lag = 0.0
+    completed = []
+    block_metrics = []
+    rss_samples = []
+
+    def consume_block(block, *, final=False):
+        nonlocal total_samples, worker_time, rx_processing, max_lag
+        total_samples += block.size
+        arrival = total_samples / mode.sample_rate_hz
+        worker_start = max(worker_time, arrival)
+        started = time.perf_counter()
+        packets = receiver.finish() if final else receiver.push_iq_block(block)
+        cpu = time.perf_counter() - started
+        rx_processing += cpu
+        worker_time = worker_start + cpu
+        for packet in packets:
+            started = time.perf_counter()
+            item = sink.offer(packet, worker_time)
+            cost = time.perf_counter() - started
+            worker_time += cost
+            if item is not None:
+                # Приём/сборка/JPEG уже завершены. TX используется только теперь.
+                jpeg = item.pop("jpeg")
+                image_id = item["image_id"]
+                capture = capture_lookup.get(image_id)
+                age = None if capture is None else worker_time - capture
+                completed.append({**item, "decode_complete_timestamp": worker_time,
+                                  "capture_timestamp": capture, "image_age_seconds": age,
+                                  "fresh": age is not None and 0 <= age <= .2,
+                                  "sha256": hashlib.sha256(jpeg).hexdigest(),
+                                  "sha_match": image_id in encoded_lookup and jpeg == encoded_lookup[image_id].jpeg})
+        lag = max(0., worker_time - arrival)
+        max_lag = max(max_lag, lag)
+        block_metrics.append({"samples_received": total_samples, "arrival_time": arrival,
+                              "worker_start": worker_start, "worker_end": worker_time,
+                              "rx_cpu_seconds": cpu, "packets": len(packets), "lag_seconds": lag})
+        if len(block_metrics) % 64 == 0:
+            rss_samples.append({"samples": total_samples, **process_memory()})
+
+    def push(values):
+        nonlocal pending
+        offset = 0
+        if pending.size:
+            needed = receiver.block_samples - pending.size
+            joined = np.concatenate((pending, values[:needed]))
+            offset = min(needed, values.size)
+            if joined.size < receiver.block_samples:
+                pending = joined
+                return
+            consume_block(joined)
+            pending = np.empty(0, np.complex64)
+        while offset + receiver.block_samples <= values.size:
+            consume_block(values[offset:offset + receiver.block_samples])
+            offset += receiver.block_samples
+        pending = values[offset:].copy()
+
+    def background(count):
+        while count:
+            amount = min(count, receiver.block_samples)
+            values = .2 * (random_generator.normal(size=amount) + 1j*random_generator.normal(size=amount))
+            push(values.astype(np.complex64))
+            count -= amount
+
+    for image in sequence:
+        if image.image_id not in starts:
+            continue
+        target = math.ceil(starts[image.image_id] * mode.sample_rate_hz)
+        background(max(0, target - total_samples - pending.size))
+        for frame in video.transport.build_object_frames(image.jpeg, image.image_id):
+            raw = video.phy.build_parameterized_tx_cs8(frame.plan, mode)
+            components = np.frombuffer(raw, np.int8).astype(np.float32)
+            values = components[0::2] + 1j*components[1::2]
+            indexes = np.arange(values.size)
+            values = values * np.exp(1j*(.73 + 2*np.pi*10*indexes/mode.sample_rate_hz)) * (-1 if tx_packets % 2 else 1)
+            energy = float(np.sum(abs(values)**2)) / frame.plan.frame_symbol_count
+            sigma = math.sqrt(energy / (2*10**(.7)))
+            values += sigma*(random_generator.normal(size=values.size) + 1j*random_generator.normal(size=values.size))
+            push(values.astype(np.complex64))
+            tx_packets += 1
+    background(max(0, math.ceil(horizon * mode.sample_rate_hz) + receiver.lookahead + receiver.core_samples
+                   - total_samples - pending.size))
+    if pending.size:
+        consume_block(pending)
+    consume_block(np.empty(0,np.complex64), final=True)
+    fresh = [row for row in completed if row["fresh"]]
+    displayed_in_window = [row for row in fresh if row["decode_complete_timestamp"] <= horizon]
+    decoded_in_window = [row for row in completed if row["decode_complete_timestamp"] <= horizon]
+    ages = [row["image_age_seconds"] for row in completed if row["image_age_seconds"] is not None]
+    fresh_ages = [row["image_age_seconds"] for row in fresh]
+    processing = rx_processing + sink.assembly_seconds + sink.jpeg_decode_seconds
+    return {
+        "scope": "continuous synthetic IQ; measured per-block CPU event replay; camera/display latency UNKNOWN",
+        "symbol_rate":500000,"samples_per_symbol":16,"input_fps":5,"input_frames":len(sequence),
+        "observation_seconds":horizon,"iq_seconds":total_samples/mode.sample_rate_hz,
+        "tx_images":len(starts),"tx_physical_packets":tx_packets,
+        "dropped_before_tx":len(schedule.dropped_frames),"maximum_waiting_frames":schedule.maximum_waiting_frames,
+        "rx_physical_packets":receiver.packet_count,"received_images":len(completed),
+        "object_jpeg_integrity":len(completed)==len(starts) and all(row["sha_match"] for row in completed),
+        "dropped_stale_at_display":len(completed)-len(fresh),"displayed":len(fresh),"fresh_displayed":len(fresh),
+        "raw_received_fps":len(decoded_in_window)/horizon,
+        "raw_display_fps":len(displayed_in_window)/horizon,"fresh_display_fps":len(displayed_in_window)/horizon,
+        "drain_fresh_displayed":len(fresh)-len(displayed_in_window),
+        "image_age_min_seconds":min(ages) if ages else None,
+        "image_age_median_seconds":statistics.median(ages) if ages else None,
+        "image_age_max_seconds":max(ages) if ages else None,
+        "fresh_image_age_seconds":fresh_ages,"max_video_age_seconds":.2,
+        "processing_seconds":processing,"rx_processing_seconds":rx_processing,
+        "assembly_seconds":sink.assembly_seconds,"jpeg_decode_seconds":sink.jpeg_decode_seconds,
+        "processing_per_rf_second":processing/(total_samples/mode.sample_rate_hz),
+        "physical_packet_processing_seconds":processing/tx_packets if tx_packets else None,
+        "maximum_cpu_lag_seconds":max_lag,
+        "implied_external_iq_backlog_bytes":math.ceil(max_lag*mode.sample_rate_hz)*8,
+        "live_external_queue_implemented":False,
+        "retained_iq_peak_bytes":receiver.peak_buffer_samples*8,
+        "working_arrays_upper_estimate_bytes":receiver.memory_estimate_bytes + 20*receiver.buffer_limit_samples*8,
+        "object_payload_peak_bytes":sink.retained_payload_peak,"process_memory":process_memory(),
+        "rss_samples":rss_samples,"candidate_count":receiver.candidate_count,
+        "rejected_reasons":dict(receiver.rejected_reasons),"front_statistics":dict(receiver.front_statistics),
+        "decimated_samples_reused":receiver.decimator.reused_samples,
+        "actual_decimator_input_samples":receiver.decimator.filtered_input_samples,
+        "images":completed,"blocks":block_metrics,"rf_tx":0,
+    }
+
+
+def run_memory_test(block_count: int = 4096) -> dict:
+    """33,55 с без истории IQ/пакетов; нулевой фон отделён от noisy IQ опыта."""
+    receiver = fast.StreamingReceiver()
+    values = np.zeros(receiver.block_samples,np.complex64)
+    checkpoints=[]
+    for index in range(block_count):
+        assert receiver.push_iq_block(values) == []
+        assert receiver.buffer.size <= receiver.buffer_limit_samples
+        if (index+1) % 512 == 0:
+            checkpoints.append({"block":index+1,"iq_bytes":receiver.buffer.nbytes,
+                                "decimated_bytes":receiver.decimator.values.nbytes,**process_memory()})
+    assert receiver.finish()==[]
+    return {"blocks":block_count,"model_seconds":block_count*values.size/8000000,
+            "background":"zeros; noisy background covered by six-second integration",
+            "peak_iq_bytes":receiver.peak_buffer_samples*8,"iq_limit_bytes":receiver.buffer_limit_samples*8,
+            "retained_after_finish":receiver.buffer.nbytes+receiver.decimator.values.nbytes,
+            "checkpoints":checkpoints,"rf_tx":0}
+
+
+def run_candidate_replay() -> dict:
+    metadata=json.loads((OUTPUT/"optimized_examples.json").read_text(encoding="utf-8"))
+    result={}
+    for kind, entry in metadata.items():
+        raw=np.load(OUTPUT/f"optimized_{kind}_window.npy")
+        candidate=fast.offline.CoarseCandidate(entry["offset"],entry["score"])
+        def call(optimized=False):
+            with fast.previous.streaming_mode(video.phy.PhyMode(500000)):
+                def decode():
+                    return fast.offline.decode_candidate_from_memmap(raw,entry["sample_count"],candidate,1,
+                        read_iq_window=lambda data,start,end:data[2*start:2*end].view(np.complex64))
+                if not optimized:
+                    return decode()
+                front=fast.BlockFrontEnd(raw.view(np.complex64),65536,Counter())
+                with fast.optimized_context(front): return decode()
+        old=call(False)
+        new=call(True)
+        assert (old.accepted,old.reason,old.refined_start_sample)==(new.accepted,new.reason,new.refined_start_sample)
+        result[kind]={"reason":new.reason,"baseline_arrays":array_observation(lambda:call(False)),
+                      "optimized_arrays":array_observation(lambda:call(True))}
+    return result
+
+
+def run_benchmark(sequence) -> dict:
+    original=video.StreamingReceiver
+    try:
+        video.StreamingReceiver=fast.StreamingReceiver
+        repeats=[video.benchmark_receiver(sequence[:3],500000) for unused in range(3)]
+        profiler=StageProfiler()
+        with profiler.measure(optimized=True): profiled=video.benchmark_receiver(sequence[:3],500000)
+        metadata={}
+        for kind,(raw,count,candidate) in profiler.examples.items():
+            np.save(OUTPUT/f"optimized_{kind}_window.npy",raw)
+            metadata[kind]={"sample_count":count,"offset":candidate.sample_offset,"score":candidate.score}
+        write_result("optimized_examples.json",metadata)
+        return {"repeats":repeats,"median_processing_seconds":statistics.median(row["processing_seconds"] for row in repeats),
+                "profile":profiler.summary(),"profiled_benchmark":profiled,"rf_tx":0}
+    finally:
+        video.StreamingReceiver=original
+
+
+def run_functional_tests() -> None:
+    assert video.maximum_jpeg_bytes(500000,5)==4148
+    assert fast.lab045.conv.CONSTRAINT_LENGTH==7
+
+
+def main() -> None:
+    parser=argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--stage",choices=("benchmark","stream","memory","replay","all"),default="all")
+    args=parser.parse_args()
+    OUTPUT.mkdir(parents=True,exist_ok=True)
+    with Image.open(ROOT/"data/raw/lab011_scene.jpg") as source:
+        sequence=video.encode_sequence(source,video.PROFILES[2],30)
+    for name,function in (("benchmark",lambda:run_benchmark(sequence)),("stream",lambda:run_stream(sequence)),
+                          ("memory",run_memory_test),("replay",run_candidate_replay)):
+        if args.stage not in ("all",name): continue
+        print(f"Lab051 {name}: START; RF TX=0",flush=True)
+        result=function()
+        write_result(f"{name}.json",result)
+        print(json.dumps({key:value for key,value in result.items() if key not in (
+            "blocks","images","repeats","profiled_benchmark","rss_samples")},ensure_ascii=False),flush=True)
+
+
+if __name__=="__main__":
+    main()

experiments/lab051_receiver_profile.py

@@ -0,0 +1,246 @@
+"""Lab051: воспроизводимое профилирование приёмника, без оборудования.
+
+Измерение времени отделено от медленного наблюдения массивов/tracemalloc.
+Классификация valid/false происходит только после штатного PHY/FEC/CRC.
+"""
+
+from __future__ import annotations
+
+from collections import defaultdict
+from contextlib import contextmanager
+from dataclasses import asdict, dataclass
+import ctypes
+from ctypes import wintypes
+import json
+from pathlib import Path
+import statistics
+import sys
+import time
+import tracemalloc
+import weakref
+
+if __package__ in (None, ""):
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+import numpy as np
+from PIL import Image
+
+from experiments import lab046c_offline_recovery as offline
+from experiments import lab050_video_stream as video
+from experiments.lab050_streaming_receiver import streaming_mode
+
+
+ROOT = Path(__file__).resolve().parents[1]
+OUTPUT = ROOT / "data/processed/lab051"
+lab045 = video.phy.lab045
+lab043 = lab045.lab044.lab043
+
+
+def process_memory() -> dict:
+    """Working set Windows без psutil; чужие процессы не перечисляются."""
+    if sys.platform != "win32":
+        return {"rss_bytes": None, "peak_rss_bytes": None}
+
+    class Counters(ctypes.Structure):
+        _fields_ = [("cb", wintypes.DWORD), ("PageFaultCount", wintypes.DWORD)] + [
+            (name, ctypes.c_size_t) for name in (
+                "PeakWorkingSetSize", "WorkingSetSize", "QuotaPeakPagedPoolUsage",
+                "QuotaPagedPoolUsage", "QuotaPeakNonPagedPoolUsage", "QuotaNonPagedPoolUsage",
+                "PagefileUsage", "PeakPagefileUsage")]
+
+    kernel = ctypes.WinDLL("kernel32", use_last_error=True)
+    psapi = ctypes.WinDLL("psapi", use_last_error=True)
+    kernel.GetCurrentProcess.restype = wintypes.HANDLE
+    psapi.GetProcessMemoryInfo.argtypes = [wintypes.HANDLE, ctypes.POINTER(Counters), wintypes.DWORD]
+    counters = Counters()
+    counters.cb = ctypes.sizeof(counters)
+    if not psapi.GetProcessMemoryInfo(kernel.GetCurrentProcess(), ctypes.byref(counters), counters.cb):
+        return {"rss_bytes": None, "peak_rss_bytes": None, "error": ctypes.get_last_error()}
+    return {"rss_bytes": counters.WorkingSetSize, "peak_rss_bytes": counters.PeakWorkingSetSize}
+
+
+@dataclass
+class Stage:
+    calls: int = 0
+    seconds: float = 0.0
+    exclusive_seconds: float = 0.0
+
+
+class StageProfiler:
+    """Только последовательное профилирование; подмены возвращаются в finally."""
+
+    def __init__(self):
+        self.rows = []
+        self.current = None
+        self.shared = defaultdict(Stage)
+        self.stack = []
+        self.examples = {}
+
+    @contextmanager
+    def measure(self, *, optimized: bool = False):
+        patches = []
+        targets = (
+            (offline, "_correlation_scores", "coarse_localization"),
+            (lab043, "_refine_frame_start", "refinement"),
+            (lab043, "_matched_waveform", "matched_filter"),
+            (lab043, "_preamble_timing_offset", "timing"),
+            (lab043, "_timing_offsets", "timing_fallback"),
+            (lab043, "_estimate_cfo_gain", "cfo_gain"),
+            (lab043, "_evaluate_symbols", "phase_diagnostics"),
+            (lab045, "decode_phy_header_soft", "phy_header"),
+            (lab045, "analyze_post_preamble_soft", "body_pipeline"),
+            (lab045.conv, "viterbi_decode_soft", "viterbi"),
+            (lab045, "parse_packet", "crc_parser"),
+        )
+        if optimized:
+            from experiments import lab051_streaming_receiver as fast
+            targets += ((fast.BlockFrontEnd, "refine", "refinement"),
+                        (fast.BlockFrontEnd, "coarse_scores", "coarse_localization"),
+                        (fast.BlockFrontEnd, "matched_window", "matched_filter"),
+                        (fast.BlockFrontEnd, "timing", "timing"),
+                        (fast, "fast_viterbi", "viterbi"))
+
+        def wrap(function, name):
+            def measured(*args, **kwargs):
+                scope = self.shared if self.current is None else self.current
+                stage = scope[name]
+                started = time.perf_counter()
+                entry = [0.0]
+                self.stack.append(entry)
+                try:
+                    return function(*args, **kwargs)
+                finally:
+                    elapsed = time.perf_counter() - started
+                    self.stack.pop()
+                    if self.stack:
+                        self.stack[-1][0] += elapsed
+                    stage.calls += 1
+                    stage.seconds += elapsed
+                    stage.exclusive_seconds += elapsed - entry[0]
+            return measured
+
+        for owner, name, stage in targets:
+            original = getattr(owner, name)
+            patches.append((owner, name, original))
+            setattr(owner, name, wrap(original, stage))
+        original_decode = offline.decode_candidate_from_memmap
+
+        def candidate(raw, sample_count, coarse, number, **kwargs):
+            self.current = defaultdict(Stage)
+            started = time.perf_counter()
+            try:
+                outcome = original_decode(raw, sample_count, coarse, number, **kwargs)
+                elapsed = time.perf_counter() - started
+                kind = "valid" if outcome.accepted else "false"
+                self.rows.append({"kind": kind, "seconds": elapsed, "reason": outcome.reason,
+                                  "offset": coarse.sample_offset, "refined": outcome.refined_start_sample,
+                                  "stages": {name: asdict(row) for name, row in self.current.items()}})
+                # По одному примеру после решения, только для отдельного replay.
+                if kind not in self.examples and (kind == "valid" or outcome.reason.startswith("PHY_HEADER_MAGIC")):
+                    self.examples[kind] = (raw.copy(), sample_count, coarse)
+                return outcome
+            finally:
+                self.current = None
+
+        offline.decode_candidate_from_memmap = candidate
+        try:
+            yield self
+        finally:
+            offline.decode_candidate_from_memmap = original_decode
+            for owner, name, original in reversed(patches):
+                setattr(owner, name, original)
+
+    def summary(self) -> dict:
+        result = {"shared_stages": {name: asdict(row) for name, row in self.shared.items()}}
+        for kind in ("valid", "false"):
+            selected = [row for row in self.rows if row["kind"] == kind]
+            stages = defaultdict(Stage)
+            for row in selected:
+                for name, stage in row["stages"].items():
+                    stages[name].calls += stage["calls"]
+                    stages[name].seconds += stage["seconds"]
+                    stages[name].exclusive_seconds += stage["exclusive_seconds"]
+            times = [row["seconds"] for row in selected]
+            result[kind] = {"count": len(times), "mean_seconds": statistics.mean(times) if times else None,
+                            "median_seconds": statistics.median(times) if times else None,
+                            "total_seconds": sum(times),
+                            "stages": {name: asdict(row) for name, row in stages.items()}}
+        return result
+
+
+def array_observation(function) -> dict:
+    """Нижняя граница числа массивов в локальных переменных проектного Python.
+
+    Временные выражения и внутренности C/SciPy не видны line tracer-у.
+    tracemalloc отдельно считает пик отслеживаемой памяти, включая NumPy.
+    Эти проходы НИКОГДА не используются как измерение производительности.
+    """
+    seen = {}
+    counters = {"observed_array_objects": 0, "observed_owned_arrays": 0, "observed_owned_bytes_total": 0}
+
+    def trace(frame, event, argument):
+        filename = frame.f_code.co_filename.replace("\\", "/")
+        if not ("/experiments/" in filename or "/protocol/" in filename):
+            return None
+        if event in ("line", "return"):
+            for value in frame.f_locals.values():
+                if not isinstance(value, np.ndarray):
+                    continue
+                identity = id(value)
+                if identity in seen and seen[identity]() is value:
+                    continue
+                seen[identity] = weakref.ref(value)
+                counters["observed_array_objects"] += 1
+                if value.flags.owndata:
+                    counters["observed_owned_arrays"] += 1
+                    counters["observed_owned_bytes_total"] += value.nbytes
+        return trace
+
+    previous = sys.gettrace()
+    try:
+        sys.settrace(trace)
+        function()
+    finally:
+        sys.settrace(previous)
+    tracemalloc.start()
+    try:
+        function()
+        current, peak = tracemalloc.get_traced_memory()
+    finally:
+        tracemalloc.stop()
+    return {**counters, "tracemalloc_retained_bytes": current, "tracemalloc_peak_bytes": peak,
+            "array_count_is_lower_bound": True}
+
+
+def run_functional_tests() -> None:
+    assert video.phy.PhyMode(500_000).samples_per_symbol == 16
+
+
+def main() -> None:
+    OUTPUT.mkdir(parents=True, exist_ok=True)
+    with Image.open(ROOT / "data/raw/lab011_scene.jpg") as source:
+        sequence = video.encode_sequence(source, video.PROFILES[2], 3)
+    profiler = StageProfiler()
+    with profiler.measure():
+        benchmark = video.benchmark_receiver(sequence, 500_000)
+    result = {"benchmark": benchmark, "profile": profiler.summary(), "rf_tx": 0}
+    (OUTPUT / "baseline_profile.json").write_text(json.dumps(result, indent=2) + "\n", encoding="utf-8")
+    print(json.dumps(result["profile"], indent=2), flush=True)
+    # Небольшие ограниченные окна, не полная запись. В Git не включать.
+    metadata = {}
+    for kind, (raw, count, coarse) in profiler.examples.items():
+        np.save(OUTPUT / f"baseline_{kind}_window.npy", raw)
+        metadata[kind] = {"sample_count": count, "offset": coarse.sample_offset, "score": coarse.score}
+        with streaming_mode(video.phy.PhyMode(500_000)):
+            call = lambda: offline.decode_candidate_from_memmap(
+                raw, count, coarse, 1,
+                read_iq_window=lambda data, start, end: data[2*start:2*end].view(np.complex64))
+            result.setdefault("array_observation", {})[kind] = array_observation(call)
+    (OUTPUT / "baseline_examples.json").write_text(json.dumps(metadata, indent=2) + "\n", encoding="utf-8")
+    result["process_memory_after_profiling"] = process_memory()
+    (OUTPUT / "baseline_profile.json").write_text(json.dumps(result, indent=2) + "\n", encoding="utf-8")
+    print(json.dumps(result["array_observation"]), flush=True)
+
+
+if __name__ == "__main__":
+    main()

experiments/lab051_streaming_receiver.py

@@ -0,0 +1,364 @@
+"""Lab051: общий front-end и эквивалентный timing на 500 ксимв/с.
+
+PHY, FEC, CRC и порог первого каскада заимствованы без изменения правил.
+Подмены старых функций действуют только внутри последовательного push.
+"""
+
+from __future__ import annotations
+
+from collections import Counter
+from contextlib import contextmanager
+from functools import lru_cache
+import math
+import time
+
+import numpy as np
+from scipy.signal import convolve, fftconvolve, oaconvolve, firwin, resample_poly
+
+from experiments import lab046b_continuous_iq as iq
+from experiments import lab046c_offline_recovery as offline
+from experiments import lab050_streaming_receiver as previous
+
+
+phy = previous.phy
+lab045 = phy.lab045
+lab043 = lab045.lab044.lab043
+ORIGINAL_REFINE = lab043._refine_frame_start
+ORIGINAL_TIMING = lab043._preamble_timing_offset
+ORIGINAL_MATCHED = lab043._matched_waveform
+ORIGINAL_VITERBI = lab045.conv.viterbi_decode_soft
+
+
+@lru_cache(maxsize=4)
+def timing_kernel(samples_per_symbol: int) -> np.ndarray:
+    """Те же символы преамбулы на исходной сетке отсчётов."""
+    expected = iq.radio.bpsk_modulate(lab045.lab044.preamble_bits())
+    kernel = np.zeros((expected.size - 1) * samples_per_symbol + 1, np.complex64)
+    kernel[::samples_per_symbol] = np.conj(expected)
+    return kernel[::-1].copy()
+
+
+def fast_preamble_timing(matched: np.ndarray, nominal: int, expected_symbols: np.ndarray) -> int:
+    """Все прежние 1601 смещения; отрицательные/усечённые окна имеют score=0.
+
+    Корреляция с разреженной преамбулой эквивалентна vdot каждой выборки.
+    Энергии получаются разностями сумм на каждой из 16 фаз. Диапазон не сужен.
+    """
+    samples_per_symbol = lab045.SAMPLES_PER_SYMBOL
+    offsets = np.arange(-800, 801)
+    starts = nominal + offsets
+    count = len(expected_symbols)
+    # Тело не участвует ни в одном из прежних 1601 timing-окон.
+    matched = matched[:nominal + 800 + (count - 1) * samples_per_symbol + 1]
+    valid = (starts >= 0) & (starts + (count - 1) * samples_per_symbol < matched.size)
+    scores = np.zeros(offsets.size, np.float64)
+    if np.any(valid):
+        kernel = timing_kernel(samples_per_symbol)
+        correlations = fftconvolve(matched, kernel, mode="valid")
+        padding = (-matched.size) % samples_per_symbol
+        power = np.pad(np.abs(matched) ** 2, (0, padding)).reshape(-1, samples_per_symbol)
+        cumulative = np.concatenate((np.zeros((1, samples_per_symbol)),
+                                     np.cumsum(power, axis=0, dtype=np.float64))).ravel()
+        selected = starts[valid]
+        energy = cumulative[selected + count * samples_per_symbol] - cumulative[selected]
+        scores[valid] = np.abs(correlations[selected]) / np.sqrt(count * np.maximum(energy, 0) + 1e-30)
+    largest = np.partition(scores, -2)[-2:]
+    if largest[-1] > 0 and largest[-1] - largest[-2] <= 32 * np.finfo(np.float32).eps * largest[-1]:
+        # FFT меняет последние разряды. На неразделимых пиках сохраняем
+        # исходное суммирование и его tie→первый offset, включая чистый тон.
+        return ORIGINAL_TIMING(matched, nominal, expected_symbols)
+    return int(offsets[int(np.argmax(scores))])
+
+
+@lru_cache(maxsize=8)
+def carrier_rotation(length: int) -> np.ndarray:
+    """Кэш переносов не очищается при каждом входном блоке старым адаптером."""
+    indexes = np.arange(length)
+    return np.exp(-2j * np.pi * lab043.CARRIER_OFFSET_HZ * indexes / lab045.SAMPLE_RATE_HZ).astype(np.complex64)
+
+
+@lru_cache(maxsize=8)
+def dc_response(length: int, samples_per_symbol: int) -> np.ndarray:
+    """Отклик локального окна постоянной составляющей после переноса/RRC."""
+    rotation = carrier_rotation(length)
+    return oaconvolve(rotation, lab043._matched_filter_taps(), mode="full")
+
+
+@lru_cache(maxsize=4)
+def decimation_taps(decimation: int) -> np.ndarray:
+    """Штатные коэффициенты resample_poly(up=1, window=('kaiser', 8))."""
+    return firwin(20 * decimation + 1, 1 / decimation, window=("kaiser", 8)).astype(np.float32)
+
+
+@lru_cache(maxsize=8)
+def detection_rotation(length: int, decimation: int) -> np.ndarray:
+    return np.exp(-2j * np.pi * lab045.CARRIER_OFFSET_HZ * np.arange(length)
+                  / (lab045.SAMPLE_RATE_HZ / decimation)).astype(np.complex64)
+
+
+class DecimationCache:
+    """Повторно использовать внутреннюю часть FIR, пересчитать только края.
+
+    resample_poly имеет поддержку ±10 выходных отсчётов. После сдвига окна
+    неизменная внутренняя область даёт ровно те же значения; левый нулевой
+    край и новое продолжение считаются с прежними коэффициентами заново.
+    """
+
+    def __init__(self, decimation: int):
+        self.decimation = decimation
+        self.base = self.input_count = 0
+        self.values = np.empty(0, np.complex64)
+        self.reused_samples = self.filtered_input_samples = 0
+
+    def get(self, samples: np.ndarray, base: int) -> np.ndarray:
+        stride = self.decimation
+        count = math.ceil(samples.size / stride)
+        shift = (base - self.base) // stride
+        low = max(10, -shift + 10)
+        high = min(count - 10, self.values.size - 11 - shift)
+
+        def filtered(array):
+            self.filtered_input_samples += array.size
+            return resample_poly(array, 1, stride, window=decimation_taps(stride))
+
+        if not self.values.size or (base - self.base) % stride or high <= low:
+            output = filtered(samples)
+        else:
+            output = np.empty(count, np.complex64)
+            output[low:high] = self.values[low + shift:high + shift]
+            self.reused_samples += high - low
+            prefix_end = min(samples.size, (low + 11) * stride)
+            output[:low] = filtered(samples[:prefix_end])[:low]
+            suffix_start = (high - 11) * stride
+            output[high:] = filtered(samples[suffix_start:])[11:11 + count - high]
+        self.values = output
+        self.base, self.input_count = base, samples.size
+        return output
+
+
+def fast_viterbi(soft_values: np.ndarray, information_bit_count: int) -> np.ndarray:
+    """Та же евклидова метрика и tie→branch 0, с готовыми метриками ветвей."""
+    conv = lab045.conv
+    if not isinstance(information_bit_count, (int, np.integer)) or information_bit_count <= 0:
+        return ORIGINAL_VITERBI(soft_values, information_bit_count)
+    values = np.asarray(soft_values, dtype=np.float64)
+    if (values.ndim != 1 or values.size != conv.encoded_bit_count(information_bit_count)
+            or not np.all(np.isfinite(values))):
+        return ORIGINAL_VITERBI(soft_values, information_bit_count)
+    step_count = int(information_bit_count) + conv.TAIL_BIT_COUNT
+    received = values.reshape(step_count, 2)
+    # Всего четыре различных пары BPSK-символов. Порядок арифметики исходный.
+    negative = (received + 1.0) ** 2
+    positive = (received - 1.0) ** 2
+    metrics = np.column_stack((negative[:, 0] + negative[:, 1], negative[:, 0] + positive[:, 1],
+                               positive[:, 0] + negative[:, 1], positive[:, 0] + positive[:, 1]))
+    symbols = conv._BRANCH_SYMBOLS
+    codes = (2 * (symbols[:, :, 0] > 0) + (symbols[:, :, 1] > 0)).astype(np.intp)
+    metric_zero = metrics[:, codes[:, 0]]
+    metric_one = metrics[:, codes[:, 1]]
+    predecessor_zero = conv._PREDECESSORS[:, 0]
+    predecessor_one = conv._PREDECESSORS[:, 1]
+    path = np.full(conv.STATE_COUNT, 1e18)
+    path[0] = 0.0
+    choices = np.empty((step_count, conv.STATE_COUNT), np.uint8)
+    for step in range(step_count):
+        zero = path[predecessor_zero] + metric_zero[step]
+        one = path[predecessor_one] + metric_one[step]
+        choices[step] = one < zero
+        path = np.minimum(zero, one)
+    decoded = np.empty(step_count, np.uint8)
+    state = 0
+    for step in range(step_count - 1, -1, -1):
+        decoded[step] = state >> (conv.MEMORY - 1)
+        state = int(conv._PREDECESSORS[state, choices[step, state]])
+    assert state == 0
+    return decoded[:information_bit_count]
+
+
+class BlockFrontEnd:
+    """Один ограниченный общий участок; кандидаты читают его представления.
+
+    Конечные края и локальное вычитание среднего восстановлены отдельно:
+    нельзя без этого подменять фильтрацию изолированного окна непрерывной.
+    """
+
+    def __init__(self, samples: np.ndarray, owned_end: int, statistics: Counter,
+                 *, decimator: DecimationCache | None = None, base: int = 0):
+        self.samples = samples
+        self.statistics = statistics
+        self.decimator, self.base = decimator, base
+        self.reference = lab045.lab044._preamble_reference_samples()
+        self.reference_length = (16 + lab043.PILOT_SYMBOL_COUNT + 10) * lab045.SAMPLES_PER_SYMBOL
+        self.probe_symbols = lab045.PREAMBLE_BIT_COUNT + lab045.PHY_HEADER_BIT_COUNT
+        self.probe_waveform = (self.probe_symbols + 42) * lab045.SAMPLES_PER_SYMBOL
+        self.front_length = min(samples.size, owned_end + self.probe_waveform + offline.REFINEMENT_MARGIN_SAMPLES)
+        self.owned_end = owned_end
+        self.prefix_correlation = None
+        self.changes = None
+        self.matched = None
+        self.baseband = None
+        self.soft_cache = {}
+
+    def coarse_scores(self, samples: np.ndarray, reference: np.ndarray) -> np.ndarray:
+        """Сохранить DC всего старого окна, считать корреляцию только owned+1."""
+        decimation = lab043.DETECTION_DECIMATION
+        needed = math.ceil(self.owned_end / decimation) + reference.size + 1
+        raw = (resample_poly(samples, 1, decimation, window=decimation_taps(decimation)) if self.decimator is None
+               else self.decimator.get(samples, self.base))
+        mean = np.mean(raw, dtype=np.complex128).astype(np.complex64)
+        values = raw[:needed].copy()
+        values -= mean
+        values *= detection_rotation(values.size, decimation)
+        if values.size < reference.size:
+            return np.empty(0, np.float32)
+        scores = lab043._normalized_sliding_correlation(values, reference).astype(np.float32)
+        scores[~np.isfinite(scores)] = 0
+        np.clip(scores, 0, 1, out=scores)
+        return scores
+
+    def local_start(self, array: np.ndarray) -> int:
+        return (array.ctypes.data - self.samples.ctypes.data) // self.samples.itemsize
+
+    def refine(self, samples: np.ndarray, reference: np.ndarray, coarse_start: int) -> int:
+        start = self.local_start(samples)
+        low = max(0, coarse_start - 2000)
+        high = min(samples.size, coarse_start + self.reference_length + 2000)
+        if high - low < self.reference_length or start < 0:
+            return ORIGINAL_REFINE(samples, reference, coarse_start)
+        if self.changes is None:
+            self.changes = np.empty(self.front_length, np.int64)
+            self.changes[0] = 0
+            np.cumsum(self.samples[1:self.front_length] != self.samples[:self.front_length-1],
+                      dtype=np.int64, out=self.changes[1:])
+        if self.changes[start + samples.size - 1] == self.changes[start]:
+            # После вычитания локального среднего точно нулевой сигнал.
+            # Общая FFT может оставить шум округления вместо нулевого tie.
+            return low
+        if self.prefix_correlation is None:
+            centered = self.reference[:self.reference_length] - np.mean(self.reference[:self.reference_length])
+            self.centered_reference_sum = np.sum(np.conj(centered), dtype=np.complex128)
+            self.prefix_correlation = fftconvolve(
+                self.samples[:self.front_length], np.conj(centered[::-1]), mode="valid")
+            self.statistics["shared_refinement_correlations"] += 1
+        first = start + low
+        last = start + high - self.reference_length + 1
+        correlation = self.prefix_correlation[first:last]
+        if not correlation.size:
+            return ORIGINAL_REFINE(samples, reference, coarse_start)
+        # Та же DC-поправка, что у корреляции (samples - mean(probe)).
+        correlation = correlation - np.mean(samples) * self.centered_reference_sum
+        magnitudes = np.abs(correlation)
+        if magnitudes.size > 1:
+            largest = np.partition(magnitudes, -2)[-2:]
+            if largest[-1] - largest[-2] <= 32*np.finfo(np.float32).eps*largest[-1]:
+                self.statistics["refinement_tie_fallbacks"] += 1
+                return ORIGINAL_REFINE(samples, reference, coarse_start)
+        return low + int(np.argmax(magnitudes))
+
+    def matched_window(self, samples: np.ndarray, start: int, length: int):
+        absolute = self.local_start(samples) + start
+        taps = lab043._matched_filter_taps()
+        edge = taps.size - 1
+        if absolute < 0 or absolute + length + edge > self.front_length:
+            self.statistics["local_body_matched_calls"] += 1
+            segment = samples[start:start + length] - np.mean(samples)
+            baseband = segment * carrier_rotation(length)
+            return oaconvolve(baseband, taps, mode="full"), 16 * lab045.SAMPLES_PER_SYMBOL + edge
+        if self.matched is None:
+            self.baseband = self.samples[:self.front_length] * carrier_rotation(self.front_length)
+            self.matched = oaconvolve(self.baseband, taps, mode="full")
+            self.statistics["shared_matched_calls"] += 1
+        matched = self.matched[absolute:absolute + length + edge].copy()
+        # Восстановить нулевые края исходного локального окна, без соседнего IQ.
+        matched[:edge] = convolve(self.baseband[absolute:absolute + edge], taps, method="direct")[:edge]
+        matched[-edge:] = convolve(self.baseband[absolute + length - edge:absolute + length], taps,
+                                  method="direct")[edge:]
+        phase = np.complex64(np.exp(2j * np.pi * lab043.CARRIER_OFFSET_HZ * absolute / lab045.SAMPLE_RATE_HZ))
+        matched *= phase
+        mean = np.mean(samples)
+        matched -= mean * dc_response(length, lab045.SAMPLES_PER_SYMBOL)
+        nominal = 16 * lab045.SAMPLES_PER_SYMBOL + edge
+        return matched, nominal
+
+    def timing(self, matched, nominal, expected):
+        """Локальный FFT оказался дешевле дополнительных общих timing-массивов."""
+        return fast_preamble_timing(matched, nominal, expected)
+
+    def extract(self, samples: np.ndarray, frame_start: int, symbol_count: int) -> iq.SoftWindow:
+        key = (self.local_start(samples) + frame_start, symbol_count)
+        # Усечение зависит и от доступности текущего probe, а не только от start.
+        length = (symbol_count + 42) * lab045.SAMPLES_PER_SYMBOL
+        if frame_start < 0 or frame_start + length > samples.size:
+            return iq.SoftWindow(False, np.empty(0), 0, 0, 0, {}, "TRUNCATED_IQ_WINDOW")
+        if key in self.soft_cache:
+            self.statistics["soft_window_cache_hits"] += 1
+            return self.soft_cache[key]
+        segment = samples[frame_start:frame_start + length]
+        matched, nominal = lab043._matched_waveform(segment, 0, length)
+        bits = lab045.lab044.preamble_bits()
+        expected = iq.radio.bpsk_modulate(bits)
+        timing = lab043._preamble_timing_offset(matched, nominal, expected)
+        carrier = lab043._evaluate_symbols(matched, nominal, timing, bits, lab045.PREAMBLE_BIT_COUNT)
+        if carrier["abs_h"] == 0:
+            result = iq.SoftWindow(False, np.empty(0), timing, 0, nominal, carrier, "ZERO_GAIN")
+        else:
+            symbols = matched[nominal + timing:nominal + timing + symbol_count * lab045.SAMPLES_PER_SYMBOL:
+                              lab045.SAMPLES_PER_SYMBOL].astype(np.complex128)
+            if symbols.size != symbol_count:
+                result = iq.SoftWindow(False, np.empty(0), timing, 0, nominal, carrier, "TRUNCATED_SYMBOLS")
+            else:
+                symbol_time = (np.arange(symbol_count, dtype=np.float64) + 0.5) / lab045.SYMBOL_RATE
+                normalized = symbols * np.exp(-2j * math.pi * carrier["cfo_hz"] * symbol_time)
+                normalized /= carrier["abs_h"] * np.exp(1j * carrier["phase_rad"])
+                result = iq.SoftWindow(True, normalized.real, int(timing),
+                                      int((frame_start + nominal + timing) % lab045.SAMPLES_PER_SYMBOL),
+                                      nominal, carrier, None)
+        self.soft_cache[key] = result
+        self.statistics["unique_soft_windows"] += 1
+        return result
+
+
+@contextmanager
+def optimized_context(front: BlockFrontEnd):
+    targets = ((lab043, "_refine_frame_start", front.refine),
+               (lab043, "_matched_waveform", front.matched_window),
+               (lab043, "_preamble_timing_offset", front.timing),
+               (iq, "_extract_soft_window", front.extract),
+               (lab045.conv, "viterbi_decode_soft", fast_viterbi),
+               (offline, "_correlation_scores", front.coarse_scores))
+    originals = [(owner, name, getattr(owner, name)) for owner, name, unused in targets]
+    try:
+        for owner, name, function in targets:
+            setattr(owner, name, function)
+        yield
+    finally:
+        for owner, name, function in reversed(originals):
+            setattr(owner, name, function)
+
+
+class StreamingReceiver(previous.StreamingReceiver):
+    """Совместимый API Lab050, ускорение только для рабочей точки 500 ксимв/с."""
+
+    def __init__(self, symbol_rate: int = 500_000, *, block_samples: int = 65_536):
+        if symbol_rate != 500_000:
+            raise ValueError("Lab051 исследует только 500 ксимв/с")
+        super().__init__(symbol_rate, block_samples=block_samples)
+        self.front_statistics = Counter()
+        self.decimator = DecimationCache(self.decimation)
+
+    def _scan(self, owned_end: int) -> list[previous.ReceivedPacket]:
+        front = BlockFrontEnd(self.buffer, owned_end - self.base_sample, self.front_statistics,
+                              decimator=self.decimator, base=self.base_sample)
+        with optimized_context(front):
+            return super()._scan(owned_end)
+
+    def finish(self):
+        result = super().finish()
+        self.decimator.values = np.empty(0, np.complex64)
+        return result
+
+
+def run_functional_tests() -> None:
+    receiver = StreamingReceiver()
+    assert receiver.mode.samples_per_symbol == 16
+    assert receiver.finish() == []