375 lines
20 KiB
Python
375 lines
20 KiB
Python
"""Lab052: воспроизводимый профиль обнаружителя, только программный IQ.
|
||
|
||
Диагностическая история принадлежит эксперименту. Приёмник не получает её
|
||
обратно; эталон используется только после независимых решений PHY/FEC/CRC.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import cProfile
|
||
from collections import Counter
|
||
from dataclasses import replace
|
||
import hashlib
|
||
import json
|
||
from pathlib import Path
|
||
import sys
|
||
import time
|
||
from types import SimpleNamespace
|
||
|
||
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 lab051_streaming_receiver as previous
|
||
from experiments import lab051_receiver_performance as performance
|
||
from experiments.lab051_receiver_profile import process_memory
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
OUTPUT = ROOT / "data/processed/lab052"
|
||
|
||
|
||
def write_result(name: str, value: dict) -> None:
|
||
OUTPUT.mkdir(parents=True, exist_ok=True)
|
||
(OUTPUT / name).write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
|
||
|
||
def sequence_fixture():
|
||
"""Тот же генератор Lab051, готовность JPEG фиксируется первым замером.
|
||
|
||
Прежний опыт не сохранял encode_seconds, поэтому побитную историческую
|
||
временную сетку восстановить нельзя. Все новые сравнения используют одну
|
||
сохранённую сетку, один seed и те же байты JPEG, включая реальную encode cost.
|
||
"""
|
||
with Image.open(ROOT / "data/raw/lab011_scene.jpg") as source:
|
||
sequence = performance.video.encode_sequence(source, performance.video.PROFILES[2], 30)
|
||
path = OUTPUT / "source_fixture.json"
|
||
if path.exists():
|
||
fixture = json.loads(path.read_text(encoding="utf-8"))
|
||
for item, recorded in zip(sequence, fixture["images"], strict=True):
|
||
assert hashlib.sha256(item.jpeg).hexdigest() == recorded["sha256"]
|
||
return [replace(item, encode_seconds=recorded["encode_seconds"])
|
||
for item, recorded in zip(sequence, fixture["images"], strict=True)]
|
||
write_result(path.name, {"seed": 50050, "description": "Lab051 generator; frozen measured JPEG readiness",
|
||
"images": [{"image_id": item.image_id, "bytes": len(item.jpeg),
|
||
"encode_seconds": item.encode_seconds,
|
||
"sha256": hashlib.sha256(item.jpeg).hexdigest()} for item in sequence]})
|
||
return sequence
|
||
|
||
|
||
def quantiles(values) -> dict:
|
||
values = np.asarray(values, dtype=np.float64)
|
||
if not values.size:
|
||
return {"count": 0}
|
||
median = float(np.median(values))
|
||
return {"count": values.size, "min": float(values.min()), "median": median,
|
||
"mad": float(np.median(np.abs(values - median))),
|
||
"p90": float(np.quantile(values, .9)), "p99": float(np.quantile(values, .99)),
|
||
"p999": float(np.quantile(values, .999)), "max": float(values.max())}
|
||
|
||
|
||
class DetectionObserver:
|
||
"""Полная история только для конечного лабораторного профиля, вне RX."""
|
||
|
||
def __init__(self):
|
||
self.peaks = []
|
||
self.blocks = []
|
||
self.outcomes = []
|
||
self.cheap_rows = []
|
||
|
||
def scores(self, scores, base, low, high, decimation):
|
||
indexes = previous.offline._all_local_peak_indexes(scores)
|
||
offsets = base + indexes * decimation
|
||
selected = (offsets >= low) & (offsets < high)
|
||
self.peaks.append(np.column_stack((offsets[selected], scores[indexes[selected]])))
|
||
first = max(0, (low - base + decimation - 1) // decimation)
|
||
last = min(len(scores), (high - base + decimation - 1) // decimation)
|
||
self.blocks.append({"start_sample": low, "end_sample": high,
|
||
"score_statistics": quantiles(scores[first:last]),
|
||
"peak_statistics": quantiles(scores[indexes[selected]])})
|
||
|
||
def outcome(self, base, candidate, outcome):
|
||
self.outcomes.append((base + candidate.sample_offset, candidate.score,
|
||
base + outcome.refined_start_sample, int(outcome.accepted)))
|
||
|
||
def cheap(self, absolute, score, parts, passed):
|
||
self.cheap_rows.append([absolute, score, int(passed), *parts])
|
||
|
||
def save(self, name, result):
|
||
peaks = np.concatenate(self.peaks) if self.peaks else np.empty((0, 2))
|
||
outcomes = np.asarray(self.outcomes).reshape(-1, 4)
|
||
np.savez_compressed(OUTPUT / f"{name}_observations.npz", peaks=peaks, outcomes=outcomes)
|
||
if self.cheap_rows:
|
||
np.save(OUTPUT / f"{name}_cheap.npy", np.asarray(self.cheap_rows))
|
||
threshold = previous.phy.lab047.FIXED_HARDWARE_CANDIDATE_THRESHOLD
|
||
valid = outcomes[:, 3] == 1
|
||
# Сначала строгий вердикт, затем классификация корреляционной статистики.
|
||
result["detection_profile"] = {
|
||
"total_local_peaks": len(peaks), "first_stage_candidates": int(np.sum(peaks[:, 1] >= threshold)),
|
||
"strict_candidates": len(outcomes), "strict_true_candidates": int(np.sum(valid)),
|
||
"strict_false_candidates": int(np.sum(~valid)),
|
||
"candidates_per_rf_second": len(outcomes) / result["iq_seconds"],
|
||
"false_candidates_per_rf_second": int(np.sum(~valid)) / result["iq_seconds"],
|
||
"all_local_peaks": quantiles(peaks[:, 1]),
|
||
"strict_true_scores": quantiles(outcomes[valid, 1]),
|
||
"strict_false_scores": quantiles(outcomes[~valid, 1]),
|
||
"blocks": self.blocks,
|
||
}
|
||
|
||
|
||
class ProfileBaseline(previous.StreamingReceiver):
|
||
"""Штатный Lab051 с наблюдением после корреляции и после решения."""
|
||
|
||
def __init__(self, observer):
|
||
super().__init__()
|
||
self.observer = observer
|
||
|
||
def _scan(self, owned_end):
|
||
front = previous.BlockFrontEnd(self.buffer, owned_end - self.base_sample, self.front_statistics,
|
||
decimator=self.decimator, base=self.base_sample)
|
||
original_scores = front.coarse_scores
|
||
|
||
def scores(samples, reference):
|
||
values = original_scores(samples, reference)
|
||
self.observer.scores(values, self.base_sample, self.scan_sample, owned_end, self.decimation)
|
||
return values
|
||
|
||
front.coarse_scores = scores
|
||
original_decode = previous.offline.decode_candidate_from_memmap
|
||
|
||
def decode(raw, count, candidate, number, **kwargs):
|
||
outcome = original_decode(raw, count, candidate, number, **kwargs)
|
||
self.observer.outcome(self.base_sample, candidate, outcome)
|
||
return outcome
|
||
|
||
try:
|
||
previous.offline.decode_candidate_from_memmap = decode
|
||
with previous.optimized_context(front):
|
||
return previous.previous.StreamingReceiver._scan(self, owned_end)
|
||
finally:
|
||
previous.offline.decode_candidate_from_memmap = original_decode
|
||
|
||
|
||
def measured_stream(sequence, factory):
|
||
"""Повторно использовать исходную событийную шкалу с фактическими push times."""
|
||
instances = []
|
||
|
||
def create():
|
||
receiver = factory()
|
||
instances.append(receiver)
|
||
return receiver
|
||
|
||
original = performance.fast
|
||
try:
|
||
performance.fast = SimpleNamespace(StreamingReceiver=create)
|
||
result = performance.run_stream(sequence)
|
||
finally:
|
||
performance.fast = original
|
||
receiver = instances[0]
|
||
result["coarse_seconds"] = receiver.coarse_seconds
|
||
result["strict_seconds"] = receiver.strict_seconds
|
||
result["peak_candidates_in_block"] = receiver.peak_candidate_count
|
||
if hasattr(receiver, "detection_statistics"):
|
||
result["detector"] = {**dict(receiver.detection_statistics),
|
||
"strict_true_candidates": receiver.packet_count,
|
||
"strict_false_candidates": receiver.candidate_count - receiver.packet_count,
|
||
"strict_checks_per_rf_second": receiver.candidate_count / result["iq_seconds"],
|
||
"cheap_seconds": receiver.cheap_seconds, "cluster_seconds": receiver.cluster_seconds,
|
||
"true_strict_seconds": receiver.true_strict_seconds,
|
||
"false_strict_seconds": receiver.false_strict_seconds,
|
||
"peak_cluster_count": receiver.peak_cluster_count,
|
||
"peak_eligible_count": receiver.peak_eligible_count,
|
||
"candidate_state_after_finish": receiver.candidate_state_count,
|
||
"detection_arrays_peak_estimate_bytes": receiver.working_detection_arrays_peak_bytes}
|
||
return result, receiver
|
||
|
||
|
||
def profile_baseline(sequence):
|
||
observer = DetectionObserver()
|
||
result, receiver = measured_stream(sequence, lambda: ProfileBaseline(observer))
|
||
observer.save("baseline", result)
|
||
result["timing_includes_diagnostic_observer"] = True
|
||
write_result("baseline_profile.json", result)
|
||
print(json.dumps({key: value for key, value in result.items() if key in (
|
||
"candidate_count", "received_images", "rx_physical_packets", "processing_per_rf_second",
|
||
"coarse_seconds", "strict_seconds")}), flush=True)
|
||
print(json.dumps({key: value for key, value in result["detection_profile"].items() if key != "blocks"}), flush=True)
|
||
|
||
|
||
def unprofiled_baseline(sequence):
|
||
result, receiver = measured_stream(sequence, previous.StreamingReceiver)
|
||
result["timing_includes_diagnostic_observer"] = False
|
||
write_result("baseline_stream.json", result)
|
||
print(json.dumps({key: result[key] for key in ("candidate_count", "processing_seconds",
|
||
"processing_per_rf_second", "rx_physical_packets", "received_images", "object_jpeg_integrity")}), flush=True)
|
||
|
||
|
||
def run_functional_tests() -> None:
|
||
assert quantiles([1, 2, 3])["median"] == 2
|
||
|
||
|
||
def compare_rules(sequence):
|
||
from experiments import lab052_candidate_detection as detection
|
||
baseline = json.loads((OUTPUT / "baseline_profile.json").read_text(encoding="utf-8"))
|
||
results = []
|
||
for parts in (0, 1, 2, 3):
|
||
observer = DetectionObserver()
|
||
rule = detection.DetectorRule(parts=parts)
|
||
result, receiver = measured_stream(sequence, lambda: detection.StreamingReceiver(rule=rule, observer=observer))
|
||
observer.save(f"parts{parts}", result)
|
||
result["rule"] = {"parts": parts, "width_samples": rule.cluster_width_samples,
|
||
"part_threshold": detection.part_threshold(parts) if parts else None}
|
||
result["true_candidates_lost"] = baseline["rx_physical_packets"] - result["rx_physical_packets"]
|
||
false_count = result["detector"]["strict_false_candidates"]
|
||
baseline_false = baseline["detection_profile"]["strict_false_candidates"]
|
||
result["false_candidate_reduction"] = baseline_false / false_count if false_count else None
|
||
result["false_candidate_reduction_fraction"] = 1 - false_count / baseline_false
|
||
result["timing_includes_diagnostic_observer"] = True
|
||
write_result(f"parts{parts}.json", result)
|
||
compact = {key: result[key] for key in ("rule", "received_images", "object_jpeg_integrity",
|
||
"true_candidates_lost", "false_candidate_reduction", "detector",
|
||
"processing_per_rf_second", "fresh_display_fps")}
|
||
results.append(compact)
|
||
print(json.dumps(compact), flush=True)
|
||
write_result("rule_comparison.json", {"rules": results, "rf_tx": 0})
|
||
|
||
|
||
def optimized_stream(sequence):
|
||
from experiments import lab052_candidate_detection as detection
|
||
results = []
|
||
for number in range(1, 4):
|
||
result, receiver = measured_stream(sequence, detection.StreamingReceiver)
|
||
result["timing_includes_diagnostic_observer"] = False
|
||
write_result(f"stream_repeat_{number}.json", result)
|
||
results.append(result)
|
||
print(json.dumps({"repeat": number, "ratio": result["processing_per_rf_second"],
|
||
"fresh_fps": result["fresh_display_fps"], "packets": result["rx_physical_packets"]}), flush=True)
|
||
result = sorted(results, key=lambda row: row["processing_seconds"])[1]
|
||
result["selection"] = "median processing time of three complete measured passes; one whole timeline"
|
||
write_result("stream.json", result)
|
||
print(json.dumps({key: value for key, value in result.items() if key not in (
|
||
"blocks", "images", "rss_samples", "fresh_image_age_seconds")}), flush=True)
|
||
|
||
|
||
def profile_cpu(sequence):
|
||
from experiments import lab052_candidate_detection as detection
|
||
import pstats
|
||
from io import StringIO
|
||
profiler = cProfile.Profile()
|
||
|
||
class ProfiledReceiver(detection.StreamingReceiver):
|
||
def push_iq_block(self, samples):
|
||
return profiler.runcall(super().push_iq_block, samples)
|
||
|
||
def finish(self):
|
||
return profiler.runcall(super().finish)
|
||
|
||
result, receiver = measured_stream(sequence, ProfiledReceiver)
|
||
profiler.dump_stats(str(OUTPUT / "receiver_cpu.prof"))
|
||
report = StringIO()
|
||
stats = pstats.Stats(profiler, stream=report).strip_dirs()
|
||
stats.sort_stats("tottime").print_stats(30)
|
||
stats.sort_stats("cumtime").print_stats(30)
|
||
(OUTPUT / "receiver_cpu.txt").write_text(report.getvalue(), encoding="utf-8")
|
||
result["timing_includes_cprofile"] = True
|
||
write_result("cpu_profile.json", result)
|
||
print(report.getvalue(), flush=True)
|
||
|
||
|
||
def long_state_test(sequence=None, block_count=2048):
|
||
"""16,78 с распределённого шума, меняющаяся мощность, без истории в RX."""
|
||
from experiments import lab052_candidate_detection as detection
|
||
receiver = detection.StreamingReceiver()
|
||
random_generator = np.random.default_rng(52030)
|
||
checkpoints = []
|
||
accepted = 0
|
||
for number in range(block_count):
|
||
components = random_generator.standard_normal((2, receiver.block_samples), dtype=np.float32)
|
||
scale = (.002, .2, 20., .02)[(number // 257) % 4]
|
||
values = (components[0] + 1j * components[1]) * scale
|
||
accepted += len(receiver.push_iq_block(values))
|
||
assert receiver.buffer.size <= receiver.buffer_limit_samples
|
||
assert receiver.candidate_state_count == 0
|
||
if (number + 1) % 256 == 0:
|
||
checkpoints.append({"block": number + 1, "iq_bytes": receiver.buffer.nbytes,
|
||
"decimated_bytes": receiver.decimator.values.nbytes,
|
||
"retained_candidates": receiver.candidate_state_count,
|
||
"peak_eligible_candidates": receiver.peak_eligible_count,
|
||
"counter_keys": len(receiver.detection_statistics) + len(receiver.rejected_reasons),
|
||
**process_memory()})
|
||
accepted += len(receiver.finish())
|
||
result = {"blocks": block_count, "iq_seconds": block_count * receiver.block_samples / 8000000,
|
||
"background_scales": [.002, .2, 20, .02], "accepted_noise_packets": accepted,
|
||
"peak_retained_iq_bytes": receiver.peak_buffer_samples * 8,
|
||
"iq_limit_bytes": receiver.buffer_limit_samples * 8,
|
||
"retained_after_finish_bytes": receiver.buffer.nbytes + receiver.decimator.values.nbytes,
|
||
"constant_template_bytes": receiver.raw_reference.nbytes + receiver.matched_taps.nbytes + receiver.preamble_bits.nbytes,
|
||
"candidate_state_after_finish": receiver.candidate_state_count,
|
||
"peak_eligible_candidates": receiver.peak_eligible_count,
|
||
"peak_cluster_count": receiver.peak_cluster_count,
|
||
"detector": dict(receiver.detection_statistics), "strict_candidates": receiver.candidate_count,
|
||
"processing_seconds": receiver.processing_seconds, "checkpoints": checkpoints,
|
||
"process_memory": process_memory(), "rf_tx": 0}
|
||
write_result("memory.json", result)
|
||
print(json.dumps(result), flush=True)
|
||
|
||
|
||
def sensitivity_test(sequence=None):
|
||
"""До/после на одинаковом IQ около границы работоспособности FEC."""
|
||
from experiments import lab052_candidate_detection as detection
|
||
physical = previous.phy.lab045
|
||
payload = bytes(range(256))
|
||
plan = physical.build_transmission_plan(payload)
|
||
raw = previous.phy.build_parameterized_tx_cs8(plan, previous.phy.PhyMode(500000))
|
||
components = np.frombuffer(raw, np.int8).astype(np.float32)
|
||
clean = components[0::2] + 1j * components[1::2]
|
||
energy = float(np.sum(np.abs(clean) ** 2)) / plan.frame_symbol_count
|
||
rows = []
|
||
for ebn0 in (0, 1, 2, 3, 7):
|
||
for cfo in (-100, 10, 100):
|
||
for seed, scale in ((52031, .002), (52032, .2), (52033, 1.)):
|
||
random_generator = np.random.default_rng(seed)
|
||
sigma = np.sqrt(energy / (2 * 10 ** (ebn0 / 10)))
|
||
phase = .73 + (seed % 2) * np.pi
|
||
values = clean * np.exp(1j * (phase + 2*np.pi*cfo*np.arange(len(clean))/8000000))
|
||
values += sigma * (random_generator.normal(size=len(clean)) + 1j*random_generator.normal(size=len(clean)))
|
||
samples = np.concatenate((np.zeros(63217, np.complex64), (values * scale).astype(np.complex64),
|
||
np.zeros(350000, np.complex64)))
|
||
outcomes = []
|
||
for constructor in (previous.StreamingReceiver, detection.StreamingReceiver):
|
||
receiver = constructor()
|
||
packets = []
|
||
for start in range(0, len(samples), receiver.block_samples):
|
||
packets.extend(receiver.push_iq_block(samples[start:start + receiver.block_samples]))
|
||
packets.extend(receiver.finish())
|
||
# Оценка содержимого только после завершения независимого RX.
|
||
outcomes.append({"accepted": len(packets),
|
||
"payload_match": len(packets) == 1 and packets[0].payload == payload,
|
||
"reasons": dict(receiver.rejected_reasons)})
|
||
rows.append({"ebn0_db": ebn0, "cfo_hz": cfo, "seed": seed, "amplitude_scale": scale,
|
||
"baseline": outcomes[0], "optimized": outcomes[1],
|
||
"lost_baseline_packet": outcomes[0]["payload_match"] and not outcomes[1]["payload_match"]})
|
||
result = {"cases": rows, "count": len(rows),
|
||
"baseline_correct": sum(row["baseline"]["payload_match"] for row in rows),
|
||
"optimized_correct": sum(row["optimized"]["payload_match"] for row in rows),
|
||
"lost_baseline_packets": sum(row["lost_baseline_packet"] for row in rows), "rf_tx": 0}
|
||
write_result("sensitivity.json", result)
|
||
print(json.dumps({key: value for key, value in result.items() if key != "cases"}), flush=True)
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument("--stage", choices=("profile", "baseline", "rules", "stream", "cpu", "memory", "sensitivity"), default="profile")
|
||
arguments = parser.parse_args()
|
||
OUTPUT.mkdir(parents=True, exist_ok=True)
|
||
sequence = sequence_fixture()
|
||
print(f"Lab052 {arguments.stage}: START; RF TX=0", flush=True)
|
||
{"profile": profile_baseline, "baseline": unprofiled_baseline, "rules": compare_rules, "stream": optimized_stream,
|
||
"cpu": profile_cpu, "memory": long_state_test, "sensitivity": sensitivity_test}[arguments.stage](sequence)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|