| """Independent mathematical and geometry guards for the overnight study.""" |
|
|
| import numpy as np |
| import pytest |
| from ans_delta import decoded_chunks, encode |
| from dense_waveforms import completed, same_solver_inputs, spectral_derivative |
| from pipeline.femm_utils.build_model import simplify_collinear_segments |
|
|
|
|
| @pytest.mark.parametrize("count", [96, 192, 384, 768, 1536]) |
| def test_linkage_derivative_matches_analytic_periodic_signal(count): |
| period = 2 * np.pi / 7 |
| angle = (np.arange(count) + 0.23) * period / count |
| values = np.column_stack( |
| [np.sin(7 * angle), np.cos(21 * angle), 0.1 * np.sin(84 * angle)] |
| ) |
| expected = np.column_stack( |
| [7 * np.cos(7 * angle), -21 * np.sin(21 * angle), 8.4 * np.cos(84 * angle)] |
| ) |
| np.testing.assert_allclose( |
| spectral_derivative(values, period), expected, atol=2e-10, rtol=1e-10 |
| ) |
|
|
|
|
| @pytest.mark.parametrize("junction", [False, True]) |
| def test_compact_femm_keeps_exact_straight_geometry_and_junctions(junction): |
| nodes = np.array( |
| [[0.0, 0.0], [0.5, 0.0], [1.0, 0.0], [1.0, 1.0], [0.0, 1.0], [0.5, 0.5]] |
| ) |
| segments = [(0, 1), (1, 2), (2, 3), (3, 4), (4, 0)] + ([(1, 5)] if junction else []) |
| source = { |
| "nodes_mm": nodes, |
| "segments": segments, |
| "used_nodes": list(range(len(nodes))), |
| "boundary_edges": {"bore": [], "outer": []}, |
| } |
| result = simplify_collinear_segments(source) |
| if junction: |
| assert (0, 1) in result["segments"] and (1, 2) in result["segments"] |
| else: |
| assert (0, 2) in result["segments"] and result[ |
| "collinear_vertices_removed" |
| ] == 1 |
| length = lambda edges: sum(np.linalg.norm(nodes[a] - nodes[b]) for a, b in edges) |
| assert length(result["segments"]) == pytest.approx(length(segments), abs=1e-14) |
| assert source["segments"] == segments |
|
|
|
|
| def test_prescribed_boundary_vertices_are_never_removed(): |
| source = { |
| "nodes_mm": np.array([[0.0, 0.0], [0.5, 0.0], [1.0, 0.0]]), |
| "segments": [(0, 1), (1, 2)], |
| "used_nodes": [0, 1, 2], |
| "boundary_edges": {"bore": [(0, 1), (1, 2)], "outer": []}, |
| } |
| assert simplify_collinear_segments(source)["segments"] == source["segments"] |
|
|
|
|
| @pytest.mark.parametrize( |
| "before,after", |
| [ |
| (b"", b""), |
| (b"x\ny\n", b"x\nz\n"), |
| (b"abc", b"abcd\nmore"), |
| (b"a\nb\nc\n", b"a\n"), |
| (b"", b"one\ntwo\n"), |
| (b"same\n", b"same\n"), |
| ], |
| ) |
| def test_answer_archive_reconstructs_exact_bytes(before, after): |
| from io import BytesIO |
|
|
| delta = BytesIO() |
| encode(BytesIO(before), BytesIO(after), delta) |
| delta.seek(0) |
| assert b"".join(decoded_chunks(BytesIO(before), delta)) == after |
|
|
|
|
| def test_femm_intermediate_readback_is_not_a_completed_dense_sample(tmp_path): |
| import json |
|
|
| directory = tmp_path / "tasks/zero-0000" |
| directory.mkdir(parents=True) |
| (directory / "result.json").write_text( |
| json.dumps( |
| { |
| "status": "accepted", |
| "execution_sha256": "execution", |
| "request_sha256": "request", |
| "outer_rotor_torque_nm": 1.0, |
| } |
| ) |
| ) |
| assert ( |
| completed( |
| tmp_path, {"id": "zero-0000", "request_sha256": "request"}, "execution" |
| ) |
| is None |
| ) |
|
|
|
|
| def test_archiver_waits_for_final_driver_publication(tmp_path): |
| import json |
|
|
| from ans_delta import convert |
|
|
| state = tmp_path / "femm/tasks/loaded-0000" |
| state.mkdir(parents=True) |
| (state / "result.json").write_text(json.dumps({"status": "accepted"})) |
| (state / "base.ans.gz").write_bytes(b"not yet a finished gzip") |
| assert not convert(tmp_path, state) |
| assert (state / "base.ans.gz").exists() |
|
|
|
|
| def test_completed_data_renders_all_requested_comparison_charts(tmp_path): |
| import json |
|
|
| from dense_waveform_report import write_report |
| from dense_waveforms import task |
|
|
| count = 96 |
| period = 2 * np.pi / 7 |
| angles = (np.arange(count) + 0.5) * period / count |
| patterns = np.array([[1.0, -1.0, 0.0]] * count) |
| plan = { |
| "identity": "test-plan", |
| "max_angles": count, |
| "target_angles": count, |
| "angular_levels": [count], |
| "electrical_period_mechanical_rad": period, |
| "angles_mechanical_rad": angles.tolist(), |
| "currents_abc_a": patterns.tolist(), |
| "derivative_step_rad": period * 1e-4, |
| "n_theta": 1344, |
| "harmonics": 630, |
| "boundary_subdivisions": 64, |
| "max_area_mm2": 0.0125, |
| "femm_mesh_size_mm": 0.125, |
| "femm_precision": 1e-10, |
| "targets": { |
| "kt_relative": 0.0025, |
| "ke_relative": 0.0025, |
| "torque_extrema_absolute_nm": 0.0001, |
| "derivative_relative": 0.0025, |
| }, |
| } |
| (tmp_path / "plan.json").write_text(json.dumps(plan)) |
| for solver in ("jax", "femm"): |
| directory = tmp_path / solver |
| directory.mkdir() |
| (directory / "execution.json").write_text( |
| json.dumps({"identity": "test-execution"}) |
| ) |
| (directory / "status.json").write_text( |
| json.dumps( |
| { |
| "phase": "complete", |
| "current_grid": count, |
| "accepted_tasks": 2 * count, |
| } |
| ) |
| ) |
| for index, angle in enumerate(angles): |
| for mode in ("zero", "loaded"): |
| request = task(plan, mode, index) |
| target = directory / "tasks" / request["id"] |
| target.mkdir(parents=True) |
| data = { |
| **request, |
| "execution_sha256": "test-execution", |
| "status": "accepted", |
| "torque_nm": float( |
| 0.002 * np.sin(84 * angle) - (0.015 if mode == "loaded" else 0) |
| ), |
| "phase_linkage_wb_turn": [ |
| float(0.001 * np.cos(7 * angle + phase)) |
| for phase in (0, 2 * np.pi / 3, 4 * np.pi / 3) |
| ], |
| "saved_artifacts": {}, |
| } |
| (target / "result.json").write_text(json.dumps(data)) |
| summary = write_report(tmp_path) |
| page = (tmp_path / "report.html").read_text() |
| for label in ( |
| "NEW JAX and FEMM scalar values", |
| "NEW FEMM and JAX sample overlays", |
| "NEW back-EMF waveform comparison", |
| "Loaded torque", |
| "Zero-current torque", |
| ): |
| assert label in page |
| with np.load(tmp_path / "waveforms.npz") as arrays: |
| assert arrays["jax/torque"].shape == (2, count) |
| assert arrays["femm/phase_ke"].shape == (count, 3) |
| assert summary["complete"] |
| assert not summary["resolution_checks_passed"] |
|
|
|
|
| def test_checkpoint_compatibility_separates_solver_physics_from_sampling_policy(): |
| from copy import deepcopy |
|
|
| original = { |
| "reference_sha256": "reference", |
| "numerical_source_sha256": "source", |
| "n_theta": 1344, |
| "boundary_subdivisions": 64, |
| "harmonics": 630, |
| "max_area_mm2": 0.0125, |
| "controls": {"tolerance": 1e-9}, |
| "angles_mechanical_rad": [0.1, 0.2], |
| "currents_abc_a": [[1, -1, 0], [1, 0, -1]], |
| "derivative_step_rad": 1e-4, |
| "femm_mesh_size_mm": 0.125, |
| "femm_precision": 1e-10, |
| "target_angles": 384, |
| } |
| changed = deepcopy(original) |
| changed["target_angles"] = 768 |
| assert same_solver_inputs(original, changed, "jax") and same_solver_inputs( |
| original, changed, "femm" |
| ) |
| changed["femm_mesh_size_mm"] = 0.0625 |
| assert same_solver_inputs(original, changed, "jax") and not same_solver_inputs( |
| original, changed, "femm" |
| ) |
| changed = deepcopy(original) |
| changed["max_area_mm2"] /= 2 |
| assert not same_solver_inputs(original, changed, "jax") and same_solver_inputs( |
| original, changed, "femm" |
| ) |
| changed = deepcopy(original) |
| changed["numerical_source_sha256"] = "different physics" |
| assert not same_solver_inputs(original, changed, "jax") and not same_solver_inputs( |
| original, changed, "femm" |
| ) |
|
|
