tests/unit_tests/test_cli_smooth.py

0001 """pyrobird smooth: point columns found by name, a size guard before the
0002 interpolation, and archives that read back."""
0003
0004 import json
0005 import zipfile
0006
0007 import pytest
0008 from click.testing import CliRunner
0009
0010 from pyrobird.cli.smooth import smooth, apply_smoothing, count_interpolated_points, add_time_interpolation, \
0011     SmoothingTooLargeError
0012 from pyrobird.dex_utils import read_dex_json
0013
0014
0015 def trajectory_dex(point_columns, points):
0016     """A DEX 1.0 document with one event holding one PointTrajectory piece."""
0017     return {
0018         "type": "firebird-dex-json",
0019         "version": "1.0",
0020         "events": [{
0021             "id": 0,
0022             "pieces": [{
0023                 "name": "Tracks",
0024                 "type": "PointTrajectory",
0025                 "version": "1.0",
0026                 "count": len(points),
0027                 "columns": {},
0028                 "pointColumns": point_columns,
0029                 "points": points,
0030             }],
0031         }],
0032     }
0033
0034
0035 def write_json(path, document):
0036     path.write_text(json.dumps(document))
0037     return str(path)
0038
0039
0040 def test_time_column_found_by_name(tmp_path):
0041     """With t first, points are sorted and interpolated by column 0, and the cut reads x, y, z by name."""
0042     dex = trajectory_dex(["t", "x", "y", "z"], [[
0043         [1.0, 10.0, 0.0, 0.0],
0044         [0.0, 0.0, 0.0, 0.0],
0045     ]])
0046     smoothed = apply_smoothing(dex, step_time=0.25)
0047     points = smoothed["events"][0]["pieces"][0]["points"][0]
0048     assert [point[0] for point in points] == [0.0, 0.25, 0.5, 0.75, 1.0]
0049     assert [point[1] for point in points] == [0.0, 2.5, 5.0, 7.5, 10.0]
0050
0051
0052 def test_piece_without_time_column_is_cut_but_not_interpolated():
0053     dex = trajectory_dex(["x", "y", "z"], [[
0054         [0.0, 0.0, 0.0],
0055         [0.0, 0.0, 100.0],
0056         [9000.0, 0.0, 0.0],   # r = 9 m: outside every cut volume, the cut starts here
0057         [0.0, 0.0, 200.0],
0058     ]])
0059     points = apply_smoothing(dex, step_time=0.1)["events"][0]["pieces"][0]["points"][0]
0060     assert points == [[0.0, 0.0, 0.0], [0.0, 0.0, 100.0]]
0061
0062
0063 def test_predicted_count_matches_interpolation():
0064     points = [[0, 0, 0, 0.0], [0, 0, 1, 1.0], [0, 0, 2, 1.1], [0, 0, 3, 5.0]]
0065     predicted = count_interpolated_points(points, 0.2, 3)
0066     built = add_time_interpolation(points, 0.2, 3)
0067     assert abs(len(points) + predicted - len(built)) <= 2   # at most one point per gap
0068
0069
0070 def test_oversized_result_is_refused_before_it_is_built(tmp_path):
0071     dex = trajectory_dex(["x", "y", "z", "t"], [[[0, 0, 0, 0.0], [0, 0, 10, 1.0e6]]])
0072     with pytest.raises(SmoothingTooLargeError, match=r"grow 2 trajectory points to about 5000001"):
0073         apply_smoothing(dex, step_time=0.2, max_points=1000)
0074     # Nothing was changed
0075     assert dex["events"][0]["pieces"][0]["points"][0] == [[0, 0, 0, 0.0], [0, 0, 10, 1.0e6]]
0076
0077
0078 def test_cli_refuses_an_oversized_result_and_writes_nothing(tmp_path):
0079     source = write_json(tmp_path / "in.firebird.json",
0080                         trajectory_dex(["x", "y", "z", "t"], [[[0, 0, 0, 0.0], [0, 0, 10, 1000.0]]]))
0081     output = tmp_path / "out.firebird.json"
0082     result = CliRunner().invoke(smooth, [source, "-o", str(output), "--step-time", "0.25", "--max-points", "100"])
0083     assert result.exit_code == 1
0084     assert "more than --max-points=100" in result.output
0085     assert not output.exists()
0086
0087     # 0 disables the check
0088     result = CliRunner().invoke(smooth, [source, "-o", str(output), "--step-time", "0.25", "--max-points", "0"])
0089     assert result.exit_code == 0, result.output
0090     assert len(read_dex_json(str(output))["events"][0]["pieces"][0]["points"][0]) == 4001
0091
0092
0093 def test_cli_writes_a_zip_archive_that_reads_back(tmp_path):
0094     source = write_json(tmp_path / "in.firebird.json",
0095                         trajectory_dex(["x", "y", "z", "t"], [[[0, 0, 0, 0.0], [0, 0, 10, 1.0]]]))
0096     output = tmp_path / "out.firebird.zip"
0097     result = CliRunner().invoke(smooth, [source, "-o", str(output)])
0098     assert result.exit_code == 0, result.output
0099     assert zipfile.is_zipfile(output)
0100     assert len(read_dex_json(str(output))["events"][0]["pieces"][0]["points"][0]) == 6
0101
0102
0103 def test_cli_names_the_upgrade_for_a_dex_004_file(tmp_path):
0104     old = {"type": "firebird-dex-json", "version": "0.04",
0105            "events": [{"id": "event_0", "groups": [{"name": "Tracks", "type": "PointTrajectory"}]}]}
0106     source = write_json(tmp_path / "old.firebird.json", old)
0107     result = CliRunner().invoke(smooth, [source, "-o", str(tmp_path / "out.json")])
0108     assert result.exit_code != 0
0109     assert f"DEX version 0.04 file. Convert it once with: pyrobird upgrade {source}" in result.output