"""Installed-wheel contracts for role-based node entry points.""" from __future__ import annotations import json from pathlib import Path import pytest from .conftest import InstalledWheel ROOT = Path(__file__).resolve().parents[3] @pytest.mark.e2e def test_installed_node_entry_points_load_from_distribution_metadata( installed_wheel: InstalledWheel, tmp_path: Path ) -> None: manifest = ROOT / "ci" / "integration" / "coverage_manifest.json" result = installed_wheel.run( str(installed_wheel.python), "-c", "\n".join( ( "import json", "from rfgen.plugins import EntryPointRegistry", "from rfgen.nodes.source import Source", "from rfgen.nodes.transform import Transform", f"manifest = json.load(open({str(manifest)!r}))", "roles = {'rfgen.nodes.source': Source, 'rfgen.nodes.transform': Transform}", "loaded = {}", "for group, role in roles.items():", " registry = EntryPointRegistry(group); registry.discover()", " expected = manifest['covered'][group]", " assert set(expected) <= set(registry.available()), (group, expected, registry.available())", " for selector in expected:", " implementation = registry.load_class(selector)", " assert isinstance(implementation, type) and issubclass(implementation, role)", " assert implementation.params_type.model_json_schema()['type'] == 'object'", " loaded[group] = sorted(expected)", "print(json.dumps(loaded, sort_keys=True))", ) ), cwd=tmp_path, ) loaded = json.loads(result.stdout) assert loaded["rfgen.nodes.source"] assert loaded["rfgen.nodes.transform"] @pytest.mark.e2e def test_installed_node_roles_execute_a_real_typed_chain( installed_wheel: InstalledWheel, tmp_path: Path ) -> None: result = installed_wheel.run( str(installed_wheel.python), "-c", "\n".join( ( "import torch", "from rfgen.plugins import EntryPointRegistry", "from rfgen.graph import materialize", "from rfgen.graph.scene_plan import ScenePlan", "from rfgen.nodes import Ref, When", "from rfgen.nodes.plan import SceneFactsParams, SceneFactsPlan", "sources = EntryPointRegistry('rfgen.nodes.source'); sources.discover()", "transforms = EntryPointRegistry('rfgen.nodes.transform'); transforms.discover()", "scene = SceneFactsPlan(SceneFactsParams(sample_rate_hz=1_000_000.0, duration_s=64e-6, bandwidth_hz=800_000.0, carrier_hz=2.4e9, clock_id='scene', origin_s=0.0))", "scene_output = materialize(ScenePlan({'scene': scene}), sample_index=0, run_seed=0).node('scene').plan_outputs", "assert scene_output is not None", "source_type = sources.load_class('chirp_radar')", "source = source_type.from_params({'sample_rate_hz': 1_000_000.0, 'duration_s': 40e-6, 'bandwidth_hz': 200_000.0, 'pulse_duration_s': 20e-6, 'direction': 'up', 'clock_id': 'rx0'})", "source_output = source.evaluate({}, torch.Generator().manual_seed(1))", "carrier_type = transforms.load_class('carrier_translation')", "carrier = carrier_type.from_params({'sample_rate_hz': 1_000_000.0, 'frequency_offset_hz': 100_000.0, 'occupied_bandwidth_hz': 200_000.0, 'scene_bandwidth_hz': 800_000.0}, signal_type=source_output['signal'].vtype, refs={'signal': Ref('signal', 'source', 'signal'), 'source_facts': Ref('source_facts', 'source', 'facts')}, input_types={'signal': source_output['signal'].vtype, 'source_facts': source_output['facts'].vtype}, input_whens={'signal': When.REALIZED, 'source_facts': When.REALIZED})", "carrier_output = carrier.evaluate({'signal': source_output['signal'], 'source_facts': source_output['facts']}, torch.Generator().manual_seed(2))", "time_type = transforms.load_class('time_placement')", "time = time_type.from_params({'start_sample': 8}, signal_type=carrier_output['signal'].vtype, refs={'signal': Ref('signal', 'carrier', 'signal'), 'carrier_facts': Ref('carrier_facts', 'carrier', 'facts'), 'source_facts': Ref('source_facts', 'carrier', 'source_facts'), 'capture_grid': Ref('capture_grid', 'scene', 'sample_grid')}, input_types={'signal': carrier_output['signal'].vtype, 'carrier_facts': carrier_output['facts'].vtype, 'source_facts': carrier_output['source_facts'].vtype, 'capture_grid': scene_output['sample_grid'].vtype}, input_whens={'signal': When.REALIZED, 'carrier_facts': When.REALIZED, 'source_facts': When.REALIZED, 'capture_grid': When.PLAN})", "time_output = time.evaluate({'signal': carrier_output['signal'], 'carrier_facts': carrier_output['facts'], 'source_facts': carrier_output['source_facts'], 'capture_grid': scene_output['sample_grid']}, torch.Generator().manual_seed(3))", "evidence_type = transforms.load_class('placement_evidence')", "evidence = evidence_type.from_params({'scene_center_hz': 2.4e9, 'scene_bandwidth_hz': 800_000.0}, refs={'source_facts': Ref('source_facts', 'time', 'source_facts'), 'carrier_facts': Ref('carrier_facts', 'time', 'carrier_facts'), 'time_facts': Ref('time_facts', 'time', 'facts')}, input_types={'source_facts': time_output['source_facts'].vtype, 'carrier_facts': time_output['carrier_facts'].vtype, 'time_facts': time_output['facts'].vtype}, input_whens={'source_facts': When.REALIZED, 'carrier_facts': When.REALIZED, 'time_facts': When.REALIZED})", "evidence_output = evidence.evaluate({'source_facts': time_output['source_facts'], 'carrier_facts': time_output['carrier_facts'], 'time_facts': time_output['facts']}, torch.Generator().manual_seed(4))", "local_time = torch.arange(source_output['signal'].payload.numel(), dtype=torch.float64) / 1_000_000.0", "expected = (source_output['signal'].payload.to(torch.complex128) * torch.exp(2j * torch.pi * 100_000.0 * local_time)).to(torch.complex64)", "assert time_output['signal'].payload.dtype == torch.complex64", "assert time_output['signal'].payload.shape == (64,)", "assert time_output['signal'].vtype.qualifiers.sample_grid.clock.clock_id == 'scene'", "assert time_output['signal'].vtype.qualifiers.sample_grid.origin == 0.0", "assert torch.isfinite(time_output['signal'].payload).all()", "assert torch.count_nonzero(time_output['signal'].payload[:8]) == 0", "assert torch.count_nonzero(time_output['signal'].payload[48:]) == 0", "assert torch.equal(time_output['signal'].payload[8:48], expected)", "phase_step = carrier_output['signal'].payload[1] / source_output['signal'].payload[1]", "assert phase_step.imag > 0.0 and not torch.isclose(phase_step.imag, torch.tensor(0.0))", "observed_offset_hz = float(torch.angle(phase_step)) * 1_000_000.0 / (2.0 * float(torch.pi))", "assert evidence_output['facts'].payload['start_sample'] == 8", "assert evidence_output['facts'].payload['end_sample'] == 28", "assert evidence_output['facts'].payload['frequency_offset_hz'] == 100_000.0", "assert abs(observed_offset_hz - evidence_output['facts'].payload['frequency_offset_hz']) < 0.02", "assert evidence_output['facts'].payload['carrier_hz'] == 2_400_100_000.0", ) ), cwd=tmp_path, ) assert result.returncode == 0 @pytest.mark.e2e def test_installed_auxiliary_entry_points_load_and_execute( installed_wheel: InstalledWheel, tmp_path: Path ) -> None: result = installed_wheel.run( str(installed_wheel.python), "-c", "\n".join( ( "from rfgen.plugins import EntryPointRegistry", "metrics = EntryPointRegistry('rfgen.metrics'); metrics.discover()", "assert callable(metrics.load_class('paes'))", ) ), cwd=tmp_path, ) assert result.returncode == 0