Ryan Malloy
c7320e599b
4 new tools (tidal_phase, deployment_briefing, catch_tidal_context, water_level_anomaly) and 2 prompts (smartpot_deployment, crab_pot_analysis) for autonomous crab pot deployment planning and catch correlation. Pure tidal phase classification in tidal.py with no MCP dependencies. 65 tests passing, lint clean.
205 lines
6.7 KiB
Python
205 lines
6.7 KiB
Python
"""Unit tests for pure tidal phase classification — no MCP, no I/O."""
|
|
|
|
from datetime import datetime
|
|
|
|
from mcnoaa_tides.tidal import (
|
|
classify_tidal_phase,
|
|
interpolate_predictions,
|
|
parse_hilo_predictions,
|
|
)
|
|
|
|
# Synthetic hilo data: a full day with 2 highs and 2 lows
|
|
RAW_HILO = [
|
|
{"t": "2026-02-21 04:30", "v": "4.521", "type": "H"},
|
|
{"t": "2026-02-21 10:42", "v": "-0.123", "type": "L"},
|
|
{"t": "2026-02-21 16:55", "v": "5.012", "type": "H"},
|
|
{"t": "2026-02-21 23:08", "v": "0.234", "type": "L"},
|
|
]
|
|
|
|
|
|
# --- parse_hilo_predictions ---
|
|
|
|
|
|
def test_parse_hilo_basic():
|
|
events = parse_hilo_predictions(RAW_HILO)
|
|
assert len(events) == 4
|
|
assert events[0]["type"] == "H"
|
|
assert events[0]["v"] == 4.521
|
|
assert isinstance(events[0]["dt"], datetime)
|
|
|
|
|
|
def test_parse_hilo_filters_non_hilo():
|
|
"""Records without a 'type' field (6-min interval data) should be dropped."""
|
|
mixed = [
|
|
{"t": "2026-02-21 04:00", "v": "3.200"}, # no type — filtered out
|
|
{"t": "2026-02-21 04:30", "v": "4.521", "type": "H"},
|
|
]
|
|
events = parse_hilo_predictions(mixed)
|
|
assert len(events) == 1
|
|
assert events[0]["type"] == "H"
|
|
|
|
|
|
def test_parse_hilo_sorts_by_time():
|
|
reversed_data = list(reversed(RAW_HILO))
|
|
events = parse_hilo_predictions(reversed_data)
|
|
times = [e["dt"] for e in events]
|
|
assert times == sorted(times)
|
|
|
|
|
|
def test_parse_hilo_empty():
|
|
assert parse_hilo_predictions([]) == []
|
|
|
|
|
|
# --- classify_tidal_phase ---
|
|
|
|
|
|
def _events():
|
|
return parse_hilo_predictions(RAW_HILO)
|
|
|
|
|
|
def test_classify_ebb():
|
|
"""Between high (04:30) and low (10:42) → ebb."""
|
|
now = datetime(2026, 2, 21, 7, 30)
|
|
result = classify_tidal_phase(now, _events())
|
|
assert result["phase"] == "ebb"
|
|
assert result["previous"]["type"] == "high"
|
|
assert result["next"]["type"] == "low"
|
|
assert result["progress_pct"] is not None
|
|
assert 0 < result["progress_pct"] < 100
|
|
|
|
|
|
def test_classify_flood():
|
|
"""Between low (10:42) and high (16:55) → flood."""
|
|
now = datetime(2026, 2, 21, 13, 0)
|
|
result = classify_tidal_phase(now, _events())
|
|
assert result["phase"] == "flood"
|
|
assert result["previous"]["type"] == "low"
|
|
assert result["next"]["type"] == "high"
|
|
|
|
|
|
def test_classify_slack_high_after():
|
|
"""Within 30 min after high (04:30) → slack_high."""
|
|
now = datetime(2026, 2, 21, 4, 45) # 15 min after H
|
|
result = classify_tidal_phase(now, _events())
|
|
assert result["phase"] == "slack_high"
|
|
|
|
|
|
def test_classify_slack_high_before():
|
|
"""Within 30 min before high (16:55) → slack_high."""
|
|
now = datetime(2026, 2, 21, 16, 30) # 25 min before H
|
|
result = classify_tidal_phase(now, _events())
|
|
assert result["phase"] == "slack_high"
|
|
|
|
|
|
def test_classify_slack_low_after():
|
|
"""Within 30 min after low (10:42) → slack_low."""
|
|
now = datetime(2026, 2, 21, 11, 0) # 18 min after L
|
|
result = classify_tidal_phase(now, _events())
|
|
assert result["phase"] == "slack_low"
|
|
|
|
|
|
def test_classify_slack_low_before():
|
|
"""Within 30 min before low (23:08) → slack_low."""
|
|
now = datetime(2026, 2, 21, 22, 50) # 18 min before L
|
|
result = classify_tidal_phase(now, _events())
|
|
assert result["phase"] == "slack_low"
|
|
|
|
|
|
def test_classify_progress_midpoint():
|
|
"""At the midpoint between two events, progress should be ~50%."""
|
|
events = _events()
|
|
# Midpoint between 04:30 H and 10:42 L = ~07:36
|
|
mid = datetime(2026, 2, 21, 7, 36)
|
|
result = classify_tidal_phase(mid, events)
|
|
assert result["progress_pct"] is not None
|
|
assert 45 < result["progress_pct"] < 55
|
|
|
|
|
|
def test_classify_minutes_timing():
|
|
"""Check that minutes_since and minutes_to are reasonable."""
|
|
now = datetime(2026, 2, 21, 7, 30) # 3h after H at 04:30
|
|
result = classify_tidal_phase(now, _events())
|
|
assert result["minutes_since_previous"] == 180 # 3 hours
|
|
assert result["minutes_to_next"] is not None
|
|
assert result["minutes_to_next"] > 0
|
|
|
|
|
|
def test_classify_before_all_events():
|
|
"""Timestamp before all hilo events — no previous event."""
|
|
now = datetime(2026, 2, 21, 1, 0) # Before 04:30 H
|
|
result = classify_tidal_phase(now, _events())
|
|
assert result["previous"] is None
|
|
assert result["next"] is not None
|
|
assert result["phase"] in ("flood", "ebb")
|
|
|
|
|
|
def test_classify_after_all_events():
|
|
"""Timestamp after all hilo events — no next event."""
|
|
now = datetime(2026, 2, 22, 3, 0) # After 23:08 L
|
|
result = classify_tidal_phase(now, _events())
|
|
assert result["previous"] is not None
|
|
assert result["next"] is None
|
|
assert result["phase"] in ("flood", "ebb")
|
|
|
|
|
|
def test_classify_empty_events():
|
|
result = classify_tidal_phase(datetime(2026, 2, 21, 12, 0), [])
|
|
assert result["phase"] == "unknown"
|
|
|
|
|
|
# --- interpolate_predictions ---
|
|
|
|
|
|
def test_interpolate_midpoint():
|
|
"""Interpolation at midpoint should return average of bracketing values."""
|
|
times = [datetime(2026, 2, 21, 0, 0), datetime(2026, 2, 21, 1, 0)]
|
|
values = [2.0, 4.0]
|
|
result = interpolate_predictions(datetime(2026, 2, 21, 0, 30), times, values)
|
|
assert result is not None
|
|
assert abs(result - 3.0) < 0.01
|
|
|
|
|
|
def test_interpolate_at_boundary():
|
|
"""Interpolation exactly at a prediction point should return that value."""
|
|
times = [datetime(2026, 2, 21, 0, 0), datetime(2026, 2, 21, 1, 0)]
|
|
values = [2.0, 4.0]
|
|
result = interpolate_predictions(times[0], times, values)
|
|
assert result is not None
|
|
assert abs(result - 2.0) < 0.01
|
|
|
|
|
|
def test_interpolate_outside_window():
|
|
"""Timestamp outside prediction window should return None."""
|
|
times = [datetime(2026, 2, 21, 0, 0), datetime(2026, 2, 21, 1, 0)]
|
|
values = [2.0, 4.0]
|
|
assert interpolate_predictions(datetime(2026, 2, 20, 23, 0), times, values) is None
|
|
assert interpolate_predictions(datetime(2026, 2, 21, 2, 0), times, values) is None
|
|
|
|
|
|
def test_interpolate_multiple_segments():
|
|
"""Interpolation across multiple 6-min segments."""
|
|
times = [
|
|
datetime(2026, 2, 21, 0, 0),
|
|
datetime(2026, 2, 21, 0, 6),
|
|
datetime(2026, 2, 21, 0, 12),
|
|
]
|
|
values = [2.0, 3.0, 5.0]
|
|
# At 0:03 (midpoint of first segment) → 2.5
|
|
result = interpolate_predictions(datetime(2026, 2, 21, 0, 3), times, values)
|
|
assert result is not None
|
|
assert abs(result - 2.5) < 0.01
|
|
# At 0:09 (midpoint of second segment) → 4.0
|
|
result = interpolate_predictions(datetime(2026, 2, 21, 0, 9), times, values)
|
|
assert result is not None
|
|
assert abs(result - 4.0) < 0.01
|
|
|
|
|
|
def test_interpolate_insufficient_data():
|
|
"""Less than 2 prediction points → None."""
|
|
assert interpolate_predictions(datetime(2026, 2, 21, 0, 0), [], []) is None
|
|
assert interpolate_predictions(
|
|
datetime(2026, 2, 21, 0, 0),
|
|
[datetime(2026, 2, 21, 0, 0)],
|
|
[2.0],
|
|
) is None
|