import logging from typing import List, Optional from sqlalchemy.orm import Session from ..models.activity import Activity from ..models.bike_setup import BikeSetup logger = logging.getLogger(__name__) # Constants WHEEL_CIRCUMFERENCE_M = 2.1 # Approx 700x23c/28c generic TOLERANCE_PERCENT = 0.15 def calculate_observed_ratio(speed_mps: float, cadence_rpm: float) -> float: """ Calculate gear ratio from speed and cadence. Speed = (Cadence * Ratio * Circumference) / 60 Ratio = (Speed * 60) / (Cadence * Circumference) """ if not cadence_rpm or cadence_rpm == 0: return 0.0 return (speed_mps * 60) / (cadence_rpm * WHEEL_CIRCUMFERENCE_M) def match_activity_to_bike(db: Session, activity: Activity) -> Optional[BikeSetup]: """ Match an activity to a bike setup based on gear ratio. """ if not activity.activity_type: return None type_lower = activity.activity_type.lower() # Generic "cycling" check covers most (cycling, gravel_cycling, indoor_cycling) # But explicitly: 'road_biking', 'mountain_biking', 'gravel_cycling', 'cycling' # User asked for "all types of cycling". # We essentially want to filter OUT known non-cycling stuff if it doesn't match keys. # But safer to be inclusive of keywords. is_cycling = ( 'cycling' in type_lower or 'road_biking' in type_lower or 'mountain_biking' in type_lower or 'mtb' in type_lower or 'cyclocross' in type_lower ) if not is_cycling: # Not cycling return None if 'indoor' in type_lower: # Indoor cycling - ignore return None if not activity.avg_speed or not activity.avg_cadence: # Not enough data return None observed_ratio = calculate_observed_ratio(activity.avg_speed, activity.avg_cadence) if observed_ratio == 0: return None setups = db.query(BikeSetup).all() if not setups: return None best_match = None min_diff = float('inf') for setup in setups: if not setup.chainring or not setup.rear_cog: continue mechanical_ratio = setup.chainring / setup.rear_cog diff = abs(observed_ratio - mechanical_ratio) # Check tolerance # e.g., if ratio match is within 15% if diff / mechanical_ratio <= TOLERANCE_PERCENT: if diff < min_diff: min_diff = diff best_match = setup return best_match def process_activity_matching(db: Session, activity_id: int): """ Process matching for a specific activity and save result. """ activity = db.query(Activity).filter(Activity.id == activity_id).first() if not activity: return match = match_activity_to_bike(db, activity) if match: activity.bike_setup_id = match.id logger.info(f"Matched Activity {activity.id} to Setup {match.frame} (Found Ratio: {calculate_observed_ratio(activity.avg_speed, activity.avg_cadence):.2f})") else: # Implicitly "Generic" if None, but user requested explicit default logic. generic = db.query(BikeSetup).filter(BikeSetup.name == "GenericBike").first() if generic: activity.bike_setup_id = generic.id else: activity.bike_setup_id = None # Truly unknown db.commit() def run_matching_for_all(db: Session): """ Run matching for all activities that don't have a setup. """ from sqlalchemy import or_ activities = db.query(Activity).filter( Activity.bike_setup_id == None, or_( Activity.activity_type.ilike('%cycling%'), Activity.activity_type.ilike('%road_biking%'), Activity.activity_type.ilike('%mountain%'), # catch mountain_biking Activity.activity_type.ilike('%mtb%'), Activity.activity_type.ilike('%cyclocross%') ), Activity.activity_type.notilike('%indoor%') ).all() count = 0 for act in activities: process_activity_matching(db, act.id) count += 1 logger.info(f"Ran matching for {count} activities.")