diff --git a/src/openflight/cloud/commands.py b/src/openflight/cloud/commands.py index aacd3e6da..972804dfc 100644 --- a/src/openflight/cloud/commands.py +++ b/src/openflight/cloud/commands.py @@ -128,6 +128,7 @@ def cmd_push( "offline": False, "needs_relink": False, "dry_run": dry_run, + "skipped_empty": 0, } if not dry_run and not config.is_active(): @@ -158,11 +159,23 @@ def cmd_push( # keep the (tiny) shot lines. Loading it whole would risk OOM on a Pi. result = filtering.filter_session_file(path, config.device_id) session_id = result.session_id + shot_count = result.kept_type_counts.get("shot_detected", 0) + session_ended = result.kept_type_counts.get("session_end", 0) > 0 if dry_run: _describe_dry_run(path.name, session_id, result, out) continue + if shot_count == 0: + if session_ended: + spool.mark_skipped(path, reason="no_shots", shot_count=0) + summary["skipped_empty"] += 1 + out(f"{path.name}: skipped (0 shots).") + else: + summary["deferred"] += 1 + out(f"{path.name}: waiting for shots before upload.") + continue + try: body = filtering.build_upload_body(result) except filtering.BodyTooLargeError as exc: diff --git a/src/openflight/cloud/spool.py b/src/openflight/cloud/spool.py index e6a82f62a..0f8c7bc9c 100644 --- a/src/openflight/cloud/spool.py +++ b/src/openflight/cloud/spool.py @@ -19,6 +19,7 @@ PUSHED_SUFFIX = ".pushed" PARKED_SUFFIX = ".parked" +SKIPPED_SUFFIX = ".skipped" STATE_SUFFIX = ".state" SESSION_GLOB = "session_*.jsonl" @@ -59,9 +60,18 @@ def is_parked(path: Path) -> bool: return _sidecar(path, PARKED_SUFFIX).exists() +def is_skipped(path: Path) -> bool: + """True if a ``.skipped`` marker exists for this session.""" + return _sidecar(path, SKIPPED_SUFFIX).exists() + + def pending_sessions(log_dir: Path) -> List[Path]: - """Session files that are neither pushed nor parked.""" - return [p for p in session_files(log_dir) if not is_pushed(p) and not is_parked(p)] + """Session files that are not in a terminal upload state.""" + return [ + p + for p in session_files(log_dir) + if not is_pushed(p) and not is_parked(p) and not is_skipped(p) + ] def read_attempts(path: Path) -> int: @@ -150,6 +160,17 @@ def mark_parked(path: Path, reason: str, attempts: int, last_error: Optional[str ) +def mark_skipped(path: Path, reason: str, shot_count: Optional[int] = None) -> None: + """Mark a session as intentionally skipped by upload policy.""" + _write_json( + _sidecar(path, SKIPPED_SUFFIX), + {"reason": reason, "shot_count": shot_count, "skipped_at": _now()}, + ) + state_path = _sidecar(path, STATE_SUFFIX) + if state_path.exists(): + state_path.unlink() + + def clear_markers(path: Path, include_pushed: bool = False) -> List[str]: """Remove terminal/retry markers so a session becomes pending again. @@ -158,7 +179,7 @@ def clear_markers(path: Path, include_pushed: bool = False) -> List[str]: the server already stored (idempotent server-side). Returns the suffixes actually removed. """ - suffixes = [PARKED_SUFFIX, STATE_SUFFIX] + suffixes = [PARKED_SUFFIX, SKIPPED_SUFFIX, STATE_SUFFIX] if include_pushed: suffixes.append(PUSHED_SUFFIX) cleared: List[str] = [] @@ -175,10 +196,14 @@ def summarize(log_dir: Path) -> Dict[str, int]: files = session_files(log_dir) pushed = sum(1 for p in files if is_pushed(p)) parked = sum(1 for p in files if is_parked(p)) - pending = sum(1 for p in files if not is_pushed(p) and not is_parked(p)) + skipped = sum(1 for p in files if is_skipped(p)) + pending = sum( + 1 for p in files if not is_pushed(p) and not is_parked(p) and not is_skipped(p) + ) return { "total": len(files), "pushed": pushed, "parked": parked, + "skipped": skipped, "pending": pending, } diff --git a/tests/test_cloud_commands.py b/tests/test_cloud_commands.py index e4beb4356..7b7938a76 100644 --- a/tests/test_cloud_commands.py +++ b/tests/test_cloud_commands.py @@ -98,8 +98,18 @@ def test_dry_run_does_not_upload_or_mark(self, tmp_path): assert "rolling_buffer_capture" not in printed def test_relink_stops_and_flags(self, tmp_path): - _write_session(tmp_path, "session_a.jsonl", {"type": "session_start", "session_uuid": "a"}) - _write_session(tmp_path, "session_b.jsonl", {"type": "session_start", "session_uuid": "b"}) + _write_session( + tmp_path, + "session_a.jsonl", + {"type": "session_start", "session_uuid": "a"}, + {"type": "shot_detected", "ball_speed_mph": 90}, + ) + _write_session( + tmp_path, + "session_b.jsonl", + {"type": "session_start", "session_uuid": "b"}, + {"type": "shot_detected", "ball_speed_mph": 91}, + ) client = FakeClient( uploads=[UploadResult(401, action="relink", reason="invalid_or_revoked_token")] ) @@ -110,7 +120,10 @@ def test_relink_stops_and_flags(self, tmp_path): def test_park_on_422(self, tmp_path): path = _write_session( - tmp_path, "session_a.jsonl", {"type": "session_start", "session_uuid": "a"} + tmp_path, + "session_a.jsonl", + {"type": "session_start", "session_uuid": "a"}, + {"type": "shot_detected", "ball_speed_mph": 90}, ) client = FakeClient(uploads=[UploadResult(422, action="park", reason="invalid_gzip")]) commands.cmd_push(_linked_config(), tmp_path, client, out=lambda _m: None) @@ -118,7 +131,10 @@ def test_park_on_422(self, tmp_path): def test_quota_sets_cooldown_not_park(self, tmp_path): path = _write_session( - tmp_path, "session_a.jsonl", {"type": "session_start", "session_uuid": "a"} + tmp_path, + "session_a.jsonl", + {"type": "session_start", "session_uuid": "a"}, + {"type": "shot_detected", "ball_speed_mph": 90}, ) client = FakeClient(uploads=[UploadResult(402, action="quota", reason="quota_exceeded")]) commands.cmd_push(_linked_config(), tmp_path, client, out=lambda _m: None) @@ -127,7 +143,10 @@ def test_quota_sets_cooldown_not_park(self, tmp_path): def test_5xx_records_failure_and_leaves_pending(self, tmp_path): path = _write_session( - tmp_path, "session_a.jsonl", {"type": "session_start", "session_uuid": "a"} + tmp_path, + "session_a.jsonl", + {"type": "session_start", "session_uuid": "a"}, + {"type": "shot_detected", "ball_speed_mph": 90}, ) client = FakeClient(uploads=[UploadResult(503, action="retry")]) commands.cmd_push(_linked_config(), tmp_path, client, out=lambda _m: None) @@ -149,7 +168,10 @@ def test_oversize_body_parks(self, tmp_path, monkeypatch): def test_skips_sessions_in_cooldown(self, tmp_path): path = _write_session( - tmp_path, "session_a.jsonl", {"type": "session_start", "session_uuid": "a"} + tmp_path, + "session_a.jsonl", + {"type": "session_start", "session_uuid": "a"}, + {"type": "shot_detected", "ball_speed_mph": 90}, ) spool.record_cooldown(path, "quota_exceeded", seconds=spool.QUOTA_COOLDOWN_S) client = FakeClient(uploads=[UploadResult(201, action="success")]) @@ -157,11 +179,46 @@ def test_skips_sessions_in_cooldown(self, tmp_path): assert client.uploaded == [] assert result["deferred"] == 1 + def test_ended_zero_shot_session_is_skipped_not_uploaded(self, tmp_path): + path = _write_session( + tmp_path, + "session_empty.jsonl", + {"type": "session_start", "session_uuid": "empty"}, + {"type": "session_end"}, + ) + client = FakeClient(uploads=[UploadResult(201, action="success", shot_count=0)]) + out = [] + result = commands.cmd_push(_linked_config(), tmp_path, client, out=out.append) + + assert client.uploaded == [] + assert spool.is_skipped(path) + assert result["skipped_empty"] == 1 + assert "skipped (0 shots)" in "\n".join(out) + + def test_active_zero_shot_session_waits_for_later_shots(self, tmp_path): + path = _write_session( + tmp_path, + "session_active.jsonl", + {"type": "session_start", "session_uuid": "active"}, + ) + client = FakeClient(uploads=[UploadResult(201, action="success", shot_count=0)]) + out = [] + result = commands.cmd_push(_linked_config(), tmp_path, client, out=out.append) + + assert client.uploaded == [] + assert not spool.is_skipped(path) + assert path in spool.pending_sessions(tmp_path) + assert result["deferred"] == 1 + assert "waiting for shots" in "\n".join(out) + class TestPushRetry: def test_retry_all_reuploads_parked_session(self, tmp_path): path = _write_session( - tmp_path, "session_a.jsonl", {"type": "session_start", "session_uuid": "a"} + tmp_path, + "session_a.jsonl", + {"type": "session_start", "session_uuid": "a"}, + {"type": "shot_detected", "ball_speed_mph": 90}, ) spool.mark_parked(path, reason="max_attempts", attempts=20, last_error="503") client = FakeClient(uploads=[UploadResult(201, action="success", shot_count=1)]) @@ -174,7 +231,10 @@ def test_retry_all_reuploads_parked_session(self, tmp_path): def test_retry_all_reuploads_cooled_down_session(self, tmp_path): path = _write_session( - tmp_path, "session_a.jsonl", {"type": "session_start", "session_uuid": "a"} + tmp_path, + "session_a.jsonl", + {"type": "session_start", "session_uuid": "a"}, + {"type": "shot_detected", "ball_speed_mph": 90}, ) spool.record_cooldown(path, "quota_exceeded", seconds=spool.QUOTA_COOLDOWN_S) client = FakeClient(uploads=[UploadResult(201, action="success", shot_count=1)]) @@ -193,7 +253,10 @@ def test_retry_all_leaves_pushed_sessions_alone(self, tmp_path): def test_retry_named_session_force_reuploads_pushed(self, tmp_path): path = _write_session( - tmp_path, "session_20260527_x.jsonl", {"type": "session_start", "session_uuid": "a"} + tmp_path, + "session_20260527_x.jsonl", + {"type": "session_start", "session_uuid": "a"}, + {"type": "shot_detected", "ball_speed_mph": 90}, ) spool.mark_pushed(path, "a", 0) # previously "uploaded" with 0 shots client = FakeClient(uploads=[UploadResult(201, action="success", shot_count=5)]) diff --git a/tests/test_cloud_spool.py b/tests/test_cloud_spool.py index ba3497498..391c01947 100644 --- a/tests/test_cloud_spool.py +++ b/tests/test_cloud_spool.py @@ -29,8 +29,10 @@ def test_pending_excludes_pushed_and_parked(self, tmp_path): a = _session(tmp_path, "session_a.jsonl") b = _session(tmp_path, "session_b.jsonl") c = _session(tmp_path, "session_c.jsonl") + d = _session(tmp_path, "session_d.jsonl") spool.mark_pushed(b, session_id="id-b", shot_count=3) spool.mark_parked(c, reason="quota_exceeded", attempts=20, last_error="402") + spool.mark_skipped(d, reason="no_shots", shot_count=0) pending = {p.name for p in spool.pending_sessions(tmp_path)} assert pending == {"session_a.jsonl"} assert a # referenced @@ -68,6 +70,25 @@ def test_mark_parked_creates_sidecar(self, tmp_path): assert spool.is_parked(path) +class TestSkippedMarker: + def test_mark_skipped_creates_sidecar(self, tmp_path): + path = _session(tmp_path) + spool.mark_skipped(path, reason="no_shots", shot_count=0) + marker = tmp_path / (path.name + ".skipped") + assert marker.exists() + data = json.loads(marker.read_text()) + assert data["reason"] == "no_shots" + assert data["shot_count"] == 0 + assert spool.is_skipped(path) + + def test_mark_skipped_clears_attempt_state(self, tmp_path): + path = _session(tmp_path) + spool.record_failure(path, "5xx") + spool.mark_skipped(path, reason="no_shots", shot_count=0) + assert spool.read_attempts(path) == 0 + assert not (tmp_path / (path.name + ".state")).exists() + + class TestAttemptCounter: def test_attempts_start_at_zero(self, tmp_path): path = _session(tmp_path) @@ -117,13 +138,16 @@ def test_clears_parked_and_state_keeps_pushed_by_default(self, tmp_path): path = _session(tmp_path) spool.mark_pushed(path, "id", 1) spool.mark_parked(path, reason="r", attempts=2, last_error="e") + spool.mark_skipped(path, reason="no_shots", shot_count=0) spool.record_failure(path, "e") # writes .state cleared = spool.clear_markers(path) assert spool.PARKED_SUFFIX in cleared + assert spool.SKIPPED_SUFFIX in cleared assert spool.STATE_SUFFIX in cleared assert spool.PUSHED_SUFFIX not in cleared assert spool.is_pushed(path) assert not spool.is_parked(path) + assert not spool.is_skipped(path) def test_include_pushed_clears_everything(self, tmp_path): path = _session(tmp_path) @@ -151,9 +175,11 @@ def test_summarizes_counts(self, tmp_path): c = _session(tmp_path, "session_c.jsonl") spool.mark_pushed(a, session_id="a", shot_count=1) spool.mark_parked(b, reason="r", attempts=20, last_error="402") + spool.mark_skipped(c, reason="no_shots", shot_count=0) assert c summary = spool.summarize(tmp_path) assert summary["pushed"] == 1 assert summary["parked"] == 1 - assert summary["pending"] == 1 + assert summary["skipped"] == 1 + assert summary["pending"] == 0 assert summary["total"] == 3