Data Flow Architecture

import tif1

# Create session (no data loaded yet)
session = tif1.get_session(2025, "Monaco Grand Prix", "Race")

# Access laps property - triggers loading pipeline
laps = session.laps  # Returns pandas/polars DataFrame

# Access specific driver data
verstappen = session.get_driver("VER")
verstappen_laps = verstappen.laps  # Filtered view, no additional loading

# Access telemetry (separate loading pipeline)
fastest_lap = verstappen.get_fastest_lap()
telemetry = fastest_lap.get_telemetry()  # Triggers telemetry loading

# Control what data gets loaded
session = tif1.get_session(
    2025, "Monaco", "Race",
    laps=True,        # Load lap data
    telemetry=False,  # Skip telemetry (faster)
    weather=True,     # Load weather data
    messages=False    # Skip race control messages
)

# Choose backend
session = tif1.get_session(2025, "Monaco", "Race", lib="polars")  # Use polars
session = tif1.get_session(2025, "Monaco", "Race", lib="pandas")  # Use pandas (default)

# Lap data key
key = f"laps_{year}_{gp_name}_{session_type}_{backend}"
# Example: "laps_2025_monaco_race_pandas"

# Telemetry key
key = f"telemetry_{year}_{gp_name}_{session_type}_{driver}_{backend}"
# Example: "telemetry_2025_monaco_race_VER_pandas"

# Weather key
key = f"weather_{year}_{gp_name}_{session_type}"
# Example: "weather_2025_monaco_race"

# Typical memory footprint per cached item:
# - Lap data (20 drivers, 60 laps each): ~2-5 MB
# - Telemetry (single driver, full lap): ~10-20 MB
# - Weather data: ~0.1-0.5 MB
# - Race control messages: ~0.5-1 MB

# Total memory for 1024 items (worst case): ~10-20 GB
# Typical usage (mixed data): ~2-5 GB

# Good: Reuse session objects
session = tif1.get_session(2025, "Monaco", "Race")
for analysis in range(10):
    laps = session.laps  # Cache hit after first access
    # ... analysis code

# Bad: Create new sessions repeatedly
for analysis in range(10):
    session = tif1.get_session(2025, "Monaco", "Race")
    laps = session.laps  # Cache miss every time (different object)

CREATE TABLE IF NOT EXISTS cache (
    key TEXT PRIMARY KEY,
    value BLOB NOT NULL,           -- Compressed JSON
    timestamp REAL NOT NULL,       -- Unix timestamp
    size INTEGER,                  -- Uncompressed size in bytes
    access_count INTEGER DEFAULT 0,
    last_access REAL
);

CREATE INDEX IF NOT EXISTS idx_timestamp ON cache(timestamp);
CREATE INDEX IF NOT EXISTS idx_last_access ON cache(last_access);

import orjson
import zstandard

# Serialization (write to cache)
json_bytes = orjson.dumps(data)
compressed = zstandard.ZstdCompressor(level=1).compress(json_bytes)
# Typical compression ratio: ~4:1 (better than zlib-3, ~4.5x faster)

# Deserialization (read from cache)
json_bytes = zstandard.ZstdDecompressor().decompress(compressed)
data = orjson.loads(json_bytes)

# Default TTL: 7 days
# Configurable via TIF1_CACHE_TTL environment variable

# Check if cache entry is expired
import time
current_time = time.time()
entry_age = current_time - entry_timestamp
is_expired = entry_age > (7 * 24 * 60 * 60)  # 7 days in seconds

# Automatic cleanup on cache access
# Expired entries are removed lazily during lookups

cache = tif1.get_cache()

# Get cache information
print(f"Cache directory: {cache.cache_dir}")
print(f"Cache size: {cache.get_size_mb():.2f} MB")
print(f"Entry count: {cache.get_entry_count()}")
print(f"Hit rate: {cache.get_hit_rate():.2%}")

# Clear cache
cache.clear()  # Remove all entries
cache.clear_expired()  # Remove only expired entries
cache.clear_before(date)  # Remove entries older than date

def get_data(key):
    """Simplified cache lookup logic."""

    # Step 1: Check memory cache (L1)
    if key in memory_cache:
        logger.debug(f"Memory cache hit: {key}")
        return memory_cache[key]

    logger.debug(f"Memory cache miss: {key}")

    # Step 2: Check SQLite cache (L2)
    sqlite_data = sqlite_cache.get(key)
    if sqlite_data is not None:
        logger.debug(f"SQLite cache hit: {key}")

        # Deserialize and reconstruct DataFrame
        data = deserialize(sqlite_data)

        # Promote to memory cache (L1)
        memory_cache[key] = data

        return data

    logger.debug(f"SQLite cache miss: {key}")

    # Step 3: Fetch from CDN (cache miss)
    data = fetch_from_cdn(key)

    # Step 4: Save to both cache tiers
    sqlite_cache.set(key, serialize(data))
    memory_cache[key] = data

    return data

# Strategy 1: Pre-warm on application startup
def warm_cache_for_season(year):
    """Load all race data for a season into cache."""
    events = tif1.get_events(year)
    for event in events:
        for session_type in ["Practice 1", "Practice 2", "Practice 3", "Qualifying", "Race"]:
            try:
                session = tif1.get_session(year, event, session_type)
                _ = session.laps  # Trigger load
                logger.info(f"Cached: {year} {event} {session_type}")
            except tif1.DataNotFoundError:
                continue

# Strategy 2: Background cache warming
import threading

def warm_cache_background(year):
    """Warm cache in background thread."""
    thread = threading.Thread(target=warm_cache_for_season, args=(year,))
    thread.daemon = True
    thread.start()

# Strategy 3: Selective warming (only races)
def warm_cache_races_only(year):
    """Load only race sessions (fastest to load)."""
    events = tif1.get_events(year)
    for event in events:
        session = tif1.get_session(year, event, "Race")
        _ = session.laps

# Manual invalidation
cache = tif1.get_cache()
cache.invalidate(key)  # Remove specific entry

# Automatic invalidation (TTL-based)
# Entries older than 7 days are automatically removed

# Force refresh (bypass cache)
session = tif1.get_session(2025, "Monaco", "Race", force_refresh=True)
# Note: force_refresh not currently implemented, but planned

# Lap data URL construction
year = 2025
gp_name = "monaco"  # Normalized (lowercase, no spaces)
session_type = "race"  # Normalized
driver = "VER"

# Primary URL (jsDelivr)
base_url = f"https://cdn.jsdelivr.net/gh/TracingInsights/{year}@main"
lap_url = f"{base_url}/laps/{gp_name}/{session_type}/driver_{driver}.json"
# Result: https://cdn.staticdelivr.com/gh/TracingInsights/2025/main/laps/monaco/race/driver_VER.json

# Telemetry URL
telemetry_url = f"{base_url}/telemetry/{gp_name}/{session_type}/driver_{driver}_lap_{lap_number}.json"

# Weather URL
weather_url = f"{base_url}/weather/{gp_name}/{session_type}/weather.json"

# Race control messages URL
messages_url = f"{base_url}/messages/{gp_name}/{session_type}/messages.json"

async def fetch_with_fallback(url_path):
    """Fetch data with automatic CDN fallback."""

    # Attempt 1: jsDelivr CDN (primary)
    try:
        staticdelivr_url = f"https://cdn.staticdelivr.com/gh/TracingInsights/{year}/main/{url_path}"
        response = await http_client.get(staticdelivr_url, timeout=30)
        if response.status_code == 200:
            logger.info(f"StaticDelivr success: {url_path}")
            return response.content
        logger.warning(f"StaticDelivr returned {response.status_code}")
    except Exception as e:
        logger.warning(f"StaticDelivr failed: {e}")

    # Attempt 2: jsDelivr CDN (fallback)
    try:
        jsdelivr_url = f"https://cdn.jsdelivr.net/gh/TracingInsights/{year}@main/{url_path}"
        response = await http_client.get(jsdelivr_url, timeout=30)
        if response.status_code == 200:
            logger.info(f"jsDelivr success: {url_path}")
            return response.content
        logger.warning(f"jsDelivr returned {response.status_code}")
    except Exception as e:
        logger.warning(f"jsDelivr failed: {e}")

    # Attempt 3: Hugging Face bucket (backup)
    try:
        hf_url = f"https://huggingface.co/buckets/tracinginsights/{year}/resolve/{url_path}"
        response = await http_client.get(hf_url, timeout=30)
        if response.status_code == 200:
            logger.info(f"Hugging Face success: {url_path}")
            return response.content
        logger.warning(f"Hugging Face returned {response.status_code}")
    except Exception as e:
        logger.warning(f"Hugging Face failed: {e}")

    # All sources failed
    raise tif1.NetworkError(
        f"Failed to fetch {url_path} from all CDN sources",
        url=url_path,
        attempts=3
    )

class CircuitBreaker:
    """Circuit breaker for network requests."""

    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = "CLOSED"

    async def call(self, func, *args, **kwargs):
        """Execute function with circuit breaker protection."""

        # Check if circuit is open
        if self.state == "OPEN":
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = "HALF_OPEN"
                logger.info("Circuit breaker: OPEN → HALF_OPEN")
            else:
                raise tif1.NetworkError("Circuit breaker is OPEN")

        # Attempt request
        try:
            result = await func(*args, **kwargs)

            # Success - reset failure count
            if self.state == "HALF_OPEN":
                self.state = "CLOSED"
                logger.info("Circuit breaker: HALF_OPEN → CLOSED")
            self.failure_count = 0

            return result

        except Exception as e:
            # Failure - increment count
            self.failure_count += 1
            self.last_failure_time = time.time()

            if self.failure_count >= self.failure_threshold:
                self.state = "OPEN"
                logger.error(f"Circuit breaker: CLOSED → OPEN (failures: {self.failure_count})")

            raise

# Exponential backoff with jitter
max_retries = 3
base_delay = 1.0  # seconds

for attempt in range(max_retries):
    try:
        return await fetch_data(url)
    except NetworkError as e:
        if attempt == max_retries - 1:
            raise  # Final attempt failed

        # Calculate delay with exponential backoff and jitter
        delay = base_delay * (2 ** attempt)  # 1s, 2s, 4s
        jitter = random.uniform(0, 0.1 * delay)  # ±10% jitter
        total_delay = delay + jitter

        logger.warning(f"Retry {attempt + 1}/{max_retries} after {total_delay:.2f}s")
        await asyncio.sleep(total_delay)

# Timeout configuration
TIMEOUTS = {
    "connect": 10,      # Connection establishment timeout
    "read": 30,         # Read timeout (per chunk)
    "total": 60,        # Total request timeout
}

# Usage
async with http_client.get(url, timeout=TIMEOUTS) as response:
    content = await response.read()

# Sequential fetching - SLOW
drivers = ["VER", "HAM", "LEC", "NOR", "PIA", ...]  # 20 drivers
lap_data = []

for driver in drivers:
    url = construct_url(driver)
    response = requests.get(url)  # Blocking call
    data = response.json()
    lap_data.append(data)

# Time: 20 drivers × 500ms = 10 seconds

# Parallel fetching - FAST
import asyncio
import niquests

async def fetch_all_drivers(drivers):
    """Fetch data for all drivers in parallel."""

    async def fetch_driver(driver):
        url = construct_url(driver)
        async with http_client.get(url) as response:
            return await response.json()

    # Create tasks for all drivers
    tasks = [fetch_driver(driver) for driver in drivers]

    # Execute all tasks concurrently
    results = await asyncio.gather(*tasks, return_exceptions=True)

    return results

# Time: max(500ms across all drivers) ≈ 500-800ms
# Speedup: 10-15x faster

# HTTP/2 connection reuse
async with niquests.AsyncSession() as session:
    # Single TCP connection established
    # All subsequent requests reuse this connection

    tasks = []
    for driver in drivers:
        task = session.get(construct_url(driver))
        tasks.append(task)

    # All requests multiplexed over single connection
    responses = await asyncio.gather(*tasks)

# Connection pool configuration
http_client = niquests.AsyncSession(
    pool_connections=10,      # Number of connection pools
    pool_maxsize=100,         # Max connections per pool
    pool_block=False,         # Don't block when pool is full
    max_redirects=3,          # Follow up to 3 redirects
    timeout=30,               # Default timeout
)

# Connection reuse
# First request: TCP + TLS handshake (100-200ms overhead)
# Subsequent requests: No handshake (0ms overhead)

async def fetch_with_progress(drivers):
    """Fetch data with progress tracking."""

    total = len(drivers)
    completed = 0

    async def fetch_and_track(driver):
        nonlocal completed
        try:
            data = await fetch_driver(driver)
            completed += 1
            progress = (completed / total) * 100
            logger.info(f"Progress: {progress:.1f}% ({completed}/{total})")
            return data
        except Exception as e:
            logger.error(f"Failed to fetch {driver}: {e}")
            completed += 1
            return None

    tasks = [fetch_and_track(driver) for driver in drivers]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    return results

async def fetch_with_error_handling(drivers):
    """Fetch data with robust error handling."""

    async def fetch_safe(driver):
        """Fetch with exception handling."""
        try:
            return await fetch_driver(driver)
        except tif1.NetworkError as e:
            logger.warning(f"Network error for {driver}: {e}")
            return None
        except tif1.InvalidDataError as e:
            logger.error(f"Invalid data for {driver}: {e}")
            return None
        except Exception as e:
            logger.error(f"Unexpected error for {driver}: {e}")
            return None

    # gather with return_exceptions=True prevents one failure from canceling others
    results = await asyncio.gather(
        *[fetch_safe(driver) for driver in drivers],
        return_exceptions=True
    )

    # Filter out None results (failed fetches)
    valid_results = [r for r in results if r is not None]

    success_rate = len(valid_results) / len(drivers)
    logger.info(f"Fetch success rate: {success_rate:.1%}")

    return valid_results

# For very large batches, split into smaller chunks to avoid overwhelming the server
async def fetch_in_batches(drivers, batch_size=10):
    """Fetch data in batches to control concurrency."""

    results = []

    for i in range(0, len(drivers), batch_size):
        batch = drivers[i:i + batch_size]
        logger.info(f"Fetching batch {i // batch_size + 1} ({len(batch)} drivers)")

        batch_results = await fetch_all_drivers(batch)
        results.extend(batch_results)

        # Optional: Small delay between batches to be respectful to CDN
        if i + batch_size < len(drivers):
            await asyncio.sleep(0.1)

    return results

import time
import asyncio

# Scenario: Load lap data for all 20 drivers in Monaco 2025 Race

# Sequential approach (traditional)
start = time.time()
for driver in drivers:
    data = fetch_driver_sync(driver)  # 500ms each
sequential_time = time.time() - start
# Result: ~10 seconds

# Parallel approach (tif1)
start = time.time()
data = asyncio.run(fetch_all_drivers(drivers))
parallel_time = time.time() - start
# Result: ~0.6 seconds

speedup = sequential_time / parallel_time
print(f"Speedup: {speedup:.1f}x faster")
# Output: Speedup: 16.7x faster

import orjson

async def parse_json_response(response_bytes):
    """Parse JSON response with validation."""

    # Step 1: Parse JSON bytes to Python dict
    try:
        data = orjson.loads(response_bytes)
    except orjson.JSONDecodeError as e:
        raise tif1.InvalidDataError(
            f"Failed to parse JSON: {e}",
            raw_data=response_bytes[:1000]  # First 1KB for debugging
        )

    # Step 2: Validate structure
    if not isinstance(data, dict):
        raise tif1.InvalidDataError(
            f"Expected dict, got {type(data).__name__}",
            data_type=type(data).__name__
        )

    # Step 3: Check required fields
    required_fields = ["laps", "metadata"]
    missing_fields = [f for f in required_fields if f not in data]
    if missing_fields:
        raise tif1.InvalidDataError(
            f"Missing required fields: {missing_fields}",
            missing=missing_fields,
            available=list(data.keys())
        )

    return data

{
  "metadata": {
    "year": 2025,
    "grand_prix": "Monaco Grand Prix",
    "session_type": "Race",
    "driver": "VER",
    "total_laps": 78,
    "generated_at": "2025-05-25T15:30:00Z"
  },
  "laps": [
    {
      "lap_number": 1,
      "lap_time": 95.234,
      "sector_1_time": 28.456,
      "sector_2_time": 35.123,
      "sector_3_time": 31.655,
      "speed_i1": 285.4,
      "speed_i2": 312.7,
      "speed_fl": 298.3,
      "speed_st": 276.8,
      "compound": "SOFT",
      "tyre_life": 1,
      "stint": 1,
      "is_personal_best": false,
      "position": 1,
      "track_status": "1",
      "is_accurate": true,
      "deleted": false,
      "deleted_reason": null
    },
    // ... more laps
  ]
}

{
  "metadata": {
    "year": 2025,
    "grand_prix": "Monaco Grand Prix",
    "session_type": "Race",
    "driver": "VER",
    "lap_number": 45,
    "samples": 15234,
    "frequency": 50
  },
  "telemetry": [
    {
      "time": 0.0,
      "distance": 0.0,
      "speed": 285.4,
      "rpm": 11250,
      "gear": 8,
      "throttle": 100,
      "brake": 0,
      "drs": 0,
      "x": 1234.56,
      "y": 5678.90,
      "z": 12.34
    },
    // ... 15,000+ samples
  ]
}

{
  "metadata": {
    "year": 2025,
    "grand_prix": "Monaco Grand Prix",
    "session_type": "Race",
    "samples": 156
  },
  "weather": [
    {
      "time": "2025-05-25T14:00:00Z",
      "air_temp": 28.5,
      "track_temp": 42.3,
      "humidity": 45,
      "pressure": 1013.2,
      "wind_speed": 3.2,
      "wind_direction": 180,
      "rainfall": false
    },
    // ... more samples
  ]
}

# For very large telemetry files (100MB+), consider streaming
import ijson  # Iterative JSON parser

def parse_large_telemetry(file_path):
    """Stream parse large telemetry files."""
    with open(file_path, 'rb') as f:
        # Parse telemetry array incrementally
        telemetry_points = ijson.items(f, 'telemetry.item')

        # Process in chunks
        chunk_size = 10000
        chunk = []

        for point in telemetry_points:
            chunk.append(point)

            if len(chunk) >= chunk_size:
                yield chunk
                chunk = []

        if chunk:
            yield chunk

# For memory-constrained environments
def parse_with_memory_limit(response_bytes, max_memory_mb=100):
    """Parse JSON with memory limit check."""

    # Estimate memory usage (rough approximation)
    estimated_memory = len(response_bytes) * 3  # JSON → Python objects ≈ 3x
    estimated_mb = estimated_memory / (1024 * 1024)

    if estimated_mb > max_memory_mb:
        raise tif1.InvalidDataError(
            f"Data too large: {estimated_mb:.1f}MB (limit: {max_memory_mb}MB)",
            size_mb=estimated_mb,
            limit_mb=max_memory_mb
        )

    return orjson.loads(response_bytes)

def parse_with_recovery(response_bytes):
    """Parse JSON with error recovery."""

    try:
        # Attempt normal parsing
        return orjson.loads(response_bytes)

    except orjson.JSONDecodeError as e:
        # Try to identify and fix common issues

        # Issue 1: Trailing commas
        if "trailing comma" in str(e).lower():
            logger.warning("Attempting to fix trailing commas")
            fixed = response_bytes.replace(b',]', b']').replace(b',}', b'}')
            return orjson.loads(fixed)

        # Issue 2: Incomplete JSON (truncated response)
        if "unexpected end" in str(e).lower():
            logger.error("JSON appears truncated - re-fetching")
            raise tif1.NetworkError("Incomplete response received")

        # Issue 3: Invalid UTF-8
        if "utf" in str(e).lower():
            logger.warning("Attempting UTF-8 error recovery")
            text = response_bytes.decode('utf-8', errors='replace')
            return orjson.loads(text.encode('utf-8'))

        # Unrecoverable error
        raise tif1.InvalidDataError(
            f"JSON parsing failed: {e}",
            error=str(e),
            position=e.pos if hasattr(e, 'pos') else None
        )

from pydantic import BaseModel, Field, validator
from typing import List, Optional

class LapData(BaseModel):
    """Pydantic model for lap data validation."""

    lap_number: int = Field(ge=1, le=100)
    lap_time: Optional[float] = Field(None, gt=0)
    sector_1_time: Optional[float] = Field(None, gt=0)
    sector_2_time: Optional[float] = Field(None, gt=0)
    sector_3_time: Optional[float] = Field(None, gt=0)
    compound: str = Field(pattern=r'^(SOFT|MEDIUM|HARD|INTERMEDIATE|WET)$')
    position: int = Field(ge=1, le=20)

    @validator('lap_time')
    def validate_lap_time(cls, v, values):
        """Ensure lap time matches sum of sectors."""
        if v is not None and all(k in values for k in ['sector_1_time', 'sector_2_time', 'sector_3_time']):
            sector_sum = values['sector_1_time'] + values['sector_2_time'] + values['sector_3_time']
            if abs(v - sector_sum) > 0.1:  # Allow 100ms tolerance
                raise ValueError(f"Lap time {v} doesn't match sector sum {sector_sum}")
        return v

class LapDataResponse(BaseModel):
    """Complete lap data response."""
    metadata: dict
    laps: List[LapData]

# Usage
def validate_lap_data(data):
    """Validate parsed lap data."""
    try:
        validated = LapDataResponse(**data)
        return validated.dict()
    except ValidationError as e:
        raise tif1.InvalidDataError(
            f"Data validation failed: {e}",
            errors=e.errors()
        )

def construct_lap_dataframe(lap_data, backend="pandas"):
    """Construct DataFrame from validated lap data."""

    # Step 1: Create initial DataFrame
    if backend == "pandas":
        import pandas as pd
        df = pd.DataFrame(lap_data["laps"])
    elif backend == "polars":
        import polars as pl
        df = pl.DataFrame(lap_data["laps"])
    else:
        raise ValueError(f"Unknown backend: {backend}")

    # Step 2: Rename columns (snake_case → PascalCase)
    df = df.rename(columns=COLUMN_RENAME_MAP)

    # Step 3: Set data types
    df = optimize_dtypes(df)

    # Step 4: Reorder columns
    df = df[COLUMN_ORDER]

    # Step 5: Set index (optional)
    if backend == "pandas":
        df = df.set_index("LapNumber")

    return df

COLUMN_RENAME_MAP = {
    # Lap identification
    "lap_number": "LapNumber",
    "driver": "Driver",
    "team": "Team",

    # Timing
    "lap_time": "LapTime",
    "sector_1_time": "Sector1Time",
    "sector_2_time": "Sector2Time",
    "sector_3_time": "Sector3Time",

    # Speed traps
    "speed_i1": "SpeedI1",
    "speed_i2": "SpeedI2",
    "speed_fl": "SpeedFL",
    "speed_st": "SpeedST",

    # Tyre information
    "compound": "Compound",
    "tyre_life": "TyreLife",
    "stint": "Stint",

    # Position and status
    "position": "Position",
    "track_status": "TrackStatus",

    # Flags
    "is_personal_best": "IsPersonalBest",
    "is_accurate": "IsAccurate",
    "deleted": "Deleted",
    "deleted_reason": "DeletedReason",
}

def optimize_dtypes_pandas(df):
    """Optimize pandas DataFrame dtypes for memory efficiency."""

    # Numeric columns - use smallest safe type
    numeric_optimizations = {
        "LapNumber": "uint8",        # 1-100 laps
        "Sector1Time": "float32",    # Sufficient precision
        "Sector2Time": "float32",
        "Sector3Time": "float32",
        "LapTime": "float32",
        "SpeedI1": "float32",
        "SpeedI2": "float32",
        "SpeedFL": "float32",
        "SpeedST": "float32",
        "TyreLife": "uint8",         # 1-50 laps
        "Stint": "uint8",            # 1-5 stints
        "Position": "uint8",         # 1-20 positions
    }

    for col, dtype in numeric_optimizations.items():
        if col in df.columns:
            df[col] = df[col].astype(dtype)

    # Categorical columns - huge memory savings
    categorical_columns = [
        "Driver",        # 20 unique values
        "Team",          # 10 unique values
        "Compound",      # 5 unique values
        "TrackStatus",   # 4 unique values
    ]

    for col in categorical_columns:
        if col in df.columns:
            df[col] = df[col].astype("category")

    # Boolean columns
    boolean_columns = ["IsPersonalBest", "IsAccurate", "Deleted"]
    for col in boolean_columns:
        if col in df.columns:
            df[col] = df[col].astype(bool)

    return df

# Memory savings example:
# Before optimization: 15 MB
# After optimization: 4 MB (73% reduction)

def optimize_dtypes_polars(df):
    """Optimize polars DataFrame dtypes."""

    import polars as pl

    # Polars has better default type inference, but we can still optimize
    type_mapping = {
        "LapNumber": pl.UInt8,
        "Sector1Time": pl.Float32,
        "Sector2Time": pl.Float32,
        "Sector3Time": pl.Float32,
        "LapTime": pl.Float32,
        "Driver": pl.Categorical,
        "Team": pl.Categorical,
        "Compound": pl.Categorical,
        "TrackStatus": pl.Categorical,
        "IsPersonalBest": pl.Boolean,
        "IsAccurate": pl.Boolean,
        "Deleted": pl.Boolean,
    }

    for col, dtype in type_mapping.items():
        if col in df.columns:
            df = df.with_columns(pl.col(col).cast(dtype))

    return df

COLUMN_ORDER = [
    # Identification (first)
    "LapNumber",
    "Driver",
    "Team",

    # Timing (core data)
    "LapTime",
    "Sector1Time",
    "Sector2Time",
    "Sector3Time",

    # Speed traps
    "SpeedI1",
    "SpeedI2",
    "SpeedFL",
    "SpeedST",

    # Tyre strategy
    "Compound",
    "TyreLife",
    "Stint",

    # Position
    "Position",

    # Status flags
    "TrackStatus",
    "IsPersonalBest",
    "IsAccurate",

    # Metadata (last)
    "Deleted",
    "DeletedReason",
]

def reorder_columns(df, column_order):
    """Reorder DataFrame columns."""
    # Only include columns that exist in the DataFrame
    ordered_cols = [col for col in column_order if col in df.columns]

    # Add any remaining columns not in the order list
    remaining_cols = [col for col in df.columns if col not in ordered_cols]

    return df[ordered_cols + remaining_cols]

# Option 1: LapNumber as index (default)
df = df.set_index("LapNumber")
# Pros: Fast lap lookup, natural ordering
# Cons: Loses LapNumber as regular column

# Option 2: MultiIndex (Driver + LapNumber)
df = df.set_index(["Driver", "LapNumber"])
# Pros: Fast driver + lap lookup, hierarchical grouping
# Cons: More complex indexing

# Option 3: RangeIndex (default)
# Pros: Simple, fast integer indexing
# Cons: No semantic meaning

# tif1 uses Option 1 by default

# Polars doesn't have traditional indexes
# Instead, use efficient filtering and sorting

# Fast lap lookup
lap_45 = df.filter(pl.col("LapNumber") == 45)

# Fast driver lookup
verstappen = df.filter(pl.col("Driver") == "VER")

# Combined lookup
verstappen_lap_45 = df.filter(
    (pl.col("Driver") == "VER") & (pl.col("LapNumber") == 45)
)

def handle_missing_data(df):
    """Handle missing data appropriately."""

    # Strategy 1: Fill with sentinel values
    df["LapTime"] = df["LapTime"].fillna(-1.0)  # -1 indicates missing

    # Strategy 2: Forward fill (for cumulative data)
    df["Position"] = df["Position"].fillna(method="ffill")

    # Strategy 3: Interpolate (for continuous data)
    df["SpeedI1"] = df["SpeedI1"].interpolate(method="linear")

    # Strategy 4: Leave as NaN (for optional data)
    # DeletedReason can be NaN when Deleted=False

    return df

def validate_dataframe(df):
    """Validate DataFrame structure and content."""

    # Check required columns
    required_columns = ["LapNumber", "Driver", "LapTime"]
    missing_columns = [col for col in required_columns if col not in df.columns]
    if missing_columns:
        raise tif1.InvalidDataError(
            f"Missing required columns: {missing_columns}",
            missing=missing_columns
        )

    # Check data ranges
    if (df["LapNumber"] < 1).any() or (df["LapNumber"] > 100).any():
        raise tif1.InvalidDataError("LapNumber out of valid range (1-100)")

    if (df["Position"] < 1).any() or (df["Position"] > 20).any():
        raise tif1.InvalidDataError("Position out of valid range (1-20)")

    # Check for duplicates
    duplicates = df.duplicated(subset=["Driver", "LapNumber"])
    if duplicates.any():
        dup_count = duplicates.sum()
        raise tif1.InvalidDataError(
            f"Found {dup_count} duplicate lap entries",
            duplicate_count=dup_count
        )

    # Check data consistency
    # Lap time should approximately equal sum of sectors
    df["SectorSum"] = df["Sector1Time"] + df["Sector2Time"] + df["Sector3Time"]
    inconsistent = (df["LapTime"] - df["SectorSum"]).abs() > 0.5
    if inconsistent.any():
        logger.warning(f"Found {inconsistent.sum()} laps with inconsistent sector times")

    return df

def enrich_lap_times(df):
    """Add computed time columns."""

    # LapTimeSeconds - helper for time-based calculations
    df["LapTimeSeconds"] = df["LapTime"]

    # Sector percentages
    df["Sector1Percent"] = (df["Sector1Time"] / df["LapTime"]) * 100
    df["Sector2Percent"] = (df["Sector2Time"] / df["LapTime"]) * 100
    df["Sector3Percent"] = (df["Sector3Time"] / df["LapTime"]) * 100

    # Delta to personal best
    personal_best = df.groupby("Driver")["LapTime"].transform("min")
    df["DeltaToPersonalBest"] = df["LapTime"] - personal_best

    # Delta to session best
    session_best = df["LapTime"].min()
    df["DeltaToSessionBest"] = df["LapTime"] - session_best

    # Cumulative time
    df["CumulativeTime"] = df.groupby("Driver")["LapTime"].cumsum()

    return df

def enrich_position_data(df):
    """Add position-related computed columns."""

    # Position changes
    df["PositionChange"] = df.groupby("Driver")["Position"].diff()
    df["StartPosition"] = df.groupby("Driver")["Position"].transform("first")
    df["CurrentPositionChange"] = df["StartPosition"] - df["Position"]

    # Gaps (requires sorting by position within each lap)
    df = df.sort_values(["LapNumber", "Position"])
    df["GapToLeader"] = df.groupby("LapNumber")["CumulativeTime"].transform(
        lambda x: x - x.iloc[0]
    )
    df["GapToAhead"] = df.groupby("LapNumber")["CumulativeTime"].diff()

    return df

def enrich_tyre_data(df):
    """Add tyre strategy computed columns."""

    # Stint identification (already in data, but validate)
    df["Stint"] = (df.groupby("Driver")["Compound"].shift() != df["Compound"]).groupby(df["Driver"]).cumsum() + 1

    # Stint length
    df["StintLength"] = df.groupby(["Driver", "Stint"]).cumcount() + 1

    # Tyre age at lap start
    df["TyreAge"] = df["TyreLife"]

    # Compound history
    df["PreviousCompound"] = df.groupby("Driver")["Compound"].shift()

    # Pit stop detection
    df["IsPitLap"] = df.groupby("Driver")["Stint"].diff() == 1

    # Laps since pit
    df["LapsSincePit"] = df.groupby(["Driver", "Stint"]).cumcount()

    return df

def enrich_performance_flags(df):
    """Add performance-related boolean flags."""

    # Personal best lap
    df["IsPersonalBest"] = df.groupby("Driver")["LapTime"].transform(
        lambda x: x == x.min()
    )

    # Session best lap
    df["IsSessionBest"] = df["LapTime"] == df["LapTime"].min()

    # Top 3 lap
    df["IsTop3Lap"] = df["LapTime"] <= df["LapTime"].nsmallest(3).max()

    # Outlier detection (lap time > 3 std dev from mean)
    mean_time = df.groupby("Driver")["LapTime"].transform("mean")
    std_time = df.groupby("Driver")["LapTime"].transform("std")
    df["IsOutlier"] = (df["LapTime"] - mean_time).abs() > (3 * std_time)

    # Consistent lap (within 0.5s of personal average)
    df["IsConsistent"] = (df["LapTime"] - mean_time).abs() < 0.5

    return df

def merge_weather_data(lap_df, weather_df):
    """Merge weather data with lap data."""

    # Weather data is sampled every minute
    # Need to match each lap to closest weather sample

    # Convert lap times to timestamps
    lap_df["Timestamp"] = lap_df["LapStartTime"]  # Assuming this exists

    # Merge using nearest timestamp
    lap_df = pd.merge_asof(
        lap_df.sort_values("Timestamp"),
        weather_df.sort_values("Timestamp"),
        on="Timestamp",
        direction="nearest",
        suffixes=("", "_weather")
    )

    # Add weather-related computed columns
    lap_df["TrackTempChange"] = lap_df.groupby("Driver")["TrackTemp"].diff()
    lap_df["AirTempChange"] = lap_df.groupby("Driver")["AirTemp"].diff()

    # Weather condition categories
    lap_df["WeatherCondition"] = "Dry"
    lap_df.loc[lap_df["Rainfall"] == True, "WeatherCondition"] = "Wet"
    lap_df.loc[lap_df["TrackTemp"] < 20, "WeatherCondition"] = "Cold"
    lap_df.loc[lap_df["TrackTemp"] > 50, "WeatherCondition"] = "Hot"

    return lap_df

def enrich_telemetry_acceleration(tel_df):
    """Calculate acceleration from speed data."""

    # Time delta between samples (typically 0.02s for 50Hz)
    tel_df["TimeDelta"] = tel_df["Time"].diff()

    # Speed delta
    tel_df["SpeedDelta"] = tel_df["Speed"].diff()

    # Acceleration (m/s²)
    # Convert km/h to m/s: speed / 3.6
    # Acceleration = (v2 - v1) / dt
    tel_df["Acceleration"] = (
        (tel_df["SpeedDelta"] / 3.6) / tel_df["TimeDelta"]
    )

    # Lateral acceleration (from X, Y coordinates)
    tel_df["XDelta"] = tel_df["X"].diff()
    tel_df["YDelta"] = tel_df["Y"].diff()
    tel_df["LateralAcceleration"] = (
        ((tel_df["XDelta"]**2 + tel_df["YDelta"]**2)**0.5) / tel_df["TimeDelta"]**2
    )

    # G-force (1g = 9.81 m/s²)
    tel_df["AccelerationG"] = tel_df["Acceleration"] / 9.81
    tel_df["LateralG"] = tel_df["LateralAcceleration"] / 9.81

    return tel_df

def normalize_telemetry_distance(tel_df):
    """Normalize distance to 0.0-1.0 range."""

    # Original distance is in meters
    max_distance = tel_df["Distance"].max()
    tel_df["NormalizedDistance"] = tel_df["Distance"] / max_distance

    # Percentage through lap
    tel_df["LapPercentage"] = tel_df["NormalizedDistance"] * 100

    return tel_df

def add_driver_ahead_info(tel_df, lap_df):
    """Add information about driver ahead."""

    # Get position from lap data
    position = lap_df.loc[lap_df["LapNumber"] == tel_df["LapNumber"].iloc[0], "Position"].iloc[0]

    if position > 1:
        # Find driver ahead
        driver_ahead = lap_df.loc[
            (lap_df["LapNumber"] == tel_df["LapNumber"].iloc[0]) &
            (lap_df["Position"] == position - 1),
            "Driver"
        ].iloc[0]

        tel_df["DriverAhead"] = driver_ahead
    else:
        tel_df["DriverAhead"] = None

    return tel_df

def detect_corners(tel_df, speed_threshold=200):
    """Detect corners based on speed and steering."""

    # Corner = low speed + high steering angle
    # Approximate steering from lateral G
    tel_df["IsCorner"] = (
        (tel_df["Speed"] < speed_threshold) &
        (tel_df["LateralG"].abs() > 1.5)
    )

    # Corner number (sequential numbering)
    tel_df["CornerNumber"] = (
        tel_df["IsCorner"].diff() == 1
    ).cumsum()

    # Only keep corner number where IsCorner=True
    tel_df.loc[~tel_df["IsCorner"], "CornerNumber"] = None

    return tel_df

def save_to_cache(key, data):
    """Save data to both cache tiers."""

    # Step 1: Save to SQLite (persistent)
    try:
        sqlite_cache.set(key, data)
        logger.debug(f"Saved to SQLite cache: {key}")
    except Exception as e:
        logger.error(f"Failed to save to SQLite: {e}")
        # Continue even if SQLite save fails

    # Step 2: Save to memory (fast access)
    try:
        memory_cache[key] = data
        logger.debug(f"Saved to memory cache: {key}")
    except Exception as e:
        logger.error(f"Failed to save to memory: {e}")

    return data

import orjson
import zlib

def serialize_for_cache(df, backend="pandas"):
    """Serialize DataFrame for cache storage."""

    if backend == "pandas":
        # Convert DataFrame to dict (orient='split' for efficiency)
        data_dict = {
            "data": df.to_dict(orient="split"),
            "backend": "pandas",
            "version": "1.0",
            "timestamp": time.time()
        }
    elif backend == "polars":
        # Convert to dict
        data_dict = {
            "data": df.to_dict(as_series=False),
            "backend": "polars",
            "version": "1.0",
            "timestamp": time.time()
        }

    # Serialize to JSON
    json_bytes = orjson.dumps(data_dict)

    # Compress
    compressed = zlib.compress(json_bytes, level=6)

    logger.debug(f"Serialization: {len(json_bytes)} bytes → {len(compressed)} bytes "
                f"({len(compressed)/len(json_bytes)*100:.1f}% of original)")

    return compressed

def deserialize_from_cache(compressed_data):
    """Deserialize DataFrame from cache storage."""

    # Decompress
    json_bytes = zlib.decompress(compressed_data)

    # Parse JSON
    data_dict = orjson.loads(json_bytes)

    # Reconstruct DataFrame
    backend = data_dict["backend"]

    if backend == "pandas":
        import pandas as pd
        df = pd.DataFrame(**data_dict["data"])
    elif backend == "polars":
        import polars as pl
        df = pl.DataFrame(data_dict["data"])
    else:
        raise ValueError(f"Unknown backend: {backend}")

    return df

class CacheEntry:
    """Metadata for cache entry."""

    def __init__(self, key, data, metadata=None):
        self.key = key
        self.data = data
        self.created_at = time.time()
        self.accessed_at = time.time()
        self.access_count = 0
        self.size_bytes = len(serialize_for_cache(data))
        self.metadata = metadata or {}

    def access(self):
        """Record cache access."""
        self.accessed_at = time.time()
        self.access_count += 1

    def is_expired(self, ttl_seconds=604800):  # 7 days default
        """Check if entry is expired."""
        age = time.time() - self.created_at
        return age > ttl_seconds

    def to_dict(self):
        """Convert to dictionary for storage."""
        return {
            "key": self.key,
            "created_at": self.created_at,
            "accessed_at": self.accessed_at,
            "access_count": self.access_count,
            "size_bytes": self.size_bytes,
            "metadata": self.metadata
        }

from functools import lru_cache
from collections import OrderedDict

class LRUCache:
    """LRU cache with size limit."""

    def __init__(self, max_size=1024):
        self.cache = OrderedDict()
        self.max_size = max_size

    def get(self, key):
        """Get item from cache."""
        if key not in self.cache:
            return None

        # Move to end (most recently used)
        self.cache.move_to_end(key)
        return self.cache[key]

    def set(self, key, value):
        """Set item in cache."""
        if key in self.cache:
            # Update existing item
            self.cache.move_to_end(key)
        else:
            # Add new item
            if len(self.cache) >= self.max_size:
                # Evict least recently used
                evicted_key, evicted_value = self.cache.popitem(last=False)
                logger.debug(f"Evicted from cache: {evicted_key}")

        self.cache[key] = value

    def clear(self):
        """Clear all items."""
        self.cache.clear()

def cleanup_expired_entries(cache, ttl_seconds=604800):
    """Remove expired entries from SQLite cache."""

    current_time = time.time()
    cutoff_time = current_time - ttl_seconds

    # SQL query to delete old entries
    query = "DELETE FROM cache WHERE timestamp < ?"

    cursor = cache.conn.execute(query, (cutoff_time,))
    deleted_count = cursor.rowcount

    logger.info(f"Cleaned up {deleted_count} expired cache entries")

    # Vacuum database to reclaim space
    cache.conn.execute("VACUUM")

    return deleted_count

class CacheStatistics:
    """Track cache performance statistics."""

    def __init__(self):
        self.hits = 0
        self.misses = 0
        self.evictions = 0
        self.errors = 0
        self.total_bytes_read = 0
        self.total_bytes_written = 0

    def record_hit(self, size_bytes=0):
        """Record cache hit."""
        self.hits += 1
        self.total_bytes_read += size_bytes

    def record_miss(self):
        """Record cache miss."""
        self.misses += 1

    def record_eviction(self):
        """Record cache eviction."""
        self.evictions += 1

    def record_error(self):
        """Record cache error."""
        self.errors += 1

    def get_hit_rate(self):
        """Calculate cache hit rate."""
        total = self.hits + self.misses
        return self.hits / total if total > 0 else 0.0

    def get_stats(self):
        """Get all statistics."""
        return {
            "hits": self.hits,
            "misses": self.misses,
            "hit_rate": self.get_hit_rate(),
            "evictions": self.evictions,
            "errors": self.errors,
            "bytes_read": self.total_bytes_read,
            "bytes_written": self.total_bytes_written,
        }

    def reset(self):
        """Reset all statistics."""
        self.__init__()

def finalize_and_return(df, session_info):
    """Finalize DataFrame and return to user."""

    # Step 1: Final validation
    validate_dataframe(df)

    # Step 2: Add metadata attributes (pandas only)
    if hasattr(df, 'attrs'):
        df.attrs['session_info'] = session_info
        df.attrs['loaded_at'] = time.time()
        df.attrs['tif1_version'] = tif1.__version__

    # Step 3: Log completion
    logger.info(
        f"Data loading complete: {len(df)} rows, "
        f"{len(df.columns)} columns, "
        f"{df.memory_usage(deep=True).sum() / 1024 / 1024:.2f} MB"
    )

    # Step 4: Return DataFrame
    return df

{
  "metadata": {
    "year": 2025,
    "grand_prix": "Monaco Grand Prix",
    "session_type": "Race",
    "driver": "VER",
    "total_laps": 78,
    "generated_at": "2025-05-25T15:30:00Z",
    "data_version": "2.0"
  },
  "laps": [
    {
      "lap_number": 1,
      "lap_time": 95.234,
      "sector_1_time": 28.456,
      "sector_2_time": 35.123,
      "sector_3_time": 31.655,
      "speed_i1": 285.4,
      "speed_i2": 312.7,
      "speed_fl": 298.3,
      "speed_st": 276.8,
      "compound": "SOFT",
      "tyre_life": 1,
      "stint": 1,
      "is_personal_best": false,
      "position": 1,
      "track_status": "1",
      "is_accurate": true,
      "deleted": false,
      "deleted_reason": null
    },
    {
      "lap_number": 2,
      "lap_time": 93.567,
      "sector_1_time": 27.234,
      "sector_2_time": 34.456,
      "sector_3_time": 31.877,
      "speed_i1": 287.2,
      "speed_i2": 314.5,
      "speed_fl": 299.8,
      "speed_st": 278.3,
      "compound": "SOFT",
      "tyre_life": 2,
      "stint": 1,
      "is_personal_best": true,
      "position": 1,
      "track_status": "1",
      "is_accurate": true,
      "deleted": false,
      "deleted_reason": null
    }
  ]
}