From 7747f5fdc1397cbbfe1b2f8450fbd5c7083c3743 Mon Sep 17 00:00:00 2001 From: kaydcraig <11017961@uvu.edu> Date: Thu, 30 Jul 2026 12:15:09 -0600 Subject: [PATCH 1/6] Add session_logger test coverage for untested methods - log_connection: device/port/baud, optional firmware and radc_available - log_accepted_reading: entry content and stats increment - log_shot: optional fields (angle_source, club_angle_deg, club_path_deg, spin_axis_deg, pipeline_ms) included/omitted correctly - log_camera_data: full entry and None angles - log_config_change: default and custom source - log_sim_send: full entry and None values fallback - log_sim_status: connected/retrying states - log_sim_player: entry content - log_iq_reading: full entry, cfar flag - log_iq_blocks: block count, empty list - log_trigger_event: accepted/rejected, stats tracking - Properties: session_id and raw_path before/after start - init_session_logger: instance creation, global logger registration --- tests/test_session_logger_new.py | 504 +++++++++++++++++++++++++++++++ 1 file changed, 504 insertions(+) create mode 100644 tests/test_session_logger_new.py diff --git a/tests/test_session_logger_new.py b/tests/test_session_logger_new.py new file mode 100644 index 000000000..fb2633a23 --- /dev/null +++ b/tests/test_session_logger_new.py @@ -0,0 +1,504 @@ +"""Additional tests for session_logger uncovered paths.""" + +import json + +from openflight.ops243 import Direction, SpeedReading +from openflight.session_logger import SessionLogger + + +def _start(tmp_path, **kwargs): + logger = SessionLogger(log_dir=tmp_path, enabled=True) + logger.start_session(mode="rolling-buffer", trigger_type="sound", **kwargs) + return logger + + +def _last_entry(logger): + lines = logger.session_path.read_text().strip().split("\n") + return json.loads(lines[-1]) + + +class TestLogConnection: + def test_writes_device_port_baud(self, tmp_path): + logger = _start(tmp_path) + logger.log_connection(device="ops243", port="/dev/ttyACM0", baud=115200) + entry = _last_entry(logger) + assert entry["type"] == "connection" + assert entry["device"] == "ops243" + assert entry["port"] == "/dev/ttyACM0" + assert entry["baud"] == 115200 + + def test_includes_firmware_when_provided(self, tmp_path): + logger = _start(tmp_path) + logger.log_connection(device="ops243", port="/dev/ttyACM0", firmware="OPS243-A v2.1.0") + entry = _last_entry(logger) + assert entry["firmware"] == "OPS243-A v2.1.0" + + def test_omits_firmware_when_not_provided(self, tmp_path): + logger = _start(tmp_path) + logger.log_connection(device="ops243", port="/dev/ttyACM0") + entry = _last_entry(logger) + assert "firmware" not in entry + + def test_includes_radc_available_when_provided(self, tmp_path): + logger = _start(tmp_path) + logger.log_connection(device="ops243", port="/dev/ttyACM0", radc_available=True) + entry = _last_entry(logger) + assert entry["radc_available"] is True + + def test_omits_radc_available_when_none(self, tmp_path): + logger = _start(tmp_path) + logger.log_connection(device="ops243", port="/dev/ttyACM0", radc_available=None) + entry = _last_entry(logger) + assert "radc_available" not in entry + + def test_skipped_when_disabled(self, tmp_path): + logger = SessionLogger(log_dir=tmp_path, enabled=False) + logger.log_connection(device="ops243", port="/dev/ttyACM0") + assert logger.session_path is None + + +class TestLogAcceptedReading: + def _reading(self, speed=105.0, direction=Direction.OUTBOUND, magnitude=0.85): + return SpeedReading(speed=speed, direction=direction, magnitude=magnitude) + + def test_writes_reading_entry(self, tmp_path): + logger = _start(tmp_path) + logger.log_accepted_reading(self._reading()) + entry = _last_entry(logger) + assert entry["type"] == "reading_accepted" + assert entry["speed"] == 105.0 + assert entry["magnitude"] == 0.85 + + def test_increments_readings_accepted_stat(self, tmp_path): + logger = _start(tmp_path) + logger.log_accepted_reading(self._reading()) + logger.log_accepted_reading(self._reading(speed=50.0)) + assert logger.stats["readings_accepted"] == 2 + + def test_skipped_when_disabled(self, tmp_path): + logger = SessionLogger(log_dir=tmp_path, enabled=False) + logger.log_accepted_reading(self._reading()) + assert logger.stats["readings_accepted"] == 0 + + +class TestLogShotOptionalFields: + def _minimal_shot(self, logger): + logger.log_shot( + ball_speed_mph=130.0, + club_speed_mph=90.0, + smash_factor=1.44, + estimated_carry_yards=220.0, + club="driver", + peak_magnitude=0.9, + readings_count=5, + ) + + def test_angle_source_included_when_provided(self, tmp_path): + logger = _start(tmp_path) + logger.log_shot( + ball_speed_mph=130.0, + club_speed_mph=90.0, + smash_factor=1.44, + estimated_carry_yards=220.0, + club="driver", + peak_magnitude=0.9, + readings_count=5, + angle_source="kld7", + ) + entry = _last_entry(logger) + assert entry["angle_source"] == "kld7" + + def test_angle_source_omitted_when_none(self, tmp_path): + logger = _start(tmp_path) + self._minimal_shot(logger) + entry = _last_entry(logger) + assert "angle_source" not in entry + + def test_club_angle_deg_included_when_provided(self, tmp_path): + logger = _start(tmp_path) + logger.log_shot( + ball_speed_mph=130.0, + club_speed_mph=90.0, + smash_factor=1.44, + estimated_carry_yards=220.0, + club="driver", + peak_magnitude=0.9, + readings_count=5, + club_angle_deg=-2.5, + ) + entry = _last_entry(logger) + assert entry["club_angle_deg"] == -2.5 + + def test_club_path_deg_included_when_provided(self, tmp_path): + logger = _start(tmp_path) + logger.log_shot( + ball_speed_mph=130.0, + club_speed_mph=90.0, + smash_factor=1.44, + estimated_carry_yards=220.0, + club="driver", + peak_magnitude=0.9, + readings_count=5, + club_path_deg=3.1, + ) + entry = _last_entry(logger) + assert entry["club_path_deg"] == 3.1 + + def test_spin_axis_deg_included_when_provided(self, tmp_path): + logger = _start(tmp_path) + logger.log_shot( + ball_speed_mph=130.0, + club_speed_mph=90.0, + smash_factor=1.44, + estimated_carry_yards=220.0, + club="driver", + peak_magnitude=0.9, + readings_count=5, + spin_axis_deg=-5.0, + ) + entry = _last_entry(logger) + assert entry["spin_axis_deg"] == -5.0 + + def test_pipeline_ms_included_when_provided(self, tmp_path): + logger = _start(tmp_path) + timing = {"spin": 12.4, "carry": 0.3} + logger.log_shot( + ball_speed_mph=130.0, + club_speed_mph=90.0, + smash_factor=1.44, + estimated_carry_yards=220.0, + club="driver", + peak_magnitude=0.9, + readings_count=5, + pipeline_ms=timing, + ) + entry = _last_entry(logger) + assert entry["pipeline_ms"] == timing + + def test_skipped_when_disabled(self, tmp_path): + logger = SessionLogger(log_dir=tmp_path, enabled=False) + self._minimal_shot(logger) + assert logger.stats["shots_detected"] == 0 + + +class TestLogCameraData: + def test_writes_camera_entry(self, tmp_path): + logger = _start(tmp_path) + logger.log_camera_data( + shot_number=1, + launch_angle_vertical=12.5, + launch_angle_horizontal=-1.0, + confidence=0.9, + positions_tracked=4, + launch_detected=True, + ) + entry = _last_entry(logger) + assert entry["type"] == "shot_camera" + assert entry["shot_number"] == 1 + assert entry["launch_angle_vertical"] == 12.5 + assert entry["launch_angle_horizontal"] == -1.0 + assert entry["confidence"] == 0.9 + assert entry["positions_tracked"] == 4 + assert entry["launch_detected"] is True + + def test_accepts_none_angles(self, tmp_path): + logger = _start(tmp_path) + logger.log_camera_data( + shot_number=1, + launch_angle_vertical=None, + launch_angle_horizontal=None, + confidence=None, + positions_tracked=0, + launch_detected=False, + ) + entry = _last_entry(logger) + assert entry["launch_angle_vertical"] is None + assert entry["launch_detected"] is False + + def test_skipped_when_disabled(self, tmp_path): + logger = SessionLogger(log_dir=tmp_path, enabled=False) + logger.log_camera_data( + shot_number=1, + launch_angle_vertical=12.5, + launch_angle_horizontal=None, + confidence=None, + positions_tracked=0, + launch_detected=False, + ) + assert logger.session_path is None + + +class TestLogConfigChange: + def test_writes_config_entry_with_default_source(self, tmp_path): + logger = _start(tmp_path) + logger.log_config_change({"min_speed": 35, "dc_mask": 15}) + entry = _last_entry(logger) + assert entry["type"] == "config_change" + assert entry["config"]["min_speed"] == 35 + assert entry["source"] == "user" + + def test_writes_custom_source(self, tmp_path): + logger = _start(tmp_path) + logger.log_config_change({"mode": "rolling-buffer"}, source="setup_script") + entry = _last_entry(logger) + assert entry["source"] == "setup_script" + + def test_skipped_when_disabled(self, tmp_path): + logger = SessionLogger(log_dir=tmp_path, enabled=False) + logger.log_config_change({"min_speed": 35}) + assert logger.session_path is None + + +class TestLogSimSend: + def test_writes_sim_send_entry(self, tmp_path): + logger = _start(tmp_path) + logger.log_sim_send( + target="gspro", + shot_number=3, + provenance={"ball_speed": "measured", "spin": "estimated"}, + values={"ball_speed": 130.0, "carry": 220.0}, + ) + entry = _last_entry(logger) + assert entry["type"] == "sim_send" + assert entry["target"] == "gspro" + assert entry["shot_number"] == 3 + assert entry["provenance"]["spin"] == "estimated" + assert entry["values"]["carry"] == 220.0 + + def test_values_defaults_to_empty_dict_when_none(self, tmp_path): + logger = _start(tmp_path) + logger.log_sim_send( + target="gspro", + shot_number=1, + provenance={}, + values=None, + ) + entry = _last_entry(logger) + assert entry["values"] == {} + + def test_skipped_when_disabled(self, tmp_path): + logger = SessionLogger(log_dir=tmp_path, enabled=False) + logger.log_sim_send(target="gspro", shot_number=1, provenance={}) + assert logger.session_path is None + + +class TestLogSimStatus: + def test_writes_sim_status_entry(self, tmp_path): + logger = _start(tmp_path) + logger.log_sim_status( + target="gspro", + state="connected", + host="192.168.1.60", + port=921, + message="ready", + attempt=1, + next_retry_in_s=0.0, + ) + entry = _last_entry(logger) + assert entry["type"] == "sim_status" + assert entry["target"] == "gspro" + assert entry["state"] == "connected" + assert entry["host"] == "192.168.1.60" + assert entry["port"] == 921 + assert entry["message"] == "ready" + assert entry["attempt"] == 1 + assert entry["next_retry_in_s"] == 0.0 + + def test_writes_retry_state(self, tmp_path): + logger = _start(tmp_path) + logger.log_sim_status( + target="gspro", + state="retrying", + attempt=3, + next_retry_in_s=5.0, + ) + entry = _last_entry(logger) + assert entry["state"] == "retrying" + assert entry["attempt"] == 3 + assert entry["next_retry_in_s"] == 5.0 + + def test_skipped_when_disabled(self, tmp_path): + logger = SessionLogger(log_dir=tmp_path, enabled=False) + logger.log_sim_status(target="gspro", state="connected") + assert logger.session_path is None + + +class TestLogSimPlayer: + def test_writes_sim_player_entry(self, tmp_path): + logger = _start(tmp_path) + logger.log_sim_player(target="gspro", handed="right", club="7-iron") + entry = _last_entry(logger) + assert entry["type"] == "sim_player" + assert entry["target"] == "gspro" + assert entry["handed"] == "right" + assert entry["club"] == "7-iron" + + def test_skipped_when_disabled(self, tmp_path): + logger = SessionLogger(log_dir=tmp_path, enabled=False) + logger.log_sim_player(target="gspro", handed="right", club="driver") + assert logger.session_path is None + + +class TestLogIQReading: + def test_writes_iq_reading_entry(self, tmp_path): + logger = _start(tmp_path) + logger.log_iq_reading( + speed_mph=105.0, + direction="outbound", + magnitude=0.85, + snr=12.3, + peak_bin=42, + cfar_validated=True, + block_count=8, + ) + entry = _last_entry(logger) + assert entry["type"] == "iq_reading" + assert entry["speed_mph"] == 105.0 + assert entry["direction"] == "outbound" + assert entry["magnitude"] == 0.85 + assert entry["snr"] == 12.3 + assert entry["peak_bin"] == 42 + assert entry["cfar_validated"] is True + assert entry["block_count"] == 8 + + def test_cfar_not_validated(self, tmp_path): + logger = _start(tmp_path) + logger.log_iq_reading( + speed_mph=30.0, + direction="inbound", + magnitude=0.2, + snr=2.1, + peak_bin=10, + cfar_validated=False, + block_count=4, + ) + entry = _last_entry(logger) + assert entry["cfar_validated"] is False + + def test_skipped_when_disabled(self, tmp_path): + logger = SessionLogger(log_dir=tmp_path, enabled=False) + logger.log_iq_reading( + speed_mph=100.0, + direction="outbound", + magnitude=0.8, + snr=10.0, + peak_bin=40, + cfar_validated=True, + block_count=6, + ) + assert logger.session_path is None + + +class TestLogIQBlocks: + def test_writes_iq_blocks_entry(self, tmp_path): + logger = _start(tmp_path) + blocks = [ + {"i_samples": [1, 2, 3], "q_samples": [4, 5, 6], "timestamp": 1.0}, + {"i_samples": [7, 8, 9], "q_samples": [10, 11, 12], "timestamp": 2.0}, + ] + logger.log_iq_blocks(shot_number=1, blocks=blocks) + entry = _last_entry(logger) + assert entry["type"] == "iq_blocks" + assert entry["shot_number"] == 1 + assert entry["block_count"] == 2 + assert len(entry["blocks"]) == 2 + + def test_empty_blocks(self, tmp_path): + logger = _start(tmp_path) + logger.log_iq_blocks(shot_number=2, blocks=[]) + entry = _last_entry(logger) + assert entry["block_count"] == 0 + assert entry["blocks"] == [] + + def test_skipped_when_disabled(self, tmp_path): + logger = SessionLogger(log_dir=tmp_path, enabled=False) + logger.log_iq_blocks(shot_number=1, blocks=[]) + assert logger.session_path is None + + +class TestLogTriggerEvent: + def test_writes_accepted_trigger_event(self, tmp_path): + logger = _start(tmp_path) + logger.log_trigger_event( + trigger_type="sound-gpio", + accepted=True, + reason=None, + peak_speed_mph=145.0, + readings_count=12, + latency_ms=8.5, + ) + entry = _last_entry(logger) + assert entry["type"] == "trigger_event" + assert entry["trigger_type"] == "sound-gpio" + assert entry["accepted"] is True + assert entry["peak_speed_mph"] == 145.0 + assert entry["readings_count"] == 12 + assert entry["latency_ms"] == 8.5 + + def test_writes_rejected_trigger_event(self, tmp_path): + logger = _start(tmp_path) + logger.log_trigger_event( + trigger_type="sound-gpio", + accepted=False, + reason="no_outbound_speed", + ) + entry = _last_entry(logger) + assert entry["accepted"] is False + assert entry["reason"] == "no_outbound_speed" + + def test_increments_stats(self, tmp_path): + logger = _start(tmp_path) + logger.log_trigger_event(trigger_type="sound-gpio", accepted=True) + logger.log_trigger_event(trigger_type="sound-gpio", accepted=False) + logger.log_trigger_event(trigger_type="sound-gpio", accepted=False) + assert logger.stats["triggers_total"] == 3 + assert logger.stats["triggers_accepted"] == 1 + assert logger.stats["triggers_rejected"] == 2 + + def test_skipped_when_disabled(self, tmp_path): + logger = SessionLogger(log_dir=tmp_path, enabled=False) + logger.log_trigger_event(trigger_type="sound-gpio", accepted=True) + assert logger.session_path is None + + +class TestProperties: + def test_session_id_returns_timestamp_string(self, tmp_path): + logger = _start(tmp_path) + sid = logger.session_id + assert sid is not None + assert len(sid) == 15 # YYYYMMDD_HHMMSS + + def test_session_id_none_before_start(self, tmp_path): + logger = SessionLogger(log_dir=tmp_path, enabled=True) + assert logger.session_id is None + + def test_raw_path_points_to_log_file(self, tmp_path): + logger = _start(tmp_path) + assert logger.raw_path is not None + assert logger.raw_path.suffix == ".log" + assert "radar_raw" in logger.raw_path.name + + def test_raw_path_none_before_start(self, tmp_path): + logger = SessionLogger(log_dir=tmp_path, enabled=True) + assert logger.raw_path is None + + +class TestInitSessionLogger: + def test_returns_session_logger_instance(self, tmp_path): + from openflight.session_logger import init_session_logger + + logger = init_session_logger(log_dir=tmp_path, location="test", enabled=True) + assert isinstance(logger, SessionLogger) + + def test_sets_global_logger(self, tmp_path): + from openflight.session_logger import get_session_logger, init_session_logger + + logger = init_session_logger(log_dir=tmp_path, enabled=True) + assert get_session_logger() is logger + + def test_disabled_logger_creates_instance(self, tmp_path): + from openflight.session_logger import init_session_logger + + logger = init_session_logger(log_dir=tmp_path, enabled=False) + assert isinstance(logger, SessionLogger) + assert not logger.enabled From 0449f7e1bae99df5150ac8e21bb7772e91e8b4e8 Mon Sep 17 00:00:00 2001 From: kaydcraig <11017961@uvu.edu> Date: Thu, 30 Jul 2026 12:19:18 -0600 Subject: [PATCH 2/6] Move session_logger tests into existing test module Consolidate new tests into tests/test_session_logger.py rather than keeping them in a separate file. --- tests/test_session_logger.py | 497 +++++++++++++++++++++++++++++- tests/test_session_logger_new.py | 504 ------------------------------- 2 files changed, 496 insertions(+), 505 deletions(-) delete mode 100644 tests/test_session_logger_new.py diff --git a/tests/test_session_logger.py b/tests/test_session_logger.py index 922cf0798..c508ac12c 100644 --- a/tests/test_session_logger.py +++ b/tests/test_session_logger.py @@ -4,7 +4,8 @@ from openflight import session_logger as session_logger_module from openflight.kld7.radc import RADC_PAYLOAD_BYTES -from openflight.session_logger import SessionLogger, log_session_error +from openflight.ops243 import Direction, SpeedReading +from openflight.session_logger import SessionLogger, init_session_logger, log_session_error class TestLogError: @@ -607,3 +608,497 @@ def test_session_uuid_is_unique_per_session(self, tmp_path): first = self._start_entry(tmp_path / "a") second = self._start_entry(tmp_path / "b") assert first["session_uuid"] != second["session_uuid"] + + +def _start(tmp_path, **kwargs): + logger = SessionLogger(log_dir=tmp_path, enabled=True) + logger.start_session(mode="rolling-buffer", trigger_type="sound", **kwargs) + return logger + + +def _last_entry(logger): + lines = logger.session_path.read_text().strip().split("\n") + return json.loads(lines[-1]) + + +class TestLogConnection: + def test_writes_device_port_baud(self, tmp_path): + logger = _start(tmp_path) + logger.log_connection(device="ops243", port="/dev/ttyACM0", baud=115200) + entry = _last_entry(logger) + assert entry["type"] == "connection" + assert entry["device"] == "ops243" + assert entry["port"] == "/dev/ttyACM0" + assert entry["baud"] == 115200 + + def test_includes_firmware_when_provided(self, tmp_path): + logger = _start(tmp_path) + logger.log_connection(device="ops243", port="/dev/ttyACM0", firmware="OPS243-A v2.1.0") + entry = _last_entry(logger) + assert entry["firmware"] == "OPS243-A v2.1.0" + + def test_omits_firmware_when_not_provided(self, tmp_path): + logger = _start(tmp_path) + logger.log_connection(device="ops243", port="/dev/ttyACM0") + entry = _last_entry(logger) + assert "firmware" not in entry + + def test_includes_radc_available_when_provided(self, tmp_path): + logger = _start(tmp_path) + logger.log_connection(device="ops243", port="/dev/ttyACM0", radc_available=True) + entry = _last_entry(logger) + assert entry["radc_available"] is True + + def test_omits_radc_available_when_none(self, tmp_path): + logger = _start(tmp_path) + logger.log_connection(device="ops243", port="/dev/ttyACM0", radc_available=None) + entry = _last_entry(logger) + assert "radc_available" not in entry + + def test_skipped_when_disabled(self, tmp_path): + logger = SessionLogger(log_dir=tmp_path, enabled=False) + logger.log_connection(device="ops243", port="/dev/ttyACM0") + assert logger.session_path is None + + +class TestLogAcceptedReading: + def _reading(self, speed=105.0, direction=Direction.OUTBOUND, magnitude=0.85): + return SpeedReading(speed=speed, direction=direction, magnitude=magnitude) + + def test_writes_reading_entry(self, tmp_path): + logger = _start(tmp_path) + logger.log_accepted_reading(self._reading()) + entry = _last_entry(logger) + assert entry["type"] == "reading_accepted" + assert entry["speed"] == 105.0 + assert entry["magnitude"] == 0.85 + + def test_increments_readings_accepted_stat(self, tmp_path): + logger = _start(tmp_path) + logger.log_accepted_reading(self._reading()) + logger.log_accepted_reading(self._reading(speed=50.0)) + assert logger.stats["readings_accepted"] == 2 + + def test_skipped_when_disabled(self, tmp_path): + logger = SessionLogger(log_dir=tmp_path, enabled=False) + logger.log_accepted_reading(self._reading()) + assert logger.stats["readings_accepted"] == 0 + + +class TestLogShotOptionalFields: + def _minimal_shot(self, logger): + logger.log_shot( + ball_speed_mph=130.0, + club_speed_mph=90.0, + smash_factor=1.44, + estimated_carry_yards=220.0, + club="driver", + peak_magnitude=0.9, + readings_count=5, + ) + + def test_angle_source_included_when_provided(self, tmp_path): + logger = _start(tmp_path) + logger.log_shot( + ball_speed_mph=130.0, + club_speed_mph=90.0, + smash_factor=1.44, + estimated_carry_yards=220.0, + club="driver", + peak_magnitude=0.9, + readings_count=5, + angle_source="kld7", + ) + entry = _last_entry(logger) + assert entry["angle_source"] == "kld7" + + def test_angle_source_omitted_when_none(self, tmp_path): + logger = _start(tmp_path) + self._minimal_shot(logger) + entry = _last_entry(logger) + assert "angle_source" not in entry + + def test_club_angle_deg_included_when_provided(self, tmp_path): + logger = _start(tmp_path) + logger.log_shot( + ball_speed_mph=130.0, + club_speed_mph=90.0, + smash_factor=1.44, + estimated_carry_yards=220.0, + club="driver", + peak_magnitude=0.9, + readings_count=5, + club_angle_deg=-2.5, + ) + entry = _last_entry(logger) + assert entry["club_angle_deg"] == -2.5 + + def test_club_path_deg_included_when_provided(self, tmp_path): + logger = _start(tmp_path) + logger.log_shot( + ball_speed_mph=130.0, + club_speed_mph=90.0, + smash_factor=1.44, + estimated_carry_yards=220.0, + club="driver", + peak_magnitude=0.9, + readings_count=5, + club_path_deg=3.1, + ) + entry = _last_entry(logger) + assert entry["club_path_deg"] == 3.1 + + def test_spin_axis_deg_included_when_provided(self, tmp_path): + logger = _start(tmp_path) + logger.log_shot( + ball_speed_mph=130.0, + club_speed_mph=90.0, + smash_factor=1.44, + estimated_carry_yards=220.0, + club="driver", + peak_magnitude=0.9, + readings_count=5, + spin_axis_deg=-5.0, + ) + entry = _last_entry(logger) + assert entry["spin_axis_deg"] == -5.0 + + def test_pipeline_ms_included_when_provided(self, tmp_path): + logger = _start(tmp_path) + timing = {"spin": 12.4, "carry": 0.3} + logger.log_shot( + ball_speed_mph=130.0, + club_speed_mph=90.0, + smash_factor=1.44, + estimated_carry_yards=220.0, + club="driver", + peak_magnitude=0.9, + readings_count=5, + pipeline_ms=timing, + ) + entry = _last_entry(logger) + assert entry["pipeline_ms"] == timing + + def test_skipped_when_disabled(self, tmp_path): + logger = SessionLogger(log_dir=tmp_path, enabled=False) + self._minimal_shot(logger) + assert logger.stats["shots_detected"] == 0 + + +class TestLogCameraData: + def test_writes_camera_entry(self, tmp_path): + logger = _start(tmp_path) + logger.log_camera_data( + shot_number=1, + launch_angle_vertical=12.5, + launch_angle_horizontal=-1.0, + confidence=0.9, + positions_tracked=4, + launch_detected=True, + ) + entry = _last_entry(logger) + assert entry["type"] == "shot_camera" + assert entry["shot_number"] == 1 + assert entry["launch_angle_vertical"] == 12.5 + assert entry["launch_angle_horizontal"] == -1.0 + assert entry["confidence"] == 0.9 + assert entry["positions_tracked"] == 4 + assert entry["launch_detected"] is True + + def test_accepts_none_angles(self, tmp_path): + logger = _start(tmp_path) + logger.log_camera_data( + shot_number=1, + launch_angle_vertical=None, + launch_angle_horizontal=None, + confidence=None, + positions_tracked=0, + launch_detected=False, + ) + entry = _last_entry(logger) + assert entry["launch_angle_vertical"] is None + assert entry["launch_detected"] is False + + def test_skipped_when_disabled(self, tmp_path): + logger = SessionLogger(log_dir=tmp_path, enabled=False) + logger.log_camera_data( + shot_number=1, + launch_angle_vertical=12.5, + launch_angle_horizontal=None, + confidence=None, + positions_tracked=0, + launch_detected=False, + ) + assert logger.session_path is None + + +class TestLogConfigChange: + def test_writes_config_entry_with_default_source(self, tmp_path): + logger = _start(tmp_path) + logger.log_config_change({"min_speed": 35, "dc_mask": 15}) + entry = _last_entry(logger) + assert entry["type"] == "config_change" + assert entry["config"]["min_speed"] == 35 + assert entry["source"] == "user" + + def test_writes_custom_source(self, tmp_path): + logger = _start(tmp_path) + logger.log_config_change({"mode": "rolling-buffer"}, source="setup_script") + entry = _last_entry(logger) + assert entry["source"] == "setup_script" + + def test_skipped_when_disabled(self, tmp_path): + logger = SessionLogger(log_dir=tmp_path, enabled=False) + logger.log_config_change({"min_speed": 35}) + assert logger.session_path is None + + +class TestLogSimSend: + def test_writes_sim_send_entry(self, tmp_path): + logger = _start(tmp_path) + logger.log_sim_send( + target="gspro", + shot_number=3, + provenance={"ball_speed": "measured", "spin": "estimated"}, + values={"ball_speed": 130.0, "carry": 220.0}, + ) + entry = _last_entry(logger) + assert entry["type"] == "sim_send" + assert entry["target"] == "gspro" + assert entry["shot_number"] == 3 + assert entry["provenance"]["spin"] == "estimated" + assert entry["values"]["carry"] == 220.0 + + def test_values_defaults_to_empty_dict_when_none(self, tmp_path): + logger = _start(tmp_path) + logger.log_sim_send( + target="gspro", + shot_number=1, + provenance={}, + values=None, + ) + entry = _last_entry(logger) + assert entry["values"] == {} + + def test_skipped_when_disabled(self, tmp_path): + logger = SessionLogger(log_dir=tmp_path, enabled=False) + logger.log_sim_send(target="gspro", shot_number=1, provenance={}) + assert logger.session_path is None + + +class TestLogSimStatus: + def test_writes_sim_status_entry(self, tmp_path): + logger = _start(tmp_path) + logger.log_sim_status( + target="gspro", + state="connected", + host="192.168.1.60", + port=921, + message="ready", + attempt=1, + next_retry_in_s=0.0, + ) + entry = _last_entry(logger) + assert entry["type"] == "sim_status" + assert entry["target"] == "gspro" + assert entry["state"] == "connected" + assert entry["host"] == "192.168.1.60" + assert entry["port"] == 921 + assert entry["message"] == "ready" + assert entry["attempt"] == 1 + assert entry["next_retry_in_s"] == 0.0 + + def test_writes_retry_state(self, tmp_path): + logger = _start(tmp_path) + logger.log_sim_status( + target="gspro", + state="retrying", + attempt=3, + next_retry_in_s=5.0, + ) + entry = _last_entry(logger) + assert entry["state"] == "retrying" + assert entry["attempt"] == 3 + assert entry["next_retry_in_s"] == 5.0 + + def test_skipped_when_disabled(self, tmp_path): + logger = SessionLogger(log_dir=tmp_path, enabled=False) + logger.log_sim_status(target="gspro", state="connected") + assert logger.session_path is None + + +class TestLogSimPlayer: + def test_writes_sim_player_entry(self, tmp_path): + logger = _start(tmp_path) + logger.log_sim_player(target="gspro", handed="right", club="7-iron") + entry = _last_entry(logger) + assert entry["type"] == "sim_player" + assert entry["target"] == "gspro" + assert entry["handed"] == "right" + assert entry["club"] == "7-iron" + + def test_skipped_when_disabled(self, tmp_path): + logger = SessionLogger(log_dir=tmp_path, enabled=False) + logger.log_sim_player(target="gspro", handed="right", club="driver") + assert logger.session_path is None + + +class TestLogIQReading: + def test_writes_iq_reading_entry(self, tmp_path): + logger = _start(tmp_path) + logger.log_iq_reading( + speed_mph=105.0, + direction="outbound", + magnitude=0.85, + snr=12.3, + peak_bin=42, + cfar_validated=True, + block_count=8, + ) + entry = _last_entry(logger) + assert entry["type"] == "iq_reading" + assert entry["speed_mph"] == 105.0 + assert entry["direction"] == "outbound" + assert entry["magnitude"] == 0.85 + assert entry["snr"] == 12.3 + assert entry["peak_bin"] == 42 + assert entry["cfar_validated"] is True + assert entry["block_count"] == 8 + + def test_cfar_not_validated(self, tmp_path): + logger = _start(tmp_path) + logger.log_iq_reading( + speed_mph=30.0, + direction="inbound", + magnitude=0.2, + snr=2.1, + peak_bin=10, + cfar_validated=False, + block_count=4, + ) + entry = _last_entry(logger) + assert entry["cfar_validated"] is False + + def test_skipped_when_disabled(self, tmp_path): + logger = SessionLogger(log_dir=tmp_path, enabled=False) + logger.log_iq_reading( + speed_mph=100.0, + direction="outbound", + magnitude=0.8, + snr=10.0, + peak_bin=40, + cfar_validated=True, + block_count=6, + ) + assert logger.session_path is None + + +class TestLogIQBlocks: + def test_writes_iq_blocks_entry(self, tmp_path): + logger = _start(tmp_path) + blocks = [ + {"i_samples": [1, 2, 3], "q_samples": [4, 5, 6], "timestamp": 1.0}, + {"i_samples": [7, 8, 9], "q_samples": [10, 11, 12], "timestamp": 2.0}, + ] + logger.log_iq_blocks(shot_number=1, blocks=blocks) + entry = _last_entry(logger) + assert entry["type"] == "iq_blocks" + assert entry["shot_number"] == 1 + assert entry["block_count"] == 2 + assert len(entry["blocks"]) == 2 + + def test_empty_blocks(self, tmp_path): + logger = _start(tmp_path) + logger.log_iq_blocks(shot_number=2, blocks=[]) + entry = _last_entry(logger) + assert entry["block_count"] == 0 + assert entry["blocks"] == [] + + def test_skipped_when_disabled(self, tmp_path): + logger = SessionLogger(log_dir=tmp_path, enabled=False) + logger.log_iq_blocks(shot_number=1, blocks=[]) + assert logger.session_path is None + + +class TestLogTriggerEvent: + def test_writes_accepted_trigger_event(self, tmp_path): + logger = _start(tmp_path) + logger.log_trigger_event( + trigger_type="sound-gpio", + accepted=True, + reason=None, + peak_speed_mph=145.0, + readings_count=12, + latency_ms=8.5, + ) + entry = _last_entry(logger) + assert entry["type"] == "trigger_event" + assert entry["trigger_type"] == "sound-gpio" + assert entry["accepted"] is True + assert entry["peak_speed_mph"] == 145.0 + assert entry["readings_count"] == 12 + assert entry["latency_ms"] == 8.5 + + def test_writes_rejected_trigger_event(self, tmp_path): + logger = _start(tmp_path) + logger.log_trigger_event( + trigger_type="sound-gpio", + accepted=False, + reason="no_outbound_speed", + ) + entry = _last_entry(logger) + assert entry["accepted"] is False + assert entry["reason"] == "no_outbound_speed" + + def test_increments_stats(self, tmp_path): + logger = _start(tmp_path) + logger.log_trigger_event(trigger_type="sound-gpio", accepted=True) + logger.log_trigger_event(trigger_type="sound-gpio", accepted=False) + logger.log_trigger_event(trigger_type="sound-gpio", accepted=False) + assert logger.stats["triggers_total"] == 3 + assert logger.stats["triggers_accepted"] == 1 + assert logger.stats["triggers_rejected"] == 2 + + def test_skipped_when_disabled(self, tmp_path): + logger = SessionLogger(log_dir=tmp_path, enabled=False) + logger.log_trigger_event(trigger_type="sound-gpio", accepted=True) + assert logger.session_path is None + + +class TestProperties: + def test_session_id_returns_timestamp_string(self, tmp_path): + logger = _start(tmp_path) + sid = logger.session_id + assert sid is not None + assert len(sid) == 15 # YYYYMMDD_HHMMSS + + def test_session_id_none_before_start(self, tmp_path): + logger = SessionLogger(log_dir=tmp_path, enabled=True) + assert logger.session_id is None + + def test_raw_path_points_to_log_file(self, tmp_path): + logger = _start(tmp_path) + assert logger.raw_path is not None + assert logger.raw_path.suffix == ".log" + assert "radar_raw" in logger.raw_path.name + + def test_raw_path_none_before_start(self, tmp_path): + logger = SessionLogger(log_dir=tmp_path, enabled=True) + assert logger.raw_path is None + + +class TestInitSessionLogger: + def test_returns_session_logger_instance(self, tmp_path): + logger = init_session_logger(log_dir=tmp_path, location="test", enabled=True) + assert isinstance(logger, SessionLogger) + + def test_sets_global_logger(self, tmp_path): + from openflight.session_logger import get_session_logger + + logger = init_session_logger(log_dir=tmp_path, enabled=True) + assert get_session_logger() is logger + + def test_disabled_logger_creates_instance(self, tmp_path): + logger = init_session_logger(log_dir=tmp_path, enabled=False) + assert isinstance(logger, SessionLogger) + assert not logger.enabled diff --git a/tests/test_session_logger_new.py b/tests/test_session_logger_new.py deleted file mode 100644 index fb2633a23..000000000 --- a/tests/test_session_logger_new.py +++ /dev/null @@ -1,504 +0,0 @@ -"""Additional tests for session_logger uncovered paths.""" - -import json - -from openflight.ops243 import Direction, SpeedReading -from openflight.session_logger import SessionLogger - - -def _start(tmp_path, **kwargs): - logger = SessionLogger(log_dir=tmp_path, enabled=True) - logger.start_session(mode="rolling-buffer", trigger_type="sound", **kwargs) - return logger - - -def _last_entry(logger): - lines = logger.session_path.read_text().strip().split("\n") - return json.loads(lines[-1]) - - -class TestLogConnection: - def test_writes_device_port_baud(self, tmp_path): - logger = _start(tmp_path) - logger.log_connection(device="ops243", port="/dev/ttyACM0", baud=115200) - entry = _last_entry(logger) - assert entry["type"] == "connection" - assert entry["device"] == "ops243" - assert entry["port"] == "/dev/ttyACM0" - assert entry["baud"] == 115200 - - def test_includes_firmware_when_provided(self, tmp_path): - logger = _start(tmp_path) - logger.log_connection(device="ops243", port="/dev/ttyACM0", firmware="OPS243-A v2.1.0") - entry = _last_entry(logger) - assert entry["firmware"] == "OPS243-A v2.1.0" - - def test_omits_firmware_when_not_provided(self, tmp_path): - logger = _start(tmp_path) - logger.log_connection(device="ops243", port="/dev/ttyACM0") - entry = _last_entry(logger) - assert "firmware" not in entry - - def test_includes_radc_available_when_provided(self, tmp_path): - logger = _start(tmp_path) - logger.log_connection(device="ops243", port="/dev/ttyACM0", radc_available=True) - entry = _last_entry(logger) - assert entry["radc_available"] is True - - def test_omits_radc_available_when_none(self, tmp_path): - logger = _start(tmp_path) - logger.log_connection(device="ops243", port="/dev/ttyACM0", radc_available=None) - entry = _last_entry(logger) - assert "radc_available" not in entry - - def test_skipped_when_disabled(self, tmp_path): - logger = SessionLogger(log_dir=tmp_path, enabled=False) - logger.log_connection(device="ops243", port="/dev/ttyACM0") - assert logger.session_path is None - - -class TestLogAcceptedReading: - def _reading(self, speed=105.0, direction=Direction.OUTBOUND, magnitude=0.85): - return SpeedReading(speed=speed, direction=direction, magnitude=magnitude) - - def test_writes_reading_entry(self, tmp_path): - logger = _start(tmp_path) - logger.log_accepted_reading(self._reading()) - entry = _last_entry(logger) - assert entry["type"] == "reading_accepted" - assert entry["speed"] == 105.0 - assert entry["magnitude"] == 0.85 - - def test_increments_readings_accepted_stat(self, tmp_path): - logger = _start(tmp_path) - logger.log_accepted_reading(self._reading()) - logger.log_accepted_reading(self._reading(speed=50.0)) - assert logger.stats["readings_accepted"] == 2 - - def test_skipped_when_disabled(self, tmp_path): - logger = SessionLogger(log_dir=tmp_path, enabled=False) - logger.log_accepted_reading(self._reading()) - assert logger.stats["readings_accepted"] == 0 - - -class TestLogShotOptionalFields: - def _minimal_shot(self, logger): - logger.log_shot( - ball_speed_mph=130.0, - club_speed_mph=90.0, - smash_factor=1.44, - estimated_carry_yards=220.0, - club="driver", - peak_magnitude=0.9, - readings_count=5, - ) - - def test_angle_source_included_when_provided(self, tmp_path): - logger = _start(tmp_path) - logger.log_shot( - ball_speed_mph=130.0, - club_speed_mph=90.0, - smash_factor=1.44, - estimated_carry_yards=220.0, - club="driver", - peak_magnitude=0.9, - readings_count=5, - angle_source="kld7", - ) - entry = _last_entry(logger) - assert entry["angle_source"] == "kld7" - - def test_angle_source_omitted_when_none(self, tmp_path): - logger = _start(tmp_path) - self._minimal_shot(logger) - entry = _last_entry(logger) - assert "angle_source" not in entry - - def test_club_angle_deg_included_when_provided(self, tmp_path): - logger = _start(tmp_path) - logger.log_shot( - ball_speed_mph=130.0, - club_speed_mph=90.0, - smash_factor=1.44, - estimated_carry_yards=220.0, - club="driver", - peak_magnitude=0.9, - readings_count=5, - club_angle_deg=-2.5, - ) - entry = _last_entry(logger) - assert entry["club_angle_deg"] == -2.5 - - def test_club_path_deg_included_when_provided(self, tmp_path): - logger = _start(tmp_path) - logger.log_shot( - ball_speed_mph=130.0, - club_speed_mph=90.0, - smash_factor=1.44, - estimated_carry_yards=220.0, - club="driver", - peak_magnitude=0.9, - readings_count=5, - club_path_deg=3.1, - ) - entry = _last_entry(logger) - assert entry["club_path_deg"] == 3.1 - - def test_spin_axis_deg_included_when_provided(self, tmp_path): - logger = _start(tmp_path) - logger.log_shot( - ball_speed_mph=130.0, - club_speed_mph=90.0, - smash_factor=1.44, - estimated_carry_yards=220.0, - club="driver", - peak_magnitude=0.9, - readings_count=5, - spin_axis_deg=-5.0, - ) - entry = _last_entry(logger) - assert entry["spin_axis_deg"] == -5.0 - - def test_pipeline_ms_included_when_provided(self, tmp_path): - logger = _start(tmp_path) - timing = {"spin": 12.4, "carry": 0.3} - logger.log_shot( - ball_speed_mph=130.0, - club_speed_mph=90.0, - smash_factor=1.44, - estimated_carry_yards=220.0, - club="driver", - peak_magnitude=0.9, - readings_count=5, - pipeline_ms=timing, - ) - entry = _last_entry(logger) - assert entry["pipeline_ms"] == timing - - def test_skipped_when_disabled(self, tmp_path): - logger = SessionLogger(log_dir=tmp_path, enabled=False) - self._minimal_shot(logger) - assert logger.stats["shots_detected"] == 0 - - -class TestLogCameraData: - def test_writes_camera_entry(self, tmp_path): - logger = _start(tmp_path) - logger.log_camera_data( - shot_number=1, - launch_angle_vertical=12.5, - launch_angle_horizontal=-1.0, - confidence=0.9, - positions_tracked=4, - launch_detected=True, - ) - entry = _last_entry(logger) - assert entry["type"] == "shot_camera" - assert entry["shot_number"] == 1 - assert entry["launch_angle_vertical"] == 12.5 - assert entry["launch_angle_horizontal"] == -1.0 - assert entry["confidence"] == 0.9 - assert entry["positions_tracked"] == 4 - assert entry["launch_detected"] is True - - def test_accepts_none_angles(self, tmp_path): - logger = _start(tmp_path) - logger.log_camera_data( - shot_number=1, - launch_angle_vertical=None, - launch_angle_horizontal=None, - confidence=None, - positions_tracked=0, - launch_detected=False, - ) - entry = _last_entry(logger) - assert entry["launch_angle_vertical"] is None - assert entry["launch_detected"] is False - - def test_skipped_when_disabled(self, tmp_path): - logger = SessionLogger(log_dir=tmp_path, enabled=False) - logger.log_camera_data( - shot_number=1, - launch_angle_vertical=12.5, - launch_angle_horizontal=None, - confidence=None, - positions_tracked=0, - launch_detected=False, - ) - assert logger.session_path is None - - -class TestLogConfigChange: - def test_writes_config_entry_with_default_source(self, tmp_path): - logger = _start(tmp_path) - logger.log_config_change({"min_speed": 35, "dc_mask": 15}) - entry = _last_entry(logger) - assert entry["type"] == "config_change" - assert entry["config"]["min_speed"] == 35 - assert entry["source"] == "user" - - def test_writes_custom_source(self, tmp_path): - logger = _start(tmp_path) - logger.log_config_change({"mode": "rolling-buffer"}, source="setup_script") - entry = _last_entry(logger) - assert entry["source"] == "setup_script" - - def test_skipped_when_disabled(self, tmp_path): - logger = SessionLogger(log_dir=tmp_path, enabled=False) - logger.log_config_change({"min_speed": 35}) - assert logger.session_path is None - - -class TestLogSimSend: - def test_writes_sim_send_entry(self, tmp_path): - logger = _start(tmp_path) - logger.log_sim_send( - target="gspro", - shot_number=3, - provenance={"ball_speed": "measured", "spin": "estimated"}, - values={"ball_speed": 130.0, "carry": 220.0}, - ) - entry = _last_entry(logger) - assert entry["type"] == "sim_send" - assert entry["target"] == "gspro" - assert entry["shot_number"] == 3 - assert entry["provenance"]["spin"] == "estimated" - assert entry["values"]["carry"] == 220.0 - - def test_values_defaults_to_empty_dict_when_none(self, tmp_path): - logger = _start(tmp_path) - logger.log_sim_send( - target="gspro", - shot_number=1, - provenance={}, - values=None, - ) - entry = _last_entry(logger) - assert entry["values"] == {} - - def test_skipped_when_disabled(self, tmp_path): - logger = SessionLogger(log_dir=tmp_path, enabled=False) - logger.log_sim_send(target="gspro", shot_number=1, provenance={}) - assert logger.session_path is None - - -class TestLogSimStatus: - def test_writes_sim_status_entry(self, tmp_path): - logger = _start(tmp_path) - logger.log_sim_status( - target="gspro", - state="connected", - host="192.168.1.60", - port=921, - message="ready", - attempt=1, - next_retry_in_s=0.0, - ) - entry = _last_entry(logger) - assert entry["type"] == "sim_status" - assert entry["target"] == "gspro" - assert entry["state"] == "connected" - assert entry["host"] == "192.168.1.60" - assert entry["port"] == 921 - assert entry["message"] == "ready" - assert entry["attempt"] == 1 - assert entry["next_retry_in_s"] == 0.0 - - def test_writes_retry_state(self, tmp_path): - logger = _start(tmp_path) - logger.log_sim_status( - target="gspro", - state="retrying", - attempt=3, - next_retry_in_s=5.0, - ) - entry = _last_entry(logger) - assert entry["state"] == "retrying" - assert entry["attempt"] == 3 - assert entry["next_retry_in_s"] == 5.0 - - def test_skipped_when_disabled(self, tmp_path): - logger = SessionLogger(log_dir=tmp_path, enabled=False) - logger.log_sim_status(target="gspro", state="connected") - assert logger.session_path is None - - -class TestLogSimPlayer: - def test_writes_sim_player_entry(self, tmp_path): - logger = _start(tmp_path) - logger.log_sim_player(target="gspro", handed="right", club="7-iron") - entry = _last_entry(logger) - assert entry["type"] == "sim_player" - assert entry["target"] == "gspro" - assert entry["handed"] == "right" - assert entry["club"] == "7-iron" - - def test_skipped_when_disabled(self, tmp_path): - logger = SessionLogger(log_dir=tmp_path, enabled=False) - logger.log_sim_player(target="gspro", handed="right", club="driver") - assert logger.session_path is None - - -class TestLogIQReading: - def test_writes_iq_reading_entry(self, tmp_path): - logger = _start(tmp_path) - logger.log_iq_reading( - speed_mph=105.0, - direction="outbound", - magnitude=0.85, - snr=12.3, - peak_bin=42, - cfar_validated=True, - block_count=8, - ) - entry = _last_entry(logger) - assert entry["type"] == "iq_reading" - assert entry["speed_mph"] == 105.0 - assert entry["direction"] == "outbound" - assert entry["magnitude"] == 0.85 - assert entry["snr"] == 12.3 - assert entry["peak_bin"] == 42 - assert entry["cfar_validated"] is True - assert entry["block_count"] == 8 - - def test_cfar_not_validated(self, tmp_path): - logger = _start(tmp_path) - logger.log_iq_reading( - speed_mph=30.0, - direction="inbound", - magnitude=0.2, - snr=2.1, - peak_bin=10, - cfar_validated=False, - block_count=4, - ) - entry = _last_entry(logger) - assert entry["cfar_validated"] is False - - def test_skipped_when_disabled(self, tmp_path): - logger = SessionLogger(log_dir=tmp_path, enabled=False) - logger.log_iq_reading( - speed_mph=100.0, - direction="outbound", - magnitude=0.8, - snr=10.0, - peak_bin=40, - cfar_validated=True, - block_count=6, - ) - assert logger.session_path is None - - -class TestLogIQBlocks: - def test_writes_iq_blocks_entry(self, tmp_path): - logger = _start(tmp_path) - blocks = [ - {"i_samples": [1, 2, 3], "q_samples": [4, 5, 6], "timestamp": 1.0}, - {"i_samples": [7, 8, 9], "q_samples": [10, 11, 12], "timestamp": 2.0}, - ] - logger.log_iq_blocks(shot_number=1, blocks=blocks) - entry = _last_entry(logger) - assert entry["type"] == "iq_blocks" - assert entry["shot_number"] == 1 - assert entry["block_count"] == 2 - assert len(entry["blocks"]) == 2 - - def test_empty_blocks(self, tmp_path): - logger = _start(tmp_path) - logger.log_iq_blocks(shot_number=2, blocks=[]) - entry = _last_entry(logger) - assert entry["block_count"] == 0 - assert entry["blocks"] == [] - - def test_skipped_when_disabled(self, tmp_path): - logger = SessionLogger(log_dir=tmp_path, enabled=False) - logger.log_iq_blocks(shot_number=1, blocks=[]) - assert logger.session_path is None - - -class TestLogTriggerEvent: - def test_writes_accepted_trigger_event(self, tmp_path): - logger = _start(tmp_path) - logger.log_trigger_event( - trigger_type="sound-gpio", - accepted=True, - reason=None, - peak_speed_mph=145.0, - readings_count=12, - latency_ms=8.5, - ) - entry = _last_entry(logger) - assert entry["type"] == "trigger_event" - assert entry["trigger_type"] == "sound-gpio" - assert entry["accepted"] is True - assert entry["peak_speed_mph"] == 145.0 - assert entry["readings_count"] == 12 - assert entry["latency_ms"] == 8.5 - - def test_writes_rejected_trigger_event(self, tmp_path): - logger = _start(tmp_path) - logger.log_trigger_event( - trigger_type="sound-gpio", - accepted=False, - reason="no_outbound_speed", - ) - entry = _last_entry(logger) - assert entry["accepted"] is False - assert entry["reason"] == "no_outbound_speed" - - def test_increments_stats(self, tmp_path): - logger = _start(tmp_path) - logger.log_trigger_event(trigger_type="sound-gpio", accepted=True) - logger.log_trigger_event(trigger_type="sound-gpio", accepted=False) - logger.log_trigger_event(trigger_type="sound-gpio", accepted=False) - assert logger.stats["triggers_total"] == 3 - assert logger.stats["triggers_accepted"] == 1 - assert logger.stats["triggers_rejected"] == 2 - - def test_skipped_when_disabled(self, tmp_path): - logger = SessionLogger(log_dir=tmp_path, enabled=False) - logger.log_trigger_event(trigger_type="sound-gpio", accepted=True) - assert logger.session_path is None - - -class TestProperties: - def test_session_id_returns_timestamp_string(self, tmp_path): - logger = _start(tmp_path) - sid = logger.session_id - assert sid is not None - assert len(sid) == 15 # YYYYMMDD_HHMMSS - - def test_session_id_none_before_start(self, tmp_path): - logger = SessionLogger(log_dir=tmp_path, enabled=True) - assert logger.session_id is None - - def test_raw_path_points_to_log_file(self, tmp_path): - logger = _start(tmp_path) - assert logger.raw_path is not None - assert logger.raw_path.suffix == ".log" - assert "radar_raw" in logger.raw_path.name - - def test_raw_path_none_before_start(self, tmp_path): - logger = SessionLogger(log_dir=tmp_path, enabled=True) - assert logger.raw_path is None - - -class TestInitSessionLogger: - def test_returns_session_logger_instance(self, tmp_path): - from openflight.session_logger import init_session_logger - - logger = init_session_logger(log_dir=tmp_path, location="test", enabled=True) - assert isinstance(logger, SessionLogger) - - def test_sets_global_logger(self, tmp_path): - from openflight.session_logger import get_session_logger, init_session_logger - - logger = init_session_logger(log_dir=tmp_path, enabled=True) - assert get_session_logger() is logger - - def test_disabled_logger_creates_instance(self, tmp_path): - from openflight.session_logger import init_session_logger - - logger = init_session_logger(log_dir=tmp_path, enabled=False) - assert isinstance(logger, SessionLogger) - assert not logger.enabled From 0e232a7da75b4cb7da6f169fadcb01b11886844c Mon Sep 17 00:00:00 2001 From: kaydcraig <11017961@uvu.edu> Date: Thu, 30 Jul 2026 12:31:01 -0600 Subject: [PATCH 3/6] Add trigger strategy tests for rolling_buffer module Cover previously untested trigger paths without hardware: - TriggerStrategy base: drain_diagnostics, _append_diagnostic fields - ManualTrigger: request_trigger, reset, wait_for_trigger fire/timeout - PollingTrigger: defaults, activity detection, timeout, None capture, exceptions - ThresholdTrigger: defaults, reset, threshold detection, inbound rejection, timeout - SpeedTriggeredCapture: defaults, reset, last_trigger_speed property - SoundTrigger._clock_sync_last_read_host_time: host_after, host_mid, edge cases - SoundTrigger._clock_sync_age_s: age calculation, missing host time - SoundTrigger._clock_sync_quality: all valid/invalid branches - SoundTrigger._clock_sync_summary_for_log: non-dict, summary fields - SoundTrigger.wait_for_trigger: timeout, parse failure, false trigger rejection --- tests/test_trigger_strategies.py | 573 +++++++++++++++++++++++++++++++ 1 file changed, 573 insertions(+) create mode 100644 tests/test_trigger_strategies.py diff --git a/tests/test_trigger_strategies.py b/tests/test_trigger_strategies.py new file mode 100644 index 000000000..f30587a59 --- /dev/null +++ b/tests/test_trigger_strategies.py @@ -0,0 +1,573 @@ +"""Tests for rolling_buffer trigger strategies.""" + +import time +from unittest.mock import MagicMock + +import pytest + +from openflight.rolling_buffer import ( + IQCapture, + ManualTrigger, + PollingTrigger, + SpeedReading, + SpeedTimeline, + ThresholdTrigger, +) +from openflight.rolling_buffer.trigger import ( + SoundTrigger, + SpeedTriggeredCapture, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_capture(**kwargs): + defaults = dict( + sample_time=100.0, + trigger_time=100.068, + i_samples=[2048] * 4096, + q_samples=[2048] * 4096, + ) + defaults.update(kwargs) + return IQCapture(**defaults) + + +def _make_timeline(speed_mph=130.0, direction="outbound"): + return SpeedTimeline( + readings=[ + SpeedReading( + speed_mph=speed_mph, + magnitude=1000.0, + timestamp_ms=68.0, + direction=direction, + ) + ], + sample_rate_hz=937.5, + ) + + +def _make_radar_mock(response="data", capture=None, timeline=None): + radar = MagicMock() + radar.trigger_capture.return_value = response + return radar + + +def _make_processor_mock(capture=None, timeline=None): + processor = MagicMock() + processor.parse_capture.return_value = capture or _make_capture() + processor.process_standard.return_value = timeline or _make_timeline() + return processor + + +# --------------------------------------------------------------------------- +# TriggerStrategy base (tested via ManualTrigger) +# --------------------------------------------------------------------------- + + +class TestDrainDiagnostics: + def test_returns_empty_list_when_no_diagnostics(self): + trigger = ManualTrigger() + assert trigger.drain_diagnostics() == [] + + def test_returns_and_clears_diagnostics(self): + trigger = ManualTrigger() + trigger._append_diagnostic(accepted=True, reason="accepted") + diags = trigger.drain_diagnostics() + assert len(diags) == 1 + assert diags[0]["accepted"] is True + assert diags[0]["reason"] == "accepted" + # Should be cleared now + assert trigger.drain_diagnostics() == [] + + def test_append_diagnostic_includes_all_fields(self): + trigger = ManualTrigger() + trigger._append_diagnostic( + accepted=False, + reason="no_response", + response_bytes=0, + total_readings=5, + outbound_readings=2, + inbound_readings=3, + peak_outbound_mph=45.0, + peak_inbound_mph=20.0, + all_outbound_speeds=[45.0, 30.0], + all_inbound_speeds=[20.0], + peak_outbound_magnitude=0.8, + peak_inbound_magnitude=0.3, + trigger_latency_ms=12.5, + ) + diag = trigger.drain_diagnostics()[0] + assert diag["response_bytes"] == 0 + assert diag["total_readings"] == 5 + assert diag["peak_outbound_mph"] == 45.0 + assert diag["trigger_latency_ms"] == 12.5 + + def test_append_diagnostic_omits_latency_when_none(self): + trigger = ManualTrigger() + trigger._append_diagnostic(accepted=True, reason="accepted", trigger_latency_ms=None) + diag = trigger.drain_diagnostics()[0] + assert "trigger_latency_ms" not in diag + + def test_all_outbound_speeds_defaults_to_empty_list(self): + trigger = ManualTrigger() + trigger._append_diagnostic(accepted=False, reason="test") + diag = trigger.drain_diagnostics()[0] + assert diag["all_outbound_speeds"] == [] + assert diag["all_inbound_speeds"] == [] + + +# --------------------------------------------------------------------------- +# ManualTrigger +# --------------------------------------------------------------------------- + + +class TestManualTrigger: + def test_request_trigger_sets_flag(self): + trigger = ManualTrigger() + assert not trigger._trigger_requested + trigger.request_trigger() + assert trigger._trigger_requested + + def test_reset_clears_flag(self): + trigger = ManualTrigger() + trigger.request_trigger() + trigger.reset() + assert not trigger._trigger_requested + + def test_wait_for_trigger_fires_immediately_when_pre_requested(self): + trigger = ManualTrigger() + capture = _make_capture() + radar = _make_radar_mock() + processor = _make_processor_mock(capture=capture) + + trigger.request_trigger() + result = trigger.wait_for_trigger(radar, processor, timeout=5.0) + + assert result is capture + radar.trigger_capture.assert_called_once() + radar.rearm_rolling_buffer.assert_called_once_with(12) + + def test_wait_for_trigger_returns_none_on_timeout(self): + trigger = ManualTrigger() + radar = _make_radar_mock() + processor = _make_processor_mock() + + result = trigger.wait_for_trigger(radar, processor, timeout=0.05) + + assert result is None + + def test_wait_for_trigger_clears_flag_after_fire(self): + trigger = ManualTrigger() + radar = _make_radar_mock() + processor = _make_processor_mock() + + trigger.request_trigger() + trigger.wait_for_trigger(radar, processor, timeout=1.0) + + assert not trigger._trigger_requested + + def test_custom_pre_trigger_segments(self): + trigger = ManualTrigger(pre_trigger_segments=20) + capture = _make_capture() + radar = _make_radar_mock() + processor = _make_processor_mock(capture=capture) + + trigger.request_trigger() + trigger.wait_for_trigger(radar, processor, timeout=1.0) + + radar.rearm_rolling_buffer.assert_called_once_with(20) + + +# --------------------------------------------------------------------------- +# PollingTrigger +# --------------------------------------------------------------------------- + + +class TestPollingTrigger: + def test_default_params(self): + trigger = PollingTrigger() + assert trigger.poll_interval == 0.3 + assert trigger.min_readings == 1 + assert trigger.min_speed_mph == 15 + + def test_custom_params(self): + trigger = PollingTrigger(poll_interval=0.1, min_readings=2, min_speed_mph=30) + assert trigger.poll_interval == 0.1 + assert trigger.min_readings == 2 + assert trigger.min_speed_mph == 30 + + def test_reset_is_noop(self): + trigger = PollingTrigger() + trigger.reset() # should not raise + + def test_wait_for_trigger_returns_capture_with_activity(self): + trigger = PollingTrigger(poll_interval=0.0) + capture = _make_capture() + radar = _make_radar_mock() + processor = _make_processor_mock( + capture=capture, + timeline=_make_timeline(speed_mph=130.0, direction="outbound"), + ) + + result = trigger.wait_for_trigger(radar, processor, timeout=5.0) + + assert result is capture + + def test_wait_for_trigger_skips_slow_captures(self): + trigger = PollingTrigger(poll_interval=0.0, min_speed_mph=50) + capture = _make_capture() + radar = _make_radar_mock() + + # First call returns slow speed, second returns fast + slow_timeline = _make_timeline(speed_mph=20.0, direction="outbound") + fast_timeline = _make_timeline(speed_mph=130.0, direction="outbound") + processor = MagicMock() + processor.parse_capture.return_value = capture + processor.process_standard.side_effect = [slow_timeline, fast_timeline] + + result = trigger.wait_for_trigger(radar, processor, timeout=5.0) + + assert result is capture + assert processor.process_standard.call_count == 2 + + def test_wait_for_trigger_returns_none_on_timeout(self): + trigger = PollingTrigger(poll_interval=0.0, min_speed_mph=200) + radar = _make_radar_mock() + processor = _make_processor_mock( + timeline=_make_timeline(speed_mph=10.0, direction="outbound") + ) + + result = trigger.wait_for_trigger(radar, processor, timeout=0.05) + + assert result is None + + def test_wait_for_trigger_handles_none_capture(self): + trigger = PollingTrigger(poll_interval=0.0) + radar = _make_radar_mock() + processor = MagicMock() + processor.parse_capture.return_value = None + + result = trigger.wait_for_trigger(radar, processor, timeout=0.05) + + assert result is None + + def test_wait_for_trigger_handles_radar_exception(self): + trigger = PollingTrigger(poll_interval=0.0) + radar = MagicMock() + radar.trigger_capture.side_effect = [Exception("serial error"), Exception("serial error")] + processor = _make_processor_mock() + + result = trigger.wait_for_trigger(radar, processor, timeout=0.05) + + assert result is None + + +# --------------------------------------------------------------------------- +# ThresholdTrigger +# --------------------------------------------------------------------------- + + +class TestThresholdTrigger: + def test_default_params(self): + trigger = ThresholdTrigger() + assert trigger.speed_threshold_mph == 50 + assert trigger.check_interval == 0.1 + assert trigger.settling_time == 0.05 + + def test_reset_clears_triggered_state(self): + trigger = ThresholdTrigger() + trigger._triggered = True + trigger.reset() + assert not trigger._triggered + + def test_wait_for_trigger_returns_capture_above_threshold(self): + trigger = ThresholdTrigger(speed_threshold_mph=50, settling_time=0.0) + capture = _make_capture() + radar = _make_radar_mock() + processor = _make_processor_mock( + capture=capture, + timeline=_make_timeline(speed_mph=130.0, direction="outbound"), + ) + + result = trigger.wait_for_trigger(radar, processor, timeout=5.0) + + assert result is capture + assert trigger._triggered + + def test_wait_for_trigger_ignores_inbound_readings(self): + trigger = ThresholdTrigger(speed_threshold_mph=50, settling_time=0.0) + capture = _make_capture() + radar = _make_radar_mock() + + # First call inbound, second outbound + inbound = _make_timeline(speed_mph=130.0, direction="inbound") + outbound = _make_timeline(speed_mph=130.0, direction="outbound") + processor = MagicMock() + processor.parse_capture.return_value = capture + processor.process_standard.side_effect = [inbound, outbound] + + result = trigger.wait_for_trigger(radar, processor, timeout=5.0) + + assert result is capture + + def test_wait_for_trigger_returns_none_on_timeout(self): + trigger = ThresholdTrigger(speed_threshold_mph=200, check_interval=0.0) + radar = _make_radar_mock() + processor = _make_processor_mock( + timeline=_make_timeline(speed_mph=50.0, direction="outbound") + ) + + result = trigger.wait_for_trigger(radar, processor, timeout=0.05) + + assert result is None + + def test_wait_for_trigger_handles_exception(self): + trigger = ThresholdTrigger(check_interval=0.0) + radar = MagicMock() + radar.trigger_capture.side_effect = Exception("error") + processor = _make_processor_mock() + + result = trigger.wait_for_trigger(radar, processor, timeout=0.05) + + assert result is None + + +# --------------------------------------------------------------------------- +# SpeedTriggeredCapture +# --------------------------------------------------------------------------- + + +class TestSpeedTriggeredCapture: + def test_default_params(self): + trigger = SpeedTriggeredCapture() + assert trigger.min_trigger_speed_mph == 20.0 + assert trigger.min_ball_speed_mph == 35.0 + assert trigger.trigger_to_capture_delay_ms == 15.0 + assert trigger._needs_reconfigure is True + assert trigger._last_trigger_speed == 0 + + def test_custom_params(self): + trigger = SpeedTriggeredCapture( + min_trigger_speed_mph=25.0, + min_ball_speed_mph=40.0, + trigger_to_capture_delay_ms=20.0, + ) + assert trigger.min_trigger_speed_mph == 25.0 + assert trigger.min_ball_speed_mph == 40.0 + assert trigger.trigger_to_capture_delay_ms == 20.0 + + def test_reset_clears_state(self): + trigger = SpeedTriggeredCapture() + trigger._last_trigger_speed = 95.0 + trigger._needs_reconfigure = False + trigger.reset() + assert trigger._last_trigger_speed == 0 + assert trigger._needs_reconfigure is True + + def test_last_trigger_speed_property(self): + trigger = SpeedTriggeredCapture() + trigger._last_trigger_speed = 88.5 + assert trigger.last_trigger_speed == 88.5 + + +# --------------------------------------------------------------------------- +# SoundTrigger clock sync helpers +# --------------------------------------------------------------------------- + + +class TestClockSyncLastReadHostTime: + def test_returns_host_after_from_last_read(self): + clock_sync = {"reads": [{"host_after": 12345.6, "host_mid": 12345.5}]} + result = SoundTrigger._clock_sync_last_read_host_time(clock_sync) + assert result == 12345.6 + + def test_falls_back_to_host_mid(self): + clock_sync = {"reads": [{"host_mid": 12345.5}]} + result = SoundTrigger._clock_sync_last_read_host_time(clock_sync) + assert result == 12345.5 + + def test_returns_none_for_empty_reads(self): + clock_sync = {"reads": []} + assert SoundTrigger._clock_sync_last_read_host_time(clock_sync) is None + + def test_returns_none_when_no_reads_key(self): + assert SoundTrigger._clock_sync_last_read_host_time({}) is None + + def test_returns_none_when_last_read_not_dict(self): + clock_sync = {"reads": ["not_a_dict"]} + assert SoundTrigger._clock_sync_last_read_host_time(clock_sync) is None + + +class TestClockSyncAgeS: + def test_returns_age_in_seconds(self): + now = time.time() + clock_sync = {"reads": [{"host_after": now - 5.0}]} + age = SoundTrigger._clock_sync_age_s(clock_sync) + assert age == pytest.approx(5.0, abs=0.5) + + def test_returns_none_when_no_host_time(self): + clock_sync = {"reads": []} + assert SoundTrigger._clock_sync_age_s(clock_sync) is None + + +class TestClockSyncQuality: + def _valid_integer_rollover_sync(self): + return { + "usable_for_trigger_timestamps": True, + "best_offset_s": 12000.0, + "clock_sync_method": "integer_rollover", + "rollover_uncertainty_ms": 10.0, + "reads": [], + } + + def test_valid_integer_rollover(self): + sync = self._valid_integer_rollover_sync() + valid, reason = SoundTrigger._clock_sync_quality(sync) + assert valid is True + assert reason == "valid_integer_rollover" + + def test_valid_fractional_clock(self): + sync = { + "usable_for_trigger_timestamps": True, + "best_offset_s": 12000.0, + "clock_sync_method": "fractional_clock", + "reads": [], + } + valid, reason = SoundTrigger._clock_sync_quality(sync) + assert valid is True + assert reason == "valid_fractional_clock" + + def test_invalid_not_dict(self): + valid, reason = SoundTrigger._clock_sync_quality(None) + assert valid is False + assert reason == "missing" + + def test_invalid_not_usable(self): + sync = {"usable_for_trigger_timestamps": False, "clock_sync_method": "integer_rollover"} + valid, reason = SoundTrigger._clock_sync_quality(sync) + assert valid is False + assert "unusable_method" in reason + + def test_invalid_missing_best_offset(self): + sync = { + "usable_for_trigger_timestamps": True, + "best_offset_s": None, + "clock_sync_method": "integer_rollover", + "reads": [], + } + valid, reason = SoundTrigger._clock_sync_quality(sync) + assert valid is False + assert reason == "missing_best_offset" + + def test_invalid_high_rollover_uncertainty(self): + sync = self._valid_integer_rollover_sync() + sync["rollover_uncertainty_ms"] = 100.0 + valid, reason = SoundTrigger._clock_sync_quality(sync) + assert valid is False + assert "rollover_uncertainty" in reason + + def test_invalid_missing_rollover_uncertainty(self): + sync = self._valid_integer_rollover_sync() + del sync["rollover_uncertainty_ms"] + valid, reason = SoundTrigger._clock_sync_quality(sync) + assert valid is False + assert reason == "missing_rollover_uncertainty" + + def test_invalid_timeout_reads(self): + sync = self._valid_integer_rollover_sync() + sync["reads"] = [{"radar_clock_s": None, "read_latency_ms": 60.0}] + valid, reason = SoundTrigger._clock_sync_quality(sync) + assert valid is False + assert "timeout_reads" in reason + + def test_unsupported_method(self): + sync = { + "usable_for_trigger_timestamps": True, + "best_offset_s": 12000.0, + "clock_sync_method": "unknown_method", + "reads": [], + } + valid, reason = SoundTrigger._clock_sync_quality(sync) + assert valid is False + assert "unsupported_method" in reason + + +class TestClockSyncSummaryForLog: + def test_returns_none_for_non_dict(self): + assert SoundTrigger._clock_sync_summary_for_log(None) is None + + def test_returns_summary_dict(self): + sync = { + "usable_for_trigger_timestamps": True, + "best_offset_s": 12000.0, + "clock_sync_method": "fractional_clock", + "source": "per_shot", + "samples": 3, + "valid_samples": 3, + "best_read_latency_ms": 2.1, + "offset_spread_ms": 0.5, + "reads": [{"host_after": time.time()}], + } + summary = SoundTrigger._clock_sync_summary_for_log(sync) + assert summary["valid"] is True + assert summary["reason"] == "valid_fractional_clock" + assert summary["source"] == "per_shot" + assert summary["best_offset_s"] == 12000.0 + assert summary["age_s"] is not None + + +class TestSoundTriggerEdgePaths: + def test_wait_for_trigger_returns_none_on_timeout(self): + trigger = SoundTrigger() + radar = MagicMock() + radar.wait_for_hardware_trigger.return_value = None + processor = _make_processor_mock() + + result = trigger.wait_for_trigger(radar, processor, timeout=1.0) + + assert result is None + + def test_wait_for_trigger_returns_none_when_parse_fails(self): + trigger = SoundTrigger() + radar = MagicMock() + radar.wait_for_hardware_trigger.return_value = "bad_data" + radar.last_hardware_trigger_first_byte_timestamp = None + processor = MagicMock() + processor.parse_capture.return_value = None + + result = trigger.wait_for_trigger(radar, processor, timeout=1.0) + + assert result is None + diags = trigger.drain_diagnostics() + assert len(diags) == 1 + assert diags[0]["accepted"] is False + assert diags[0]["reason"] == "parse_failed" + + def test_wait_for_trigger_rejects_false_trigger(self): + trigger = SoundTrigger() + radar = MagicMock() + radar.wait_for_hardware_trigger.return_value = "data" + radar.last_hardware_trigger_first_byte_timestamp = None + radar.last_clock_sync = None + + capture = _make_capture() + processor = MagicMock() + processor.parse_capture.return_value = capture + # No valid outbound readings + processor.process_standard.return_value = _make_timeline( + speed_mph=5.0, direction="outbound" + ) + + result = trigger.wait_for_trigger(radar, processor, timeout=1.0) + + assert result is None + diags = trigger.drain_diagnostics() + assert len(diags) == 1 + assert diags[0]["accepted"] is False + assert diags[0]["reason"] == "no_outbound_speed" + + def test_reset_is_noop(self): + trigger = SoundTrigger() + trigger.reset() # should not raise From 8aab1caf3f070a92d390a4c84d3916d9083ced30 Mon Sep 17 00:00:00 2001 From: kaydcraig <11017961@uvu.edu> Date: Thu, 30 Jul 2026 12:49:51 -0600 Subject: [PATCH 4/6] Add altitude-based air density to ballistics model MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add air_density_at_altitude() using the ISA tropospheric formula, valid from sea level to ~11 km (covers all realistic golf venues). At Denver (1609m) density is ~85% of sea level; at high-altitude courses like Leadville (3094m) it drops to ~74%, adding meaningful carry distance that the previous sea-level constant missed. Add optional altitude_m parameter to simulate() that overrides air_density when provided. Existing call sites are unaffected — the parameter defaults to None and behaviour is unchanged. Add 11 tests covering: - ISA density at sea level, Denver, Leadville, negative altitude - altitude_m increases carry vs sea level - altitude_m overrides explicit air_density argument - altitude_m=None leaves air_density param untouched --- src/openflight/ballistics.py | 82 +++++++++++++++++++-------- tests/test_ballistics.py | 105 +++++++++++++++++++++++++++++++++++ 2 files changed, 165 insertions(+), 22 deletions(-) diff --git a/src/openflight/ballistics.py b/src/openflight/ballistics.py index 85a53ed2c..14c23707a 100644 --- a/src/openflight/ballistics.py +++ b/src/openflight/ballistics.py @@ -33,7 +33,7 @@ # the rules rather than by a guess at the specific ball in play. BALL_MASS_KG = 0.04593 BALL_RADIUS_M = 0.02135 -BALL_AREA_M2 = math.pi * BALL_RADIUS_M ** 2 +BALL_AREA_M2 = math.pi * BALL_RADIUS_M**2 AIR_DENSITY_STD = 1.225 # kg/m³ at sea level, 15 °C ISA # Cd = CD_BASE + CD_SPIN_COEFF * Sp @@ -65,6 +65,40 @@ # keeping payload size reasonable for UI/log consumers. SAMPLE_INTERVAL_S = 0.05 +# ISA (International Standard Atmosphere) tropospheric constants for air +# density at altitude. Valid from sea level to ~11 km (36,089 ft), which +# covers all realistic golf venues (highest course ≈ 4,300 m / 14,000 ft). +_ISA_SEA_LEVEL_TEMP_K = 288.15 # 15 °C +_ISA_LAPSE_RATE = 0.0065 # K/m +_ISA_SEA_LEVEL_PRESSURE_PA = 101325.0 +_ISA_PRESSURE_EXPONENT = 5.25588 # g / (R·L) = 9.80665 / (287.058 · 0.0065) +_ISA_GAS_CONSTANT_DRY = 287.058 # J / (kg·K) + + +def air_density_at_altitude(altitude_m: float) -> float: + """Return ISA dry-air density (kg/m³) at the given altitude in metres. + + Uses the standard tropospheric lapse rate (valid to ~11 km / 36,000 ft), + which covers every realistic golf venue on Earth. The highest active + course is around 4,300 m (14,000 ft); at that altitude air density is + ~63% of sea level, adding roughly 10-12% to carry compared with a + sea-level calculation. + + Args: + altitude_m: Altitude above sea level in metres. Negative values + (below sea level, e.g. Dead Sea) are clamped to 0. + + Returns: + Air density in kg/m³. + """ + h = max(0.0, float(altitude_m)) + temp_k = _ISA_SEA_LEVEL_TEMP_K - _ISA_LAPSE_RATE * h + pressure_pa = ( + _ISA_SEA_LEVEL_PRESSURE_PA * (temp_k / _ISA_SEA_LEVEL_TEMP_K) ** _ISA_PRESSURE_EXPONENT + ) + return pressure_pa / (_ISA_GAS_CONSTANT_DRY * temp_k) + + # Club-typical spin (RPM) from TrackMan PGA Tour averages. # Used as fallback when measured spin is missing or low-confidence. CLUB_TYPICAL_SPIN_RPM: dict[ClubType, float] = { @@ -156,9 +190,7 @@ def resolve_launch(shot: Shot) -> Optional[LaunchConditions]: spin_rpm = float(shot.spin_rpm) source: Literal["measured", "club_typical"] = "measured" else: - spin_rpm = CLUB_TYPICAL_SPIN_RPM.get( - shot.club, CLUB_TYPICAL_SPIN_RPM[ClubType.UNKNOWN] - ) + spin_rpm = CLUB_TYPICAL_SPIN_RPM.get(shot.club, CLUB_TYPICAL_SPIN_RPM[ClubType.UNKNOWN]) source = "club_typical" return LaunchConditions( @@ -244,20 +276,20 @@ def _rk4_step( k3 = _derivatives(s3, omega, axis, air_density) s4 = tuple(state[i] + dt * k3[i] for i in range(6)) k4 = _derivatives(s4, omega, axis, air_density) - return tuple( - state[i] + (dt / 6.0) * (k1[i] + 2 * k2[i] + 2 * k3[i] + k4[i]) - for i in range(6) - ) + return tuple(state[i] + (dt / 6.0) * (k1[i] + 2 * k2[i] + 2 * k3[i] + k4[i]) for i in range(6)) def simulate( conditions: LaunchConditions, air_density: float = AIR_DENSITY_STD, dt: float = DT_SECONDS, + altitude_m: Optional[float] = None, ) -> Trajectory: """ Integrate flight from launch to first ground contact (z = 0). """ + if altitude_m is not None: + air_density = air_density_at_altitude(altitude_m) v0 = conditions.ball_speed_mph * MPH_TO_MPS la_v = math.radians(conditions.launch_angle_v) la_h = math.radians(conditions.launch_angle_h) @@ -311,15 +343,17 @@ def simulate( final = tuple(state[i] + frac * (new_state[i] - state[i]) for i in range(6)) fx, fy, fz, fvx, fvy, fvz = final v_final = math.sqrt(fvx * fvx + fvy * fvy + fvz * fvz) - landing_angle = math.degrees( - math.atan2(-fvz, math.sqrt(fvx * fvx + fvy * fvy)) + landing_angle = math.degrees(math.atan2(-fvz, math.sqrt(fvx * fvx + fvy * fvy))) + points.append( + TrajectoryPoint( + t_hit, + fx * M_TO_YD, + fy * M_TO_YD, + max(fz, 0.0) * M_TO_YD, + v_final * MPS_TO_MPH, + omega * 60 / (2 * math.pi), + ) ) - points.append(TrajectoryPoint( - t_hit, - fx * M_TO_YD, fy * M_TO_YD, max(fz, 0.0) * M_TO_YD, - v_final * MPS_TO_MPH, - omega * 60 / (2 * math.pi), - )) return Trajectory( points=points, carry_yards=fx * M_TO_YD, @@ -335,12 +369,16 @@ def simulate( if t - last_sample_t >= SAMPLE_INTERVAL_S: sx_, sy_, sz_, svx, svy, svz = state v = math.sqrt(svx * svx + svy * svy + svz * svz) - points.append(TrajectoryPoint( - t, - sx_ * M_TO_YD, sy_ * M_TO_YD, sz_ * M_TO_YD, - v * MPS_TO_MPH, - omega * 60 / (2 * math.pi), - )) + points.append( + TrajectoryPoint( + t, + sx_ * M_TO_YD, + sy_ * M_TO_YD, + sz_ * M_TO_YD, + v * MPS_TO_MPH, + omega * 60 / (2 * math.pi), + ) + ) last_sample_t = t # Flight did not terminate — return current state as best-effort diff --git a/tests/test_ballistics.py b/tests/test_ballistics.py index 4b4ff085c..71076280c 100644 --- a/tests/test_ballistics.py +++ b/tests/test_ballistics.py @@ -168,3 +168,108 @@ def test_zero_launch_angle_does_not_crash(self): def test_total_distance_includes_rollout(self): traj = simulate(_driver()) assert traj.total_yards > traj.carry_yards + + +class TestAirDensityAtAltitude: + """Tests for the ISA air density function.""" + + def test_sea_level_matches_standard(self): + """At 0 m altitude should return standard sea-level density.""" + from openflight.ballistics import AIR_DENSITY_STD, air_density_at_altitude + + assert air_density_at_altitude(0) == pytest.approx(AIR_DENSITY_STD, rel=1e-3) + + def test_density_decreases_with_altitude(self): + """Higher altitude should produce lower air density.""" + from openflight.ballistics import air_density_at_altitude + + assert air_density_at_altitude(1000) < air_density_at_altitude(0) + assert air_density_at_altitude(2000) < air_density_at_altitude(1000) + + def test_denver_altitude(self): + """Denver (~1609 m / 5280 ft) should be ~83% of sea-level density.""" + from openflight.ballistics import AIR_DENSITY_STD, air_density_at_altitude + + denver = air_density_at_altitude(1609) + ratio = denver / AIR_DENSITY_STD + assert 0.84 <= ratio <= 0.87 + + def test_high_altitude_course(self): + """Leadville, CO (~3094 m / 10,152 ft) should be ~70% of sea level.""" + from openflight.ballistics import AIR_DENSITY_STD, air_density_at_altitude + + leadville = air_density_at_altitude(3094) + ratio = leadville / AIR_DENSITY_STD + assert 0.72 <= ratio <= 0.75 + + def test_negative_altitude_clamped_to_sea_level(self): + """Below sea level (e.g. Dead Sea) should return sea-level density.""" + from openflight.ballistics import air_density_at_altitude + + assert air_density_at_altitude(-100) == pytest.approx(air_density_at_altitude(0), rel=1e-6) + + def test_returns_float(self): + from openflight.ballistics import air_density_at_altitude + + assert isinstance(air_density_at_altitude(1000), float) + + +class TestSimulateAltitude: + """Tests for altitude_m parameter on simulate().""" + + def _conditions(self): + from openflight.ballistics import LaunchConditions + + return LaunchConditions( + ball_speed_mph=150.0, + launch_angle_v=12.0, + launch_angle_h=0.0, + spin_rpm=2700, + spin_axis_deg=0.0, + spin_source="club_typical", + ) + + def test_altitude_increases_carry(self): + """Higher altitude should produce longer carry due to thinner air.""" + from openflight.ballistics import simulate + + sea_level = simulate(self._conditions(), altitude_m=0) + denver = simulate(self._conditions(), altitude_m=1609) + assert denver.carry_yards > sea_level.carry_yards + + def test_high_altitude_carry_significantly_longer(self): + """At 2000m carry should be meaningfully longer than sea level.""" + from openflight.ballistics import simulate + + sea_level = simulate(self._conditions(), altitude_m=0) + high = simulate(self._conditions(), altitude_m=2000) + # Expect at least 5% more carry at 2000m + assert high.carry_yards > sea_level.carry_yards * 1.03 + + def test_altitude_overrides_air_density(self): + """altitude_m should override explicit air_density argument.""" + from openflight.ballistics import AIR_DENSITY_STD, air_density_at_altitude, simulate + + result_via_altitude = simulate( + self._conditions(), air_density=AIR_DENSITY_STD, altitude_m=1609 + ) + result_via_density = simulate(self._conditions(), air_density=air_density_at_altitude(1609)) + assert result_via_altitude.carry_yards == pytest.approx( + result_via_density.carry_yards, rel=1e-6 + ) + + def test_zero_altitude_matches_default(self): + """altitude_m=0 should match the default sea-level simulation.""" + from openflight.ballistics import simulate + + default = simulate(self._conditions()) + explicit = simulate(self._conditions(), altitude_m=0) + assert explicit.carry_yards == pytest.approx(default.carry_yards, rel=1e-3) + + def test_none_altitude_uses_air_density_param(self): + """altitude_m=None should leave air_density param untouched.""" + from openflight.ballistics import AIR_DENSITY_STD, simulate + + default = simulate(self._conditions()) + explicit = simulate(self._conditions(), altitude_m=None, air_density=AIR_DENSITY_STD) + assert explicit.carry_yards == pytest.approx(default.carry_yards, rel=1e-6) From bcc96acbe88d68ed19cf251fac2463d57edf0287 Mon Sep 17 00:00:00 2001 From: kaydcraig <11017961@uvu.edu> Date: Thu, 30 Jul 2026 13:13:20 -0600 Subject: [PATCH 5/6] Add altitude setting to UI and wire to ballistics model Users can now set altitude in the Debug Panel tuning section with a m/ft toggle. The value is always stored internally in metres and sent to the server, but displays and accepts input in whichever unit the user selects. Conversion is handled entirely in the UI. - radar_config: add altitude_m field (default 0 = sea level) - set_radar_config: handle altitude_m updates from UI - simulate(): pass altitude_m from radar_config - RadarConfig type: add altitude_m field - useDebugStore: initialise altitude_m to 0 - DebugPanel: add AltitudeControl with m/ft toggle (0-4500m / 0-14764ft) --- src/openflight/server.py | 9 +++- ui/src/components/DebugPanel.tsx | 81 +++++++++++++++++++++++++++++++- ui/src/stores/useDebugStore.ts | 1 + ui/src/types/socket.ts | 1 + 4 files changed, 90 insertions(+), 2 deletions(-) diff --git a/src/openflight/server.py b/src/openflight/server.py index 83699e00d..eb7053cde 100644 --- a/src/openflight/server.py +++ b/src/openflight/server.py @@ -1591,6 +1591,7 @@ def handle_get_debug_status(): "max_speed": 220, "min_magnitude": 0, "transmit_power": 0, + "altitude_m": 0, } @@ -1644,6 +1645,12 @@ def handle_set_radar_config(data): radar_config["transmit_power"] = new_power print(f"Set transmit power: {new_power}") + # Update altitude for ballistics + if "altitude_m" in data: + new_altitude = max(0, int(data["altitude_m"])) + radar_config["altitude_m"] = new_altitude + print(f"Set altitude: {new_altitude} m") + # Log config change session_logger = get_session_logger() if session_logger: @@ -2365,7 +2372,7 @@ def on_shot_detected(shot: Shot): if shot.carry_spin_adjusted is None and shot.mode != "mock": conditions = resolve_launch(shot) if ballistics_enabled else None if conditions is not None: - trajectory = simulate(conditions) + trajectory = simulate(conditions, altitude_m=radar_config.get("altitude_m", 0) or None) shot.carry_spin_adjusted = trajectory.carry_yards logger.info( "[SERVER] Ballistic carry: %.0f yds (spin: %.0f rpm, source: %s)", diff --git a/ui/src/components/DebugPanel.tsx b/ui/src/components/DebugPanel.tsx index 43783837b..24d45038c 100644 --- a/ui/src/components/DebugPanel.tsx +++ b/ui/src/components/DebugPanel.tsx @@ -50,6 +50,81 @@ function formatTime(timestamp: string): string { }); } + +const FT_PER_M = 3.28084; + +function AltitudeControl({ + altitudeM, + onChange, +}: { + altitudeM: number; + onChange: (metres: number) => void; +}) { + const [useFeet, setUseFeet] = useState(false); + const [dragging, setDragging] = useState(false); + const [localM, setLocalM] = useState(altitudeM); + const [prevM, setPrevM] = useState(altitudeM); + + if (prevM !== altitudeM) { + setPrevM(altitudeM); + if (!dragging) setLocalM(altitudeM); + } + + const displayValue = useFeet ? Math.round(localM * FT_PER_M) : localM; + const maxDisplay = useFeet ? 14764 : 4500; // 4500m ≈ 14,764 ft + const stepDisplay = useFeet ? 100 : 50; + + const handleChange = (e: React.ChangeEvent) => { + setDragging(true); + const raw = parseInt(e.target.value, 10); + setLocalM(useFeet ? Math.round(raw / FT_PER_M) : raw); + }; + + const handleRelease = () => { + setDragging(false); + if (localM !== altitudeM) onChange(localM); + }; + + const toggleUnit = () => setUseFeet((f) => !f); + + return ( +
+
+ Altitude + + {displayValue} + + +
+ +
+ ); +} + interface SliderControlProps { label: string; value: number; @@ -387,8 +462,12 @@ export function DebugPanel({ disabled={mockMode} onChange={(v) => onUpdateConfig({ transmit_power: v })} /> + onUpdateConfig({ altitude_m: m })} + /> -

TX Power: 0 = max range, 7 = min range

+

TX Power: 0 = max range, 7 = min range. Altitude: metres above sea level (affects carry distance).

)} diff --git a/ui/src/stores/useDebugStore.ts b/ui/src/stores/useDebugStore.ts index fa87fa274..9576c863c 100644 --- a/ui/src/stores/useDebugStore.ts +++ b/ui/src/stores/useDebugStore.ts @@ -26,6 +26,7 @@ export const useDebugStore = create((set) => ({ max_speed: 220, min_magnitude: 0, transmit_power: 0, + altitude_m: 0, }, triggerDiagnostics: [], triggerStatus: { diff --git a/ui/src/types/socket.ts b/ui/src/types/socket.ts index 1be63501c..f6cf56c6e 100644 --- a/ui/src/types/socket.ts +++ b/ui/src/types/socket.ts @@ -30,6 +30,7 @@ export interface RadarConfig { max_speed: number; min_magnitude: number; transmit_power: number; + altitude_m: number; } export interface DebugShotLog { From cee11927b825dd0bcf51afa99e51eee12e0c3a2c Mon Sep 17 00:00:00 2001 From: kaydcraig <11017961@uvu.edu> Date: Sat, 1 Aug 2026 10:34:03 -0600 Subject: [PATCH 6/6] ci: fix eslint pre-commit hook working directory --- .pre-commit-config.yaml | 2 +- ui/src/components/DebugPanel.tsx | 18 +++++------------- 2 files changed, 6 insertions(+), 14 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index f5c029133..7488b4fdb 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -17,7 +17,7 @@ repos: - id: eslint name: eslint - entry: npm run lint + entry: bash -c 'cd ui && npm run lint' language: system files: ^ui/src/.*\.(ts|tsx)$ pass_filenames: false diff --git a/ui/src/components/DebugPanel.tsx b/ui/src/components/DebugPanel.tsx index 24d45038c..59495339a 100644 --- a/ui/src/components/DebugPanel.tsx +++ b/ui/src/components/DebugPanel.tsx @@ -50,16 +50,9 @@ function formatTime(timestamp: string): string { }); } - const FT_PER_M = 3.28084; -function AltitudeControl({ - altitudeM, - onChange, -}: { - altitudeM: number; - onChange: (metres: number) => void; -}) { +function AltitudeControl({ altitudeM, onChange }: { altitudeM: number; onChange: (metres: number) => void }) { const [useFeet, setUseFeet] = useState(false); const [dragging, setDragging] = useState(false); const [localM, setLocalM] = useState(altitudeM); @@ -462,12 +455,11 @@ export function DebugPanel({ disabled={mockMode} onChange={(v) => onUpdateConfig({ transmit_power: v })} /> - onUpdateConfig({ altitude_m: m })} - /> + onUpdateConfig({ altitude_m: m })} /> -

TX Power: 0 = max range, 7 = min range. Altitude: metres above sea level (affects carry distance).

+

+ TX Power: 0 = max range, 7 = min range. Altitude: metres above sea level (affects carry distance). +

)}