from __future__ import annotations import inspect import json from pathlib import Path import subprocess import sys from types import SimpleNamespace import numpy as np from PIL import Image import pytest from experiments import lab047_jpeg_object_transport as lab047 def _test_source(path: Path) -> Path: y, x = np.mgrid[0:96, 0:128] pixels = np.stack( ((3 * x + y) % 256, (x + 5 * y) % 256, (7 * x + 2 * y) % 256), axis=-1, ).astype(np.uint8) Image.fromarray(pixels, mode="RGB").save(path, format="PNG") return path def test_fixed_jpeg_preparation_is_reproducible(tmp_path: Path) -> None: source = _test_source(tmp_path / "source.png") first = lab047.prepare_jpeg(source, tmp_path / "first.jpg") second = lab047.prepare_jpeg(source, tmp_path / "second.jpg") assert first.sha256 == second.sha256 assert first.file_bytes == second.file_bytes assert (first.width, first.height, first.mode, first.format) == (320, 240, "RGB", "JPEG") def test_selected_photo_produces_target_size_reproducibly(tmp_path: Path) -> None: first = lab047.prepare_jpeg(lab047.SOURCE_IMAGE_PATH, tmp_path / "first.jpg") second = lab047.prepare_jpeg(lab047.SOURCE_IMAGE_PATH, tmp_path / "second.jpg") assert 8 * 1024 <= first.file_bytes <= 15 * 1024 assert first.sha256 == second.sha256 def test_jpeg_passes_full_software_transport(tmp_path: Path) -> None: source = _test_source(tmp_path / "source.png") jpeg_path = tmp_path / "tx.jpg" lab047.prepare_jpeg(source, jpeg_path) transmitted = jpeg_path.read_bytes() result = lab047.software_roundtrip_jpeg(transmitted) assert result.object_complete is True assert result.crc_pass == len(lab047.lab046.build_object_frames(transmitted, lab047.OBJECT_ID)) assert result.crc_fail == 0 assert result.sha_match is True assert result.byte_errors == 0 assert result.bit_errors == 0 assert result.recovered_bytes == transmitted recovered_path = tmp_path / "recovered.jpg" recovered_path.write_bytes(result.recovered_bytes or b"") assert lab047.validate_jpeg(recovered_path).format == "JPEG" def test_frame_is_not_accepted_without_crc_pass(monkeypatch: pytest.MonkeyPatch) -> None: data = bytes(245) monkeypatch.setattr( lab047.lab045, "analyze_post_preamble_soft", lambda soft: SimpleNamespace( analysis={"crc": lab047.lab045.CRC_FAIL}, payload_bytes=bytes(lab047.lab045.PAYLOAD_SIZE_BYTES), ), ) result = lab047.software_roundtrip_jpeg(data) assert result.crc_pass == 0 assert result.crc_fail == 2 assert result.valid_fragments == 0 assert result.object_complete is False def test_accounting_uses_actual_object_size() -> None: accounting = lab047.transport_accounting(bytes(8_724)) assert accounting.fragment_count == 36 assert accounting.rf_frame_count == 36 assert accounting.samples_per_frame == 2_306_400 assert accounting.tx_sample_count == 83_030_400 assert accounting.tx_file_bytes == 166_060_800 assert accounting.estimated_airtime_seconds == pytest.approx(10.3788) def test_dynamic_threshold_is_q999_plus_four_local_peak_mad( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: scores = np.linspace(0.001, 0.02, 20, dtype=np.float32) peaks = np.array([0.01, 0.02, 0.03, 0.04, 0.05], dtype=np.float32) monkeypatch.setattr( lab047.offline, "_background_peak_scores", lambda capture_path, interval: (scores, peaks, 0.25), ) rule, statistics = lab047.calibrate_dynamic_candidate_rule(tmp_path / "unused.cs8") expected = lab047.offline.candidate_threshold_from_local_peaks( peaks, quantile=0.999, mad_multiplier=4.0, ) assert rule.threshold == expected assert rule.local_peak_quantile == 0.999 assert statistics.interval == lab047.CALIBRATION_INTERVAL assert statistics.local_peak_mad == pytest.approx(0.01) def test_independent_receiver_api_has_no_tx_oracles() -> None: parameters = set(inspect.signature(lab047.run_independent_capture_recovery).parameters) assert parameters == {"capture_path", "output_directory"} assert not parameters & { "tx_offsets", "expected_frame_count", "tx_image", "tx_packets", "tx_coded_bits", } assert all(value is False for value in lab047.receiver_oracle_declaration().values()) reprocess_parameters = set( inspect.signature(lab047.run_saved_capture_streaming_recovery).parameters ) assert reprocess_parameters == { "capture_path", "original_result_path", "output_directory", } assert "expected_frame_count" not in reprocess_parameters def test_historical_strict_threshold_and_fixed_hardware_are_unchanged() -> None: assert lab047.lab046b.IQ_PREAMBLE_THRESHOLD == 0.65 assert lab047.hardware_parameters() == { "master_tx_serial": "000000000000000097c063dc323b97d3", "slave_rx_serial": "000000000000000097c063dc2e5327d3", "frequency_hz": 1_400_000_000, "sample_rate_hz": 8_000_000, "symbol_rate": 20_000, "samples_per_symbol": 400, "digital_carrier_hz": 50_000, "tx_vga_db": 10, "tx_rf_amp": False, "tx_bias_tee": False, "rx_lna_db": 24, "rx_vga_db": 24, "rx_rf_amp": False, "rx_bias_tee": False, "master_clkout_to_slave_clkin": True, "slave_clkout": False, "expected_reg29": 0x26, } def test_hardware_function_is_closed_without_new_authorization(tmp_path: Path) -> None: called = False def runner(command: list[str]) -> subprocess.CompletedProcess[str]: nonlocal called called = True return subprocess.CompletedProcess(command, 0, "", "") with pytest.raises(PermissionError, match="отдельного разрешения"): lab047.run_hardware_capture_once(tmp_path, runner=runner) assert called is False def test_hardware_cli_requires_all_three_keys(tmp_path: Path) -> None: script = Path(lab047.__file__).resolve() completed = subprocess.run( [sys.executable, str(script), "--hardware", "--output-directory", str(tmp_path)], capture_output=True, text=True, encoding="utf-8", errors="replace", check=False, ) assert completed.returncode == 2 assert "RF TX = 0" in completed.stdout def test_tx_reference_diagnostics_require_saved_complete_verdict(tmp_path: Path) -> None: verdict_path = tmp_path / "independent_result.json" verdict_path.write_text( json.dumps({"completed": True, "object_verdict": {"object_complete": False}}), encoding="utf-8", ) with pytest.raises(RuntimeError, match="не разрешает"): lab047.diagnose_jpeg_after_independent_verdict( verdict_path, tmp_path / "recovered.jpg", tmp_path / "tx.jpg", ) def _original_incomplete_result(capture_path: Path) -> dict: background_common = { "correlation_score_count": 100, "local_peak_count": 10, "correlation_median": 0.006, "correlation_mad": 0.002, "local_peak_quantiles": { "0.99": 0.018, "0.999": 0.020810436457395554, "1.0": 0.025, }, "candidate_count": 0, "candidates_per_second": 0.0, "elapsed_seconds": 0.1, "local_peak_median": 0.008, "local_peak_mad": 0.002538509201258421, } return { "completed": False, "capture_path": str(capture_path), "rule": { "method": "pre-TX local-peak Q99.9 + 4 * local-peak MAD", "local_peak_quantile": 0.999, "threshold": lab047.FIXED_HARDWARE_CANDIDATE_THRESHOLD, }, "calibration": { **background_common, "interval": { "name": "PRE_TX_BACKGROUND_LAB047", "start_sample": 0, "end_sample": 4_000_000, }, }, "control": { **background_common, "interval": { "name": "POST_TX_BACKGROUND_LAB047", "start_sample": 96_000_000, "end_sample": 120_000_000, }, }, "coarse": { "termination_reason": "MAX_CANDIDATES_REACHED", "max_candidates_reached": True, }, } def test_saved_hardware_threshold_is_loaded_without_change(tmp_path: Path) -> None: capture_path = tmp_path / "capture.cs8" capture_path.write_bytes(bytes(20)) result_path = tmp_path / "independent_result.json" result_path.write_text( json.dumps(_original_incomplete_result(capture_path)), encoding="utf-8", ) rule, calibration, control = lab047.load_fixed_hardware_candidate_rule( capture_path, result_path, ) assert rule.threshold == 0.030964473262429237 assert calibration.interval == lab047.CALIBRATION_INTERVAL assert control.interval == lab047.CONTROL_INTERVAL def test_changed_saved_hardware_threshold_is_rejected(tmp_path: Path) -> None: capture_path = tmp_path / "capture.cs8" capture_path.write_bytes(bytes(20)) value = _original_incomplete_result(capture_path) value["rule"]["threshold"] = 0.031 result_path = tmp_path / "independent_result.json" result_path.write_text(json.dumps(value), encoding="utf-8") with pytest.raises(RuntimeError, match="threshold был изменён"): lab047.load_fixed_hardware_candidate_rule(capture_path, result_path) def test_streaming_limit_is_not_derived_from_expected_jpeg_frames() -> None: assert lab047.offline.MAX_CANDIDATES == 100_000 source = inspect.getsource(lab047.offline.stream_process_capture) assert "36" not in source assert "expected_frame_count" not in source def test_reprocessing_persists_verdict_before_tx_diagnostics() -> None: source = inspect.getsource(lab047.main) independent_position = source.index("run_saved_capture_streaming_recovery") diagnostic_position = source.index("diagnose_hardware_jpeg_after_independent_verdict") assert independent_position < diagnostic_position def test_completion_accounting_can_be_corrected_without_decoder_rerun( tmp_path: Path, ) -> None: capture_path = tmp_path / "capture.cs8" capture_path.write_bytes(bytes(1_000)) original_path = tmp_path / "reprocess.json" original = { "completed": False, "capture_path": str(capture_path), "capture_samples": 500, "rule": {"threshold": lab047.FIXED_HARDWARE_CANDIDATE_THRESHOLD}, "coarse": { "termination_reason": "END_OF_CAPTURE", "max_candidates_reached": False, "blocks_processed": 1, "processed_until_sample": 400, }, "strict": { "termination_reason": "END_OF_CAPTURE", "max_search_iterations_reached": False, "max_valid_frames_reached": False, }, } original_path.write_text(json.dumps(original), encoding="utf-8") corrected_path = tmp_path / "corrected.json" corrected = lab047.correct_saved_streaming_completion( original_path, corrected_path, capture_path, ) assert corrected["completed"] is True assert corrected["coarse"]["processed_until_sample"] == 500 assert corrected["completion_metadata_correction"]["full_decoder_rerun"] is False assert json.loads(original_path.read_text(encoding="utf-8"))["completed"] is False def test_completion_accounting_refuses_different_capture_with_same_size( tmp_path: Path, ) -> None: saved_capture = tmp_path / "saved_capture.cs8" other_capture = tmp_path / "other_capture.cs8" saved_capture.write_bytes(bytes(1_000)) other_capture.write_bytes(bytes([1]) * 1_000) original_path = tmp_path / "reprocess.json" original_path.write_text( json.dumps( { "completed": False, "capture_path": str(saved_capture), "capture_samples": 500, "rule": {"threshold": lab047.FIXED_HARDWARE_CANDIDATE_THRESHOLD}, "coarse": { "termination_reason": "END_OF_CAPTURE", "max_candidates_reached": False, "blocks_processed": 1, "processed_until_sample": 400, }, "strict": { "termination_reason": "END_OF_CAPTURE", "max_search_iterations_reached": False, "max_valid_frames_reached": False, }, } ), encoding="utf-8", ) corrected_path = tmp_path / "corrected.json" with pytest.raises(RuntimeError, match="не доказывает полное покрытие"): lab047.correct_saved_streaming_completion( original_path, corrected_path, other_capture, ) assert not corrected_path.exists() def test_finalize_mode_corrects_result_before_opening_tx_reference() -> None: source = inspect.getsource(lab047.main) correction_position = source.index("correct_saved_streaming_completion") diagnostic_position = source.index("diagnose_saved_reprocess_after_corrected_verdict") assert correction_position < diagnostic_position def test_saved_diagnostics_refuse_changed_threshold_before_capture_access( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: corrected_path = tmp_path / "corrected.json" corrected_path.write_text( json.dumps( { "completed": True, "rule": {"threshold": lab047.FIXED_HARDWARE_CANDIDATE_THRESHOLD}, "object_verdict": {"object_complete": True}, "completion_metadata_correction": { "full_decoder_rerun": False, "threshold_changed": True, }, } ), encoding="utf-8", ) def fail_if_capture_is_touched(*_args: object, **_kwargs: object) -> None: raise AssertionError("RX-запись не должна открываться") monkeypatch.setattr(lab047.offline, "_validate_capture", fail_if_capture_is_touched) with pytest.raises(RuntimeError, match="Исправленный полный verdict не подтверждён"): lab047.diagnose_saved_reprocess_after_corrected_verdict( corrected_path, tmp_path / "capture.cs8", tmp_path / "recovered.jpg", tmp_path / "tx.jpg", ) def test_hardware_diagnostics_refuse_incomplete_in_memory_verdict(tmp_path: Path) -> None: independent = object.__new__(lab047.offline.IndependentRecoveryResult) object.__setattr__(independent, "completed", False) with pytest.raises(RuntimeError, match="не открывает TX reference"): lab047.diagnose_hardware_jpeg_after_independent_verdict( independent, tmp_path / "tx.jpg", tmp_path / "recovered.jpg", ) def test_functional_checks() -> None: lab047.run_functional_tests()