139 lines
5.0 KiB
Python
139 lines
5.0 KiB
Python
|
|
import sys
|
|
import os
|
|
import json
|
|
import logging
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
# Add backend to path
|
|
sys.path.append(os.path.join(os.getcwd(), 'backend'))
|
|
|
|
from src.models.activity import Activity
|
|
from src.models.segment import Segment
|
|
from src.utils.geo import haversine_distance, calculate_bounds
|
|
from src.services.parsers import extract_points_from_file
|
|
from src.services.segment_matcher import SegmentMatcher
|
|
from src.utils.config import config
|
|
|
|
# Setup DB
|
|
engine = create_engine(config.DATABASE_URL)
|
|
Session = sessionmaker(bind=engine)
|
|
db = Session()
|
|
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger("SegmentTest")
|
|
|
|
def test_segment_splitting(activity_garmin_id):
|
|
print(f"--- Segment Splitting Test: Activity {activity_garmin_id} ---")
|
|
|
|
activity = db.query(Activity).filter(Activity.garmin_activity_id == activity_garmin_id).first()
|
|
if not activity:
|
|
print("Activity not found")
|
|
return
|
|
|
|
if not activity.file_content:
|
|
print("No file content")
|
|
return
|
|
|
|
points = extract_points_from_file(activity.file_content, activity.file_type)
|
|
print(f"Total Points: {len(points)}")
|
|
|
|
if len(points) < 2:
|
|
print("Not enough points")
|
|
return
|
|
|
|
# Split into 1-mile segments (1609.34 meters)
|
|
MILE_IN_METERS = 1609.34
|
|
|
|
segments_to_test = []
|
|
|
|
current_segment_points = [points[0]]
|
|
current_dist = 0.0
|
|
seg_count = 1
|
|
|
|
# Simple splitting logic
|
|
for i in range(1, len(points)):
|
|
p1 = points[i-1]
|
|
p2 = points[i]
|
|
d = haversine_distance(p1[1], p1[0], p2[1], p2[0])
|
|
current_dist += d
|
|
current_segment_points.append(p2)
|
|
|
|
if current_dist >= MILE_IN_METERS:
|
|
# Finalize this segment
|
|
# Ensure it has enough points? Yes, if it's a mile long.
|
|
|
|
seg_name = f"Test_{seg_count}"
|
|
|
|
# Create a mock Segment object (not saving to DB to avoid pollution, unless needed by matcher?)
|
|
# Matcher queries DB for segments. So we probably have to save them, or mock the query.
|
|
# The user asked to "create segments named Test_...".
|
|
# Ideally we check logic without DB writes, but Matcher implementation:
|
|
# segments = self.db.query(Segment)...
|
|
|
|
# So we must persist them temporarily or modify matcher to accept list.
|
|
# Let's persist and delete after?
|
|
# Or just persist them as requested "create... segments".
|
|
|
|
# We will create meaningful Segment objects in memory and inject them into the matcher logic?
|
|
# No, Matcher.match_activity queries the DB.
|
|
# I will manually invoke _match_segment which takes specific objects.
|
|
|
|
segments_to_test.append({
|
|
"name": seg_name,
|
|
"points": current_segment_points,
|
|
"distance": current_dist
|
|
})
|
|
|
|
# Reset for next segment
|
|
# Start next segment from current point (overlap 1 point)
|
|
current_segment_points = [p2]
|
|
current_dist = 0.0
|
|
seg_count += 1
|
|
|
|
print(f"Created {len(segments_to_test)} mock segments.")
|
|
|
|
matcher = SegmentMatcher(db)
|
|
|
|
# Test each segment
|
|
success_count = 0
|
|
|
|
for mock_seg in segments_to_test:
|
|
print(f"\nTesting {mock_seg['name']} ({mock_seg['distance']:.2f}m)...")
|
|
|
|
# Create a transient Segment object
|
|
seg_obj = Segment(
|
|
id=9999 + int(mock_seg['name'].split('_')[1]), # Fake ID
|
|
name=mock_seg['name'],
|
|
activity_type=activity.activity_type,
|
|
points=json.dumps(mock_seg['points']), # Matcher needs serialized or list?
|
|
# Matcher: seg_points = json.loads(segment.points) if isinstance(segment.points, str) else segment.points
|
|
# So list is fine if we pass it directly to _match_segment
|
|
distance=mock_seg['distance'],
|
|
bounds=json.dumps(calculate_bounds(mock_seg['points']))
|
|
)
|
|
# Note: We pass list directly to _match_segment, but Matcher.match_activity queries DB.
|
|
# We will bypass match_activity lookup and call _match_segment directly.
|
|
|
|
# NOTE: _match_segment signature:
|
|
# def _match_segment(self, segment: Segment, seg_points: List[List[float]], activity: Activity, act_points: List[List[float]]) -> Optional[Tuple[int, int]]:
|
|
|
|
try:
|
|
indices = matcher._match_segment(seg_obj, mock_seg['points'], activity, points)
|
|
|
|
if indices:
|
|
s, e = indices
|
|
print(f" [PASS] Matched! Activity indexes {s} to {e}")
|
|
success_count += 1
|
|
else:
|
|
print(f" [FAIL] No match found.")
|
|
|
|
except Exception as e:
|
|
print(f" [ERROR] {e}")
|
|
|
|
print(f"\nSummary: {success_count}/{len(segments_to_test)} segments matched.")
|
|
|
|
if __name__ == "__main__":
|
|
test_segment_splitting("21368342318")
|