Utilities - tif1

Beyond the core timing and telemetry API, tif1 provides utility functions. They manage internal state, performance optimization, reliability layers, and system configuration. These utilities give fine-grained control over caching behavior, network resilience, logging verbosity, data processing pipelines, and more. This guide is an exhaustive reference for all utility functions. It covers parameters, return values, edge cases, performance implications, thread safety, and real-world usage patterns. Use these utilities when debugging data loading issues, optimizing cache performance, or building production applications.

Overview

The utilities in tif1 are organized into several categories:

Logging: Control diagnostic output verbosity
Configuration: Manage global settings and preferences
Cache Management: Control persistent and in-memory caching
Data Utilities: Convert and manipulate time and data structures
Reliability & Networking: Manage circuit breakers and CDN fallback
Memory Management: Clear caches to free resources

All utilities are designed with performance in mind and are thread-safe where applicable.

Logging

Logging is essential for debugging data loading issues, understanding cache behavior, and diagnosing network problems. The tif1 library uses Python’s standard logging module and provides a convenient setup function to control verbosity across all internal components.

`setup_logging`

Configure the logging level for all tif1 components.

def setup_logging(level: int = logging.WARNING) -> None

This function configures the root logger and sets the level for the tif1 logger hierarchy. It also suppresses noisy connection pool warnings from the underlying HTTP library. Parameters:

level (int): Standard logging level from Python’s logging module. Defaults to logging.WARNING.

Available Levels:

Level	Value	Description	Use Case
`logging.DEBUG`	10	Detailed diagnostic information including cache hits/misses, HTTP requests, data parsing steps	Development, debugging data loading issues
`logging.INFO`	20	General informational messages about major operations	Production monitoring, understanding data flow
`logging.WARNING`	30	Warning messages about potential issues (default)	Normal operation, alerts for degraded performance
`logging.ERROR`	40	Error messages for failures that don’t stop execution	Production error tracking
`logging.CRITICAL`	50	Critical errors that may cause application failure	Severe production issues

Behavior:

Configures logging format: %(asctime)s - %(name)s - %(levelname)s - %(message)s
Sets date format: %Y-%m-%d %H:%M:%S
Applies to all tif1.* loggers (core, cache, http, retry, cdn, etc.)
Suppresses urllib3_future.connectionpool warnings (set to ERROR level)
Thread-safe: Can be called from any thread

Basic Example:

import tif1
import logging

# Enable debug logging to see everything
tif1.setup_logging(logging.DEBUG)

# Now all tif1 operations will log detailed diagnostic info
session = tif1.get_session(2021, "Belgian Grand Prix", "Race")
laps = session.laps

Debug Output Example:

2024-03-15 14:23:45 - tif1.core - DEBUG - Loading session: 2021/Belgian_Grand_Prix/Race
2024-03-15 14:23:45 - tif1.cache - DEBUG - Checking cache for 2021/Belgian_Grand_Prix/Race/drivers.json
2024-03-15 14:23:45 - tif1.cache - DEBUG - Cache hit for drivers.json
2024-03-15 14:23:45 - tif1.http - DEBUG - Fetching https://cdn.jsdelivr.net/gh/TracingInsights/2021@main/Belgian_Grand_Prix/Race/laps.json
2024-03-15 14:23:46 - tif1.http - DEBUG - Response: 200 OK (1.2s, 245KB)
2024-03-15 14:23:46 - tif1.core - DEBUG - Parsing lap data: 1,234 laps
2024-03-15 14:23:46 - tif1.core - DEBUG - Loaded 20 drivers

Selective Logging Example:

import logging

# Only show warnings and errors
tif1.setup_logging(logging.WARNING)

# Or use INFO for production monitoring
tif1.setup_logging(logging.INFO)

# Disable all tif1 logging
tif1.setup_logging(logging.CRITICAL + 1)

Advanced: Per-Module Logging:

import logging

# Set global level
tif1.setup_logging(logging.INFO)

# But enable DEBUG for specific modules
logging.getLogger("tif1.cache").setLevel(logging.DEBUG)
logging.getLogger("tif1.http").setLevel(logging.DEBUG)

# Now only cache and HTTP operations show debug output

Performance Considerations:

DEBUG logging adds ~5-10% overhead due to string formatting
INFO logging adds ~1-2% overhead
WARNING and above have negligible performance impact
Logging is buffered and asynchronous in most Python implementations

Best Practices:

Use DEBUG during development to understand data flow
Use INFO in production for operational visibility
Use WARNING (default) for minimal overhead
Disable logging in performance-critical tight loops
Consider using a logging handler that writes to files for production

Configuration

The tif1 configuration system provides a flexible way to customize library behavior through environment variables, configuration files, or programmatic API calls. Configuration is managed through a singleton Config object that loads settings from multiple sources with a defined precedence order.

Configuration Loading Order

Settings are loaded in the following order (later sources override earlier ones):

Built-in defaults: Hardcoded sensible defaults
Configuration file (~/.tif1rc): JSON file in user’s home directory
Environment variables: Prefixed with TIF1_ (for example, TIF1_LIB=polars)
Programmatic calls: Using config.set() at runtime

`get_config`

Returns the global configuration singleton instance.

def get_config() -> Config

The Config object is a singleton, meaning all calls to get_config() return the same instance. This ensures configuration consistency across the entire application. Returns:

Config: Singleton configuration instance

Thread Safety:

The Config singleton is thread-safe
Multiple threads can safely call get() and set()
File operations (save()) use atomic writes

Basic Example:

import tif1

config = tif1.get_config()

# Get current settings
print(f"DataFrame library: {config.get('lib')}")
print(f"Cache enabled: {config.get('enable_cache')}")
print(f"Request timeout: {config.get('timeout')}s")
print(f"Max retries: {config.get('max_retries')}")

# Update settings at runtime
config.set("lib", "polars")
config.set("timeout", 60)

# Persist changes to disk
config.save()  # Saves to ~/.tif1rc

Config Methods

`get(key: str, default: Any = None) -> Any`

Retrieve a configuration value by key. Parameters:

key (str): Configuration key name
default (Any, optional): Value to return if key does not exist. Defaults to None.

Returns:

The configuration value, or default if key not found

Example:

config = tif1.get_config()

# Get with default fallback
lib = config.get("lib", "pandas")
timeout = config.get("timeout", 30)
custom = config.get("my_custom_key", "default_value")

# Check if key exists
if config.get("ultra_cold_start") is not None:
    print("Ultra cold start is configured")

`set(key: str, value: Any) -> None`

Update a configuration value in memory. Parameters:

key (str): Configuration key name
value (Any): New value to set

Behavior:

Changes are immediate and affect all subsequent operations
Changes are in-memory only until save() is called
Type conversion is automatic for known keys
Unknown keys are stored as-is

Example:

config = tif1.get_config()

# Set DataFrame backend
config.set("lib", "polars")  # or "pandas"

# Configure caching
config.set("enable_cache", True)
config.set("cache_dir", "/custom/cache/path")

# Network settings
config.set("timeout", 60)
config.set("max_retries", 5)

# Performance tuning
config.set("ultra_cold_start", True)
config.set("validate_data", False)

# Circuit breaker settings
config.set("circuit_breaker_threshold", 10)
config.set("circuit_breaker_timeout", 120)

`save(path: Path | str | None = None) -> None`

Save current configuration to a JSON file. Parameters:

path (Path | str | None, optional): File path to save to. If None, saves to ~/.tif1rc. Defaults to None.

Behavior:

Creates parent directories if they do not exist
Uses atomic write (write to temp file, then rename)
Saves all current settings, including defaults
File format is human-readable JSON

Example:

config = tif1.get_config()

# Save to default location (~/.tif1rc)
config.save()

# Save to custom location
config.save("/path/to/custom/config.json")
config.save("./project_tif1_config.json")

# Save to user's config directory
from pathlib import Path
config.save(Path.home() / ".config" / "tif1" / "config.json")

Configuration File Format:

{
  "lib": "polars",
  "enable_cache": true,
  "cache_dir": "<platform-specific default>",
  "timeout": 60,
  "max_retries": 5,
  "validate_data": false,
  "ultra_cold_start": true,
  "circuit_breaker_threshold": 10,
  "circuit_breaker_timeout": 120,
  "cdns": [
    "https://cdn.staticdelivr.com/gh/TracingInsights",
    "https://cdn.jsdelivr.net/gh/TracingInsights",
    "https://huggingface.co/buckets/tracinginsights"
  ]
}

Available Configuration Keys

Reference of all configuration options:

Key	Type	Default	Description	Performance Impact
`lib`	`str`	`"pandas"`	DataFrame library: `"pandas"` or `"polars"`	Polars is 2-5x faster for large datasets
`enable_cache`	`bool`	`True`	Enable multi-layer caching system	Disabling eliminates cache overhead but requires re-fetching
`cache_dir`	`str`	OS-dependent	SQLite cache database location	SSD recommended for best performance
`timeout`	`int`	`30`	Network request timeout in seconds	Increase for slow connections
`max_retries`	`int`	`3`	Maximum retry attempts for failed requests	More retries = better reliability, slower failures
`validate_data`	`bool`	`True`	Enable Pydantic validation of JSON data	Disabling saves ~10-15% processing time
`ultra_cold_start`	`bool`	`False`	Skip loading full session data, fetch on-demand	Dramatically faster for single lap queries
`circuit_breaker_threshold`	`int`	`5`	Failures before circuit breaker opens	Lower = faster failure detection
`circuit_breaker_timeout`	`int`	`60`	Circuit breaker reset timeout in seconds	Time to wait before retrying failed service
`cdns`	`list[str]`	`["https://cdn.staticdelivr.com/gh/TracingInsights", "https://cdn.jsdelivr.net/gh/TracingInsights", "https://huggingface.co/buckets/tracinginsights"]`	List of CDN base URLs for fallback	Multiple CDNs improve reliability
`cdn_use_minification`	`bool`	`False`	Use minified JSON files from CDN	Reduces bandwidth by 20-40%

Environment Variable Configuration

All configuration keys can be set via environment variables using the TIF1_ prefix:

# Set DataFrame library
export TIF1_LIB=polars

# Configure caching
export TIF1_ENABLE_CACHE=true
export TIF1_CACHE_DIR=/custom/cache/path

# Network settings
export TIF1_TIMEOUT=60
export TIF1_MAX_RETRIES=5

# Performance tuning
export TIF1_ULTRA_COLD_START=true
export TIF1_VALIDATE_DATA=false

# Circuit breaker
export TIF1_CIRCUIT_BREAKER_THRESHOLD=10
export TIF1_CIRCUIT_BREAKER_TIMEOUT=120

Type Conversion:

Booleans: "true", "1", "yes" → True; "false", "0", "no" → False
Integers: Parsed automatically
Lists: Comma-separated values (for example, "url1,url2,url3")

Configuration Examples

Development Configuration

import tif1

config = tif1.get_config()

# Fast iteration, detailed logging
config.set("lib", "pandas")  # Easier debugging
config.set("validate_data", True)  # Catch data issues
config.set("timeout", 10)  # Fail fast
config.set("enable_cache", True)  # Speed up repeated queries

config.save()

Production Configuration

import tif1

config = tif1.get_config()

# Maximum performance and reliability
config.set("lib", "polars")  # Best performance
config.set("validate_data", False)  # Skip validation overhead
config.set("timeout", 60)  # Handle slow networks
config.set("max_retries", 5)  # More resilient
config.set("circuit_breaker_threshold", 10)  # Tolerate more failures
config.set("cdn_use_minification", True)  # Reduce bandwidth

config.save()

Minimal Latency Configuration

import tif1

config = tif1.get_config()

# Optimize for single lap telemetry queries
config.set("ultra_cold_start", True)  # Skip full session load
config.set("lib", "polars")  # Faster processing
config.set("validate_data", False)  # Skip validation
config.set("enable_cache", True)  # Cache telemetry

config.save()

Custom Cache Location

import tif1
from pathlib import Path

config = tif1.get_config()

# Use project-specific cache
project_cache = Path(__file__).parent / ".cache" / "tif1"
config.set("cache_dir", str(project_cache))

# Or use system temp directory
import tempfile
temp_cache = Path(tempfile.gettempdir()) / "tif1_cache"
config.set("cache_dir", str(temp_cache))

config.save()

Configuration Best Practices

Use environment variables for deployment: Easier to configure across environments without code changes
Save configuration after tuning: Persist the optimized settings with config.save()
Use polars for production: Better performance and memory efficiency
Disable validation in production: Saves ~10-15% processing time if data quality is trusted
Increase timeout for slow networks: Default 30s may not be enough on mobile/satellite connections
Use ultra-cold start for single lap queries: Faster when only one lap’s telemetry is needed
Configure the circuit breaker for each use case: Lower threshold for fail-fast, higher for resilience
Use multiple CDNs for reliability: Configure fallback CDNs in production

Cache Management

The tif1 caching system is a performance component that reduces data loading times and network bandwidth usage. It implements a multi-layer caching strategy with both in-memory (LRU) and persistent (SQLite) storage.

Cache Architecture

The cache system consists of three layers:

In-memory LRU cache: Fast access to recently used DataFrames (separate caches for pandas and polars)
SQLite persistent cache: Disk-based storage for JSON responses and telemetry data
Automatic cache warming: Prefetches commonly accessed data

`get_cache`

Returns the persistent SQLite cache singleton instance.

def get_cache() -> Cache

The Cache object manages the SQLite database that stores fetched JSON data and telemetry. Like Config, it is a singleton ensuring consistent cache state across the application. Returns:

Cache: Singleton cache instance

Thread Safety:

The Cache singleton is thread-safe
Uses connection pooling for concurrent access
Automatic transaction management with periodic commits
Safe for use in multi-threaded applications

Basic Example:

import tif1

cache = tif1.get_cache()

# Inspect cache location
print(f"Cache directory: {cache.cache_dir}")
print(f"Cache database: {cache.cache_dir / 'cache.db'}")

# Check if specific session is cached
has_data = cache.has_session_data(2025, "Monaco_Grand_Prix", "Race")
print(f"Monaco GP cached: {has_data}")

# Clear all cached data
cache.clear()
print("Cache cleared")

Cache Methods

`get(key: str) -> Any | None`

Retrieve raw cached data by key. Parameters:

key (str): Cache key (typically in format year/gp/session/file.json)

Returns:

Cached data (usually dict or list), or None if not found

Behavior:

Checks in-memory cache first (O(1) lookup)
Falls back to SQLite if not in memory (O(log n) lookup)
Automatically deserializes JSON data
Updates in-memory cache on SQLite hit

Example:

cache = tif1.get_cache()

# Get drivers data
drivers = cache.get("2025/Monaco_Grand_Prix/Race/drivers.json")
if drivers:
    print(f"Found {len(drivers)} drivers in cache")

# Get laps data
laps = cache.get("2025/Monaco_Grand_Prix/Race/laps.json")
if laps:
    print(f"Found {len(laps)} laps in cache")

`set(key: str, data: Any) -> None`

Store data in cache. Parameters:

key (str): Cache key
data (Any): Data to cache (must be JSON-serializable)

Behavior:

Stores in both memory and SQLite
Automatically serializes to JSON
Commits to disk periodically (every 10 writes) or on close
Overwrites existing data for the same key

Example:

cache = tif1.get_cache()

# Cache custom data
cache.set("custom/analysis/results.json", {
    "fastest_lap": "1:23.456",
    "driver": "VER",
    "session": "Q3"
})

# Retrieve it later
results = cache.get("custom/analysis/results.json")

`get_telemetry(year: int, gp: str, session: str, driver: str, lap: int) -> Any | None`

Get cached telemetry data for a specific lap. Parameters:

year (int): Season year
gp (str): Grand Prix name (underscored format)
session (str): Session type (for example, “Race”, “Qualifying”)
driver (str): Driver code (for example, “VER”, “HAM”)
lap (int): Lap number

Returns:

Telemetry data dict, or None if not cached

Behavior:

Uses optimized key format: {year}/{gp}/{session}/telemetry/{driver}/{lap}
Checks memory cache first for ultra-fast access
Falls back to SQLite for persistent storage
Returns raw telemetry dict (not DataFrame)

Example:

cache = tif1.get_cache()

# Check if specific lap telemetry is cached
tel = cache.get_telemetry(2021, "Belgian_Grand_Prix", "Race", "VER", 1)
if tel:
    print(f"Cached telemetry has {len(tel)} samples")
else:
    print("Telemetry not cached, will fetch from CDN")

`set_telemetry(year: int, gp: str, session: str, driver: str, lap: int, data: Any) -> None`

Store telemetry data in cache. Parameters:

year (int): Season year
gp (str): Grand Prix name (underscored format)
session (str): Session type
driver (str): Driver code
lap (int): Lap number
data (Any): Telemetry data to cache

Behavior:

Stores in both memory and SQLite
Uses optimized storage format
Automatically commits periodically
Enables fast subsequent access

Example:

cache = tif1.get_cache()

# Manually cache telemetry (usually done automatically)
telemetry_data = {
    "Time": [0.0, 0.1, 0.2],
    "Speed": [0, 50, 100],
    "Throttle": [0, 50, 100]
}
cache.set_telemetry(2021, "Belgian_Grand_Prix", "Race", "VER", 1, telemetry_data)

`get_telemetry_batch(year: int, gp: str, session: str, driver: str, laps: list[int]) -> dict[int, Any]`

Get multiple laps of telemetry in a single operation. Parameters:

year (int): Season year
gp (str): Grand Prix name
session (str): Session type
driver (str): Driver code
laps (list[int]): List of lap numbers

Returns:

Dict mapping lap number to telemetry data (only includes cached laps)

Performance:

Single SQLite query for all laps (much faster than individual queries)
Optimized for bulk operations
Returns only cached laps (missing laps are omitted)

Example:

cache = tif1.get_cache()

# Get telemetry for multiple laps at once
laps_to_fetch = [1, 2, 3, 4, 5]
cached_telemetry = cache.get_telemetry_batch(
    2021, "Belgian_Grand_Prix", "Race", "VER", laps_to_fetch
)

print(f"Found {len(cached_telemetry)} of {len(laps_to_fetch)} laps in cache")

# Process cached laps
for lap_num, tel_data in cached_telemetry.items():
    print(f"Lap {lap_num}: {len(tel_data)} samples")

`has_session_data(year: int, gp: str, session: str) -> bool`

Check if cache contains data for a specific session. Parameters:

year (int): Season year
gp (str): Grand Prix name (underscored format)
session (str): Session type

Returns:

True if session data exists in cache, False otherwise

Behavior:

Checks for presence of key session files (drivers.json, laps.json)
Fast operation (index lookup only)
Useful for pre-flight checks before loading

Example:

cache = tif1.get_cache()

# Check multiple sessions
sessions = ["FP1", "FP2", "FP3", "Qualifying", "Race"]
for session_type in sessions:
    cached = cache.has_session_data(2025, "Monaco_Grand_Prix", session_type)
    status = "✓ cached" if cached else "✗ not cached"
    print(f"{session_type}: {status}")

# Conditional loading
if not cache.has_session_data(2025, "Monaco_Grand_Prix", "Race"):
    print("Data not cached, will fetch from CDN...")

`clear() -> None`

Remove all entries from both memory and SQLite cache. Behavior:

Clears in-memory LRU cache
Deletes all rows from SQLite tables
Vacuums database to reclaim disk space
Resets all cache statistics
Thread-safe operation

Example:

cache = tif1.get_cache()

# Clear all cached data
cache.clear()
print("All cache data cleared")

# Verify cache is empty
has_data = cache.has_session_data(2025, "Monaco_Grand_Prix", "Race")
assert not has_data, "Cache should be empty"

Use Cases:

Free disk space
Force re-fetch of all data
Clear corrupted cache entries
Reset after configuration changes

`close() -> None`

Close the SQLite database connection and flush pending writes. Behavior:

Commits any pending transactions
Closes database connection
Releases file locks
Should be called before application exit

Example:

import atexit
import tif1

cache = tif1.get_cache()

# Ensure cache is closed on exit
atexit.register(cache.close)

# Or manually close
try:
    # ... your code ...
    pass
finally:
    cache.close()

Cache Performance Characteristics

Operation	Memory Cache	SQLite Cache	Network Fetch
Lap data lookup	~0.001ms	~1-5ms	~100-500ms
Telemetry lookup	~0.001ms	~2-10ms	~50-200ms
Batch telemetry (10 laps)	~0.01ms	~5-20ms	~500-2000ms
Session data check	~0.001ms	~1ms	N/A
Cache clear	~1ms	~10-50ms	N/A

Cache Size Management

The cache grows over time as more sessions are loaded. Here is how to manage cache size: Check Cache Size:

import tif1
from pathlib import Path

cache = tif1.get_cache()
cache_db = Path(cache.cache_dir) / "cache.db"

if cache_db.exists():
    size_mb = cache_db.stat().st_size / (1024 * 1024)
    print(f"Cache size: {size_mb:.2f} MB")

Typical Cache Sizes:

Single session: ~5-15 MB
Full season (all sessions): ~500-800 MB
Multiple seasons: ~2-5 GB

Cache Cleanup Strategies:

import tif1
from pathlib import Path

cache = tif1.get_cache()

# Strategy 1: Clear everything
cache.clear()

# Strategy 2: Delete cache database file
cache.close()
cache_db = Path(cache.cache_dir) / "cache.db"
if cache_db.exists():
    cache_db.unlink()

# Strategy 3: Selective clearing (manual SQL)
# Not recommended - use clear() instead

Async Cache Operations

The cache also supports async operations for use in async applications:

import asyncio
import tif1

async def fetch_cached_data():
    cache = tif1.get_cache()

    # Async get
    data = await cache.get_async("2025/Monaco_Grand_Prix/Race/laps.json")

    # Async set
    await cache.set_async("custom/key", {"data": "value"})

    # Async telemetry get
    tel = await cache.get_telemetry_async(
        2021, "Belgian_Grand_Prix", "Race", "VER", 1
    )

    # Async batch telemetry
    batch = await cache.get_telemetry_batch_async(
        2021, "Belgian_Grand_Prix", "Race", "VER", [1, 2, 3]
    )

    return data

# Run async code
asyncio.run(fetch_cached_data())

Cache Best Practices

Let the cache warm up: First load is slow, subsequent loads are fast
Do not clear cache unnecessarily: The cache improves performance on repeated loads
Monitor cache size: Set up periodic cleanup if disk space is limited
Use batch operations: get_telemetry_batch() is much faster than individual calls
Close cache on exit: Use atexit.register(cache.close) to ensure clean shutdown
Check cache before loading: Use has_session_data() to avoid unnecessary network calls
Use SSD for cache: Faster than HDD for SQLite operations
Consider cache location: Put cache on fast local storage, not network drives

Data Utilities

The data utilities provide convenient functions for converting between different time formats and safely navigating nested data structures. These are particularly useful when working with lap times, timestamps, and JSON responses.

`to_timedelta`

Convert various time formats to pandas Timedelta objects.

def to_timedelta(x: str | float | int | timedelta) -> pd.Timedelta

This function provides a unified interface for converting lap times, sector times, and other duration values into pandas Timedelta objects. The converted values support consistent time arithmetic and comparison. Parameters:

x (str | float | int | timedelta): Time value to convert

Supported Input Formats:

Input Type	Format	Example	Result
String (MM:SS.mmm)	`"M:SS.mmm"`	`"1:23.456"`	1 min 23.456 sec
String (HH:MM:SS.mmm)	`"HH:MM:SS.mmm"`	`"00:01:23.456"`	1 min 23.456 sec
Float (seconds)	Decimal seconds	`83.456`	83.456 seconds
Integer (seconds)	Whole seconds	`83`	83 seconds
Timedelta	Passthrough	`pd.Timedelta(seconds=83.456)`	Unchanged

Behavior:

Automatically adds leading "00:" to "M:SS.mmm" format
Handles negative times (for example, for deltas)
Preserves microsecond precision
Returns pandas Timedelta for consistent arithmetic

Basic Examples:

from tif1.utils import to_timedelta
import pandas as pd

# From lap time string (M:SS.mmm format)
lap_time = to_timedelta("1:23.456")
print(lap_time)  # 0 days 00:01:23.456000

# From full time string (HH:MM:SS.mmm format)
full_time = to_timedelta("00:01:23.456")
print(full_time)  # 0 days 00:01:23.456000

# From seconds (float)
from_seconds = to_timedelta(83.456)
print(from_seconds)  # 0 days 00:01:23.456000

# From seconds (int)
from_int = to_timedelta(83)
print(from_int)  # 0 days 00:01:23

# Passthrough timedelta
td = pd.Timedelta(seconds=83.456)
result = to_timedelta(td)
assert result == td

Time Arithmetic:

from tif1.utils import to_timedelta

# Calculate lap time difference
lap1 = to_timedelta("1:23.456")
lap2 = to_timedelta("1:24.123")
delta = lap2 - lap1
print(f"Lap 2 was {delta.total_seconds():.3f}s slower")  # 0.667s slower

# Sum sector times
sector1 = to_timedelta("0:28.123")
sector2 = to_timedelta("0:32.456")
sector3 = to_timedelta("0:23.789")
total = sector1 + sector2 + sector3
print(f"Total lap time: {total}")  # 0 days 00:01:24.368000

# Calculate average lap time
lap_times = [to_timedelta("1:23.456"), to_timedelta("1:24.123"), to_timedelta("1:23.789")]
avg = sum(lap_times, pd.Timedelta(0)) / len(lap_times)
print(f"Average: {avg}")

Working with DataFrames:

import pandas as pd
from tif1.utils import to_timedelta

# Convert lap time column
df = pd.DataFrame({
    "Lap": [1, 2, 3],
    "LapTime": ["1:23.456", "1:24.123", "1:23.789"]
})

df["LapTimeDelta"] = df["LapTime"].apply(to_timedelta)
print(df["LapTimeDelta"].mean())  # Average lap time

Edge Cases:

from tif1.utils import to_timedelta

# Very fast lap (under 1 minute)
fast = to_timedelta("0:58.123")

# Very slow lap (over 2 minutes)
slow = to_timedelta("2:15.456")

# Negative delta (for time differences)
negative = to_timedelta("-0:01.234")

# Zero time
zero = to_timedelta("0:00.000")

`to_datetime`

Convert string timestamps to pandas Timestamp objects.

def to_datetime(x: str) -> pd.Timestamp

This function wraps pandas’ to_datetime for consistent datetime parsing across the library. Parameters:

x (str): Datetime string to convert

Supported Formats:

ISO 8601: "2021-08-29T14:00:00"
Standard: "2021-08-29 14:00:00"
Date only: "2021-08-29"
With timezone: "2021-08-29T14:00:00+02:00"

Examples:

from tif1.utils import to_datetime

# ISO 8601 format
dt1 = to_datetime("2021-08-29T14:00:00")
print(dt1)  # 2021-08-29 14:00:00

# Standard format
dt2 = to_datetime("2021-08-29 14:00:00")
print(dt2)  # 2021-08-29 14:00:00

# Date only
dt3 = to_datetime("2021-08-29")
print(dt3)  # 2021-08-29 00:00:00

# With timezone
dt4 = to_datetime("2021-08-29T14:00:00+02:00")
print(dt4)  # 2021-08-29 14:00:00+02:00

Working with Session Times:

from tif1.utils import to_datetime
import tif1

session = tif1.get_session(2021, "Belgian Grand Prix", "Race")

# Parse session start time
if hasattr(session, 'session_start_time'):
    start_time = to_datetime(session.session_start_time)
    print(f"Session started at: {start_time}")

`recursive_dict_get`

Safely navigate nested dictionaries without raising KeyError.

def recursive_dict_get(d: dict, *keys: str, default_none: bool = False) -> Any

This function provides safe access to deeply nested dictionary values. It returns a default value instead of raising exceptions when keys do not exist. Parameters:

d (dict): Dictionary to navigate
*keys (str): Keys to traverse in order
default_none (bool, optional): If True, return None for missing keys; if False, return {}. Defaults to False.

Returns:

The value at the nested key path, or default value if any key is missing

Basic Examples:

from tif1.utils import recursive_dict_get

data = {
    "session": {
        "info": {
            "name": "Monaco Grand Prix",
            "year": 2025
        },
        "weather": {
            "temp": 24,
            "humidity": 65
        }
    }
}

# Successful navigation
name = recursive_dict_get(data, "session", "info", "name")
print(name)  # "Monaco Grand Prix"

year = recursive_dict_get(data, "session", "info", "year")
print(year)  # 2025

# Missing key with default_none=False (returns {})
missing = recursive_dict_get(data, "session", "missing", "key")
print(missing)  # {}
print(type(missing))  # <class 'dict'>

# Missing key with default_none=True (returns None)
missing = recursive_dict_get(data, "session", "missing", "key", default_none=True)
print(missing)  # None

Practical Use Cases:

from tif1.utils import recursive_dict_get

# Parse API response safely
api_response = {
    "data": {
        "session": {
            "laps": [
                {"driver": "VER", "time": "1:23.456"},
                {"driver": "HAM", "time": "1:23.789"}
            ]
        }
    }
}

# Safe navigation
laps = recursive_dict_get(api_response, "data", "session", "laps")
if laps:
    print(f"Found {len(laps)} laps")

# Handle missing optional data
telemetry = recursive_dict_get(api_response, "data", "session", "telemetry", default_none=True)
if telemetry is None:
    print("No telemetry data available")

Comparison with Standard Access:

from tif1.utils import recursive_dict_get

data = {"a": {"b": {"c": "value"}}}

# Standard access (raises KeyError)
try:
    value = data["a"]["missing"]["c"]
except KeyError:
    value = None

# Using recursive_dict_get (no exception)
value = recursive_dict_get(data, "a", "missing", "c", default_none=True)
# Returns None without exception

`delta_time`

Calculate delta time between two laps (FastF1 compatibility function).

def delta_time(reference_lap: Lap, compare_lap: Lap) -> tuple[pd.Series, Telemetry, Telemetry]

This function provides compatibility with FastF1’s delta_time utility for comparing lap performance. Parameters:

reference_lap (Lap): Reference lap object
compare_lap (Lap): Lap object to compare against reference

Returns:

Tuple of (delta_series, ref_telemetry, comp_telemetry)
- delta_series: Time delta at each telemetry point
- ref_telemetry: Reference lap telemetry
- comp_telemetry: Comparison lap telemetry

Example:

import tif1

session = tif1.get_session(2021, "Belgian Grand Prix", "Qualifying")
laps = session.laps

# Get fastest laps for two drivers
ver_fastest = laps.pick_driver("VER").pick_fastest()
ham_fastest = laps.pick_driver("HAM").pick_fastest()

# Calculate delta
delta, ref_tel, comp_tel = tif1.utils.delta_time(ver_fastest, ham_fastest)

# Analyze where time was gained/lost
print(f"Maximum advantage: {delta.max():.3f}s")
print(f"Maximum disadvantage: {delta.min():.3f}s")

Note: This is a compatibility function. For more advanced delta analysis, consider using the telemetry comparison features in the plotting module.

Reliability & Networking

The tif1 library implements reliability patterns to handle network failures, CDN outages, and transient errors gracefully. The circuit breaker pattern prevents cascading failures, while the CDN manager provides automatic fallback to alternative data sources.

Circuit Breaker Pattern

The circuit breaker is a critical reliability component. It stops the application from repeatedly trying operations that are likely to fail. It acts like an electrical circuit breaker, “opening” after too many failures to prevent cascading issues. Circuit Breaker States:

Closed (Normal): All requests are allowed through
Open (Failing): Requests are blocked immediately without attempting
Half-Open (Testing): Limited requests allowed to test if service recovered

State Transitions:

Closed --[threshold failures]--> Open
Open --[timeout elapsed]--> Half-Open
Half-Open --[success]--> Closed
Half-Open --[failure]--> Open

`get_circuit_breaker`

Returns the global circuit breaker instance used for all network requests.

def get_circuit_breaker() -> CircuitBreaker

The circuit breaker is a singleton that tracks failures across all network operations in the application. Returns:

CircuitBreaker: Global circuit breaker instance

Thread Safety:

Fully thread-safe with atomic state transitions
Uses reentrant locks for nested calls
Safe for concurrent use across multiple threads

Basic Example:

from tif1.retry import get_circuit_breaker

cb = get_circuit_breaker()

# Inspect current state
print(f"State: {cb.state}")  # "closed", "open", or "half-open"
print(f"Failure count: {cb.failures}")
print(f"Threshold: {cb.threshold}")
print(f"Timeout: {cb.timeout}s")

# Check if last failure time is set
if cb.last_failure_time:
    print(f"Last failure: {cb.last_failure_time}")

CircuitBreaker Properties

`state: str`

Current circuit breaker state (read-only). Possible Values:

"closed": Normal operation, all requests allowed
"open": Too many failures, requests blocked
"half-open": Testing recovery, limited requests allowed

Example:

cb = get_circuit_breaker()

if cb.state == "open":
    print("Circuit breaker is open - service unavailable")
elif cb.state == "half-open":
    print("Circuit breaker testing recovery")
else:
    print("Circuit breaker closed - normal operation")

`failures: int`

Current failure count (read-only). Example:

cb = get_circuit_breaker()
print(f"Current failures: {cb.failures}/{cb.threshold}")

`threshold: int`

Number of failures before circuit opens. Default: 5 (configurable via circuit_breaker_threshold config)

`timeout: int`

Seconds to wait before transitioning from open to half-open. Default: 60 (configurable via circuit_breaker_timeout config)

`last_failure_time: datetime | None`

Timestamp of the most recent failure. Example:

from datetime import datetime
cb = get_circuit_breaker()

if cb.last_failure_time:
    elapsed = (datetime.now() - cb.last_failure_time).total_seconds()
    print(f"Last failure was {elapsed:.1f}s ago")

CircuitBreaker Methods

`record_success() -> None`

Record a successful operation (resets failure count). Behavior:

Resets failure counter to 0
Transitions from half-open to closed
Thread-safe atomic operation

Example:

cb = get_circuit_breaker()

try:
    # Your network operation
    response = fetch_data()
    cb.record_success()
except Exception:
    cb.record_failure()

`record_failure() -> None`

Record a failed operation (increments failure count). Behavior:

Atomically increments failure counter
Updates last failure timestamp
Opens circuit if threshold reached
Thread-safe operation

Example:

cb = get_circuit_breaker()

try:
    response = fetch_data()
    cb.record_success()
except NetworkError:
    cb.record_failure()
    if cb.state == "open":
        print("Circuit breaker opened!")

`check_and_update_state() -> tuple[bool, str]`

Check if request should proceed and update state if needed. Returns:

Tuple of (should_proceed, current_state)

Example:

cb = get_circuit_breaker()

should_proceed, state = cb.check_and_update_state()
if not should_proceed:
    print(f"Request blocked - circuit breaker is {state}")
else:
    print(f"Request allowed - circuit breaker is {state}")

`reset_circuit_breaker`

Reset the circuit breaker to initial closed state.

def reset_circuit_breaker() -> None

This function creates a new circuit breaker instance with configuration from the current config, effectively resetting all state. Behavior:

Creates new circuit breaker instance
Resets failure count to 0
Sets state to closed
Clears last failure time
Reloads configuration from config

Use Cases:

After resolving network issues
When switching networks
After configuration changes
For testing purposes

Example:

from tif1.retry import reset_circuit_breaker, get_circuit_breaker

# Check current state
cb = get_circuit_breaker()
print(f"Before reset: {cb.state}, failures: {cb.failures}")

# Reset circuit breaker
reset_circuit_breaker()

# Verify reset
cb = get_circuit_breaker()
print(f"After reset: {cb.state}, failures: {cb.failures}")
# Output: After reset: closed, failures: 0

CDN Management

The CDN manager handles multiple CDN sources with automatic fallback, health tracking, and failure recovery.

`get_cdn_manager`

Returns the CDN manager responsible for CDN fallback and source health.

def get_cdn_manager() -> CDNManager

The CDN manager maintains a list of CDN sources and automatically falls back to alternatives when the primary CDN fails. Returns:

CDNManager: Global CDN manager instance

Basic Example:

from tif1.cdn import get_cdn_manager

cdn = get_cdn_manager()

# Inspect CDN sources
sources = cdn.get_sources()
for source in sources:
    print(f"{source.name}: {source.base_url} (priority: {source.priority})")

# Check failure counts
print(f"Failure tracking: {cdn._failure_counts}")

CDNManager Methods

`get_sources() -> list[CDNSource]`

Get list of enabled CDN sources sorted by priority. Returns:

List of enabled CDN sources that have not exceeded failure threshold

Example:

cdn = get_cdn_manager()

sources = cdn.get_sources()
print(f"Available CDN sources: {len(sources)}")

for source in sources:
    failures = cdn._failure_counts.get(source.name, 0)
    status = "healthy" if failures == 0 else f"{failures} failures"
    print(f"  {source.name}: {status}")

`mark_failure(source_name: str) -> None`

Mark a CDN source as failed (increments failure count). Parameters:

source_name (str): Name of the CDN source

Behavior:

Increments failure counter for the source
Disables source if it reaches max failures (3)
Logs warning when source is disabled

Example:

cdn = get_cdn_manager()

try:
    # Attempt to fetch from CDN
    data = fetch_from_cdn("jsDelivr")
except Exception:
    cdn.mark_failure("jsDelivr")

`mark_success(source_name: str) -> None`

Mark a CDN source as successful (resets failure count). Parameters:

source_name (str): Name of the CDN source

Behavior:

Resets failure counter to 0
Re-enables source if it was disabled

Example:

cdn = get_cdn_manager()

try:
    data = fetch_from_cdn("jsDelivr")
    cdn.mark_success("jsDelivr")
except Exception:
    cdn.mark_failure("jsDelivr")

`reset() -> None`

Reset all CDN failure counts. Behavior:

Resets all failure counters to 0
Re-enables all CDN sources
Useful after network issues are resolved

Example:

cdn = get_cdn_manager()

# Reset all CDN health tracking
cdn.reset()
print("All CDN sources reset to healthy state")

Reliability Examples

Handle Network Failures Gracefully

import tif1
from tif1.retry import get_circuit_breaker, reset_circuit_breaker
from tif1.exceptions import NetworkError
import time

def load_session_with_retry():
    """Load session with circuit breaker awareness."""
    cb = get_circuit_breaker()

    # Check circuit breaker state before attempting
    if cb.state == "open":
        print(f"Circuit breaker is open. Waiting {cb.timeout}s...")
        time.sleep(cb.timeout)
        reset_circuit_breaker()

    try:
        session = tif1.get_session(2021, "Belgian Grand Prix", "Race")
        laps = session.laps
        print(f"Successfully loaded {len(laps)} laps")
        return session
    except NetworkError as e:
        print(f"Network error: {e.message}")
        print(f"Circuit breaker state: {cb.state}")
        print(f"Failures: {cb.failures}/{cb.threshold}")
        raise

# Use the function
session = load_session_with_retry()

Monitor Circuit Breaker Health

from tif1.retry import get_circuit_breaker
import time

def monitor_circuit_breaker(duration_seconds=60):
    """Monitor circuit breaker state over time."""
    cb = get_circuit_breaker()
    start_time = time.time()

    while time.time() - start_time < duration_seconds:
        print(f"State: {cb.state:12} | Failures: {cb.failures}/{cb.threshold}")
        time.sleep(5)

# Run monitoring
monitor_circuit_breaker(30)

CDN Fallback Testing

from tif1.cdn import get_cdn_manager
import tif1

def test_cdn_fallback():
    """Test CDN fallback behavior."""
    cdn = get_cdn_manager()

    print("Initial CDN sources:")
    for source in cdn.get_sources():
        print(f"  {source.name}: {source.base_url}")

    # Simulate failures
    print("\nSimulating failures...")
    for i in range(3):
        cdn.mark_failure("jsDelivr")
        print(f"  Failure {i+1}: {cdn._failure_counts}")

    print("\nRemaining sources:")
    remaining = cdn.get_sources()
    if remaining:
        for source in remaining:
            print(f"  {source.name}: {source.base_url}")
    else:
        print("  No sources available!")

    # Reset
    print("\nResetting CDN manager...")
    cdn.reset()
    print(f"After reset: {len(cdn.get_sources())} sources available")

test_cdn_fallback()

Production Error Handling

import tif1
from tif1.retry import get_circuit_breaker, reset_circuit_breaker
from tif1.cdn import get_cdn_manager
from tif1.exceptions import NetworkError, DataNotFoundError
import logging
import time

# Configure logging
tif1.setup_logging(logging.INFO)

def robust_session_loader(year, gp, session_type, max_attempts=3):
    """Production-grade session loader with comprehensive error handling."""
    cb = get_circuit_breaker()
    cdn = get_cdn_manager()

    for attempt in range(max_attempts):
        try:
            # Check circuit breaker
            if cb.state == "open":
                logging.warning(f"Circuit breaker open, waiting {cb.timeout}s...")
                time.sleep(cb.timeout)
                reset_circuit_breaker()

            # Attempt to load session
            logging.info(f"Loading session (attempt {attempt + 1}/{max_attempts})...")
            session = tif1.get_session(year, gp, session_type)
            laps = session.laps

            logging.info(f"Successfully loaded {len(laps)} laps")
            return session

        except DataNotFoundError as e:
            # Data genuinely doesn't exist - don't retry
            logging.error(f"Data not found: {e.message}")
            raise

        except NetworkError as e:
            logging.error(f"Network error: {e.message}")
            logging.info(f"Circuit breaker: {cb.state}, failures: {cb.failures}")
            logging.info(f"CDN sources: {len(cdn.get_sources())} available")

            if attempt < max_attempts - 1:
                wait_time = 2 ** attempt  # Exponential backoff
                logging.info(f"Retrying in {wait_time}s...")
                time.sleep(wait_time)
            else:
                logging.error("Max attempts reached, giving up")
                raise

    raise RuntimeError("Failed to load session after all attempts")

# Use in production
try:
    session = robust_session_loader(2021, "Belgian Grand Prix", "Race")
except Exception as e:
    logging.critical(f"Fatal error: {e}")
    # Alert monitoring system, etc.

Reliability Best Practices

Monitor circuit breaker state: Check state before critical operations
Respect circuit breaker: Do not bypass it when open
Reset after network changes: Call reset_circuit_breaker() when switching networks
Configure thresholds appropriately: Lower for fail-fast, higher for resilience
Use multiple CDNs: Configure fallback CDNs for production
Log failures: Track patterns to identify systemic issues
Implement exponential backoff: Do not retry immediately after failures
Handle DataNotFoundError separately: Do not retry when data genuinely does not exist

Memory Management

The tif1 library uses in-memory LRU (Least Recently Used) caches to speed up repeated access to lap DataFrames. While this improves performance, it can consume significant memory when working with many sessions. These utilities help manage memory usage.

`clear_lap_cache`

Clear the in-memory LRU cache for lap DataFrames.

def clear_lap_cache() -> None

This function clears both the pandas and polars lap DataFrame caches, freeing memory without affecting the persistent SQLite cache. Behavior:

Clears pandas lap cache
Clears polars lap cache
Does NOT affect SQLite cache (persistent storage)
Immediate effect on memory usage
Thread-safe operation

When to Use:

After processing large sessions
When switching between different analyses
To free memory in long-running applications
Before loading a different season’s data

Example:

from tif1.core import clear_lap_cache
import tif1

# Load and process a session
session = tif1.get_session(2021, "Belgian Grand Prix", "Race")
laps = session.laps

# Do analysis...
fastest_laps = laps.pick_fastest()

# Clear memory cache when done
clear_lap_cache()
print("Lap cache cleared - memory freed")

Memory Impact Example:

import tif1
from tif1.core import clear_lap_cache
import psutil
import os

def get_memory_usage():
    """Get current process memory usage in MB."""
    process = psutil.Process(os.getpid())
    return process.memory_info().rss / 1024 / 1024

# Measure memory before
mem_before = get_memory_usage()
print(f"Memory before: {mem_before:.1f} MB")

# Load multiple sessions
for gp in ["Bahrain", "Saudi_Arabia", "Australia", "Emilia_Romagna"]:
    session = tif1.get_session(2021, gp, "Race")
    laps = session.laps
    print(f"Loaded {gp}: {len(laps)} laps")

mem_after = get_memory_usage()
print(f"Memory after loading: {mem_after:.1f} MB")
print(f"Memory increase: {mem_after - mem_before:.1f} MB")

# Clear lap cache
clear_lap_cache()

mem_cleared = get_memory_usage()
print(f"Memory after clearing: {mem_cleared:.1f} MB")
print(f"Memory freed: {mem_after - mem_cleared:.1f} MB")

Typical Memory Usage:

Single session lap DataFrame: ~5-15 MB
Full season in cache: ~200-500 MB
Clearing cache can free 50-80% of lap data memory

Best Practices:

Clear cache between unrelated analyses
Clear cache in batch processing loops
Do not clear too frequently (defeats caching purpose)
Monitor memory usage in long-running applications
Consider clearing cache before loading different seasons

Advanced: Selective Cache Management:

import tif1
from tif1.core import clear_lap_cache

def process_season_with_memory_management(year):
    """Process entire season with periodic cache clearing."""
    events = tif1.get_events(year)

    for i, event in enumerate(events):
        # Load and process session
        session = tif1.get_session(year, event.EventName, "Race")
        laps = session.laps

        # Your analysis here
        fastest = laps.pick_fastest()
        print(f"{event.EventName}: Fastest lap {fastest['LapTime'].iloc[0]}")

        # Clear cache every 5 events to manage memory
        if (i + 1) % 5 == 0:
            clear_lap_cache()
            print(f"  Cleared cache after {i + 1} events")

process_season_with_memory_management(2021)

Complete Examples

This section provides examples demonstrating how to combine multiple utilities for real-world use cases.

Example 1: Configure and Save Settings

Set up the configuration for the specific use case and persist it.

import tif1
from pathlib import Path

def setup_production_config():
    """Configure tif1 for production use with optimal settings."""
    config = tif1.get_config()

    # Performance settings
    config.set("lib", "polars")  # Use polars for better performance
    config.set("validate_data", False)  # Skip validation in production
    config.set("ultra_cold_start", False)  # Load full sessions

    # Network settings
    config.set("timeout", 60)  # Longer timeout for reliability
    config.set("max_retries", 5)  # More retries for resilience

    # Circuit breaker settings
    config.set("circuit_breaker_threshold", 10)  # Tolerate more failures
    config.set("circuit_breaker_timeout", 120)  # Longer recovery time

    # Cache settings
    cache_dir = Path.home() / ".cache" / "tif1_production"
    config.set("cache_dir", str(cache_dir))
    config.set("enable_cache", True)

    # CDN settings
    config.set("cdn_use_minification", True)  # Reduce bandwidth

    # Save configuration
    config.save()
    print(f"Production configuration saved to ~/.tif1rc")

    # Print summary
    print("\nConfiguration Summary:")
    print(f"  DataFrame library: {config.get('lib')}")
    print(f"  Cache directory: {config.get('cache_dir')}")
    print(f"  Timeout: {config.get('timeout')}s")
    print(f"  Max retries: {config.get('max_retries')}")
    print(f"  Circuit breaker threshold: {config.get('circuit_breaker_threshold')}")
    print(f"  Validation: {config.get('validate_data')}")

# Run setup
setup_production_config()

Example 2: Debug Session Loading

Enable detailed logging to diagnose data loading issues.

import tif1
import logging
from datetime import datetime

def debug_session_load(year, gp, session_type):
    """Load session with comprehensive debug logging."""
    # Enable debug logging
    tif1.setup_logging(logging.DEBUG)

    print(f"{'='*60}")
    print(f"Debug Session Load: {year} {gp} {session_type}")
    print(f"Started at: {datetime.now()}")
    print(f"{'='*60}\n")

    try:
        # Load session with debug output
        start_time = datetime.now()
        session = tif1.get_session(year, gp, session_type)
        load_time = (datetime.now() - start_time).total_seconds()

        print(f"\n{'='*60}")
        print(f"Session loaded successfully in {load_time:.2f}s")
        print(f"{'='*60}\n")

        # Load laps
        start_time = datetime.now()
        laps = session.laps
        laps_time = (datetime.now() - start_time).total_seconds()

        print(f"\n{'='*60}")
        print(f"Laps loaded in {laps_time:.2f}s")
        print(f"Total laps: {len(laps)}")
        print(f"Drivers: {laps['Driver'].unique().tolist()}")
        print(f"{'='*60}\n")

        return session

    except Exception as e:
        print(f"\n{'='*60}")
        print(f"ERROR: {type(e).__name__}: {e}")
        print(f"{'='*60}\n")
        raise
    finally:
        # Restore normal logging
        tif1.setup_logging(logging.WARNING)

# Use the debug function
session = debug_session_load(2021, "Belgian Grand Prix", "Race")

Example 3: Cache Management Workflow

Complete cache inspection and management.

import tif1
from pathlib import Path
import json

def cache_management_dashboard():
    """Interactive cache management dashboard."""
    cache = tif1.get_cache()
    config = tif1.get_config()

    print("="*60)
    print("TIF1 Cache Management Dashboard")
    print("="*60)

    # Cache location
    cache_dir = Path(cache.cache_dir)
    cache_db = cache_dir / "cache.db"

    print(f"\nCache Location:")
    print(f"  Directory: {cache_dir}")
    print(f"  Database: {cache_db}")

    # Cache size
    if cache_db.exists():
        size_bytes = cache_db.stat().st_size
        size_mb = size_bytes / (1024 * 1024)
        print(f"  Size: {size_mb:.2f} MB ({size_bytes:,} bytes)")
    else:
        print(f"  Status: Not initialized")
        return

    # Check specific sessions
    print(f"\nCached Sessions:")
    test_sessions = [
        (2025, "Monaco_Grand_Prix", "Race"),
        (2025, "Monaco_Grand_Prix", "Qualifying"),
        (2024, "Abu_Dhabi_Grand_Prix", "Race"),
        (2021, "Belgian_Grand_Prix", "Race"),
    ]

    cached_count = 0
    for year, gp, session in test_sessions:
        is_cached = cache.has_session_data(year, gp, session)
        status = "✓ Cached" if is_cached else "✗ Not cached"
        print(f"  {year} {gp} {session}: {status}")
        if is_cached:
            cached_count += 1

    print(f"\nSummary: {cached_count}/{len(test_sessions)} sessions cached")

    # Configuration
    print(f"\nCache Configuration:")
    print(f"  Enabled: {config.get('enable_cache')}")
    print(f"  Library: {config.get('lib')}")

    # Management options
    print(f"\n{'='*60}")
    print("Management Options:")
    print("  1. Keep cache (recommended)")
    print("  2. Clear cache (free disk space)")
    print("  3. Export cache info")
    print(f"{'='*60}")

    choice = input("\nSelect option (1-3): ").strip()

    if choice == "2":
        confirm = input("Are you sure you want to clear the cache? (yes/no): ")
        if confirm.lower() == "yes":
            cache.clear()
            print("✓ Cache cleared successfully")
        else:
            print("✗ Cache clear cancelled")
    elif choice == "3":
        info = {
            "cache_dir": str(cache_dir),
            "cache_size_mb": size_mb if cache_db.exists() else 0,
            "cached_sessions": cached_count,
            "config": {
                "enable_cache": config.get("enable_cache"),
                "lib": config.get("lib"),
            }
        }
        output_file = "tif1_cache_info.json"
        with open(output_file, "w") as f:
            json.dump(info, f, indent=2)
        print(f"✓ Cache info exported to {output_file}")
    else:
        print("✓ Cache preserved")

# Run dashboard
cache_management_dashboard()

Example 4: Handle Network Issues

Reliable error handling with circuit breaker awareness.

import tif1
from tif1.retry import get_circuit_breaker, reset_circuit_breaker
from tif1.cdn import get_cdn_manager
from tif1.exceptions import NetworkError, DataNotFoundError
import time
import logging

# Configure logging
tif1.setup_logging(logging.INFO)

def load_with_resilience(year, gp, session_type, max_retries=3):
    """Load session with comprehensive error handling and retry logic."""
    cb = get_circuit_breaker()
    cdn = get_cdn_manager()

    print(f"Loading {year} {gp} {session_type}...")
    print(f"Circuit breaker: {cb.state}, failures: {cb.failures}/{cb.threshold}")
    print(f"Available CDNs: {len(cdn.get_sources())}")

    for attempt in range(max_retries):
        try:
            # Check circuit breaker state
            if cb.state == "open":
                wait_time = cb.timeout
                print(f"\n⚠ Circuit breaker is OPEN")
                print(f"  Waiting {wait_time}s before retry...")
                time.sleep(wait_time)
                reset_circuit_breaker()
                print(f"  Circuit breaker reset")

            # Attempt to load
            print(f"\nAttempt {attempt + 1}/{max_retries}...")
            start_time = time.time()

            session = tif1.get_session(year, gp, session_type)
            laps = session.laps

            load_time = time.time() - start_time
            print(f"✓ Success! Loaded {len(laps)} laps in {load_time:.2f}s")

            return session

        except DataNotFoundError as e:
            # Data doesn't exist - don't retry
            print(f"\n✗ Data not found: {e.message}")
            print(f"  This session may not have data available")
            raise

        except NetworkError as e:
            print(f"\n✗ Network error: {e.message}")
            print(f"  Status code: {e.status_code}")
            print(f"  Circuit breaker: {cb.state}")
            print(f"  Failures: {cb.failures}/{cb.threshold}")

            # Check CDN health
            available_cdns = cdn.get_sources()
            print(f"  Available CDNs: {len(available_cdns)}")

            if attempt < max_retries - 1:
                # Exponential backoff
                wait_time = 2 ** attempt
                print(f"  Retrying in {wait_time}s...")
                time.sleep(wait_time)
            else:
                print(f"\n✗ Failed after {max_retries} attempts")
                print(f"  Circuit breaker state: {cb.state}")
                print(f"  Consider:")
                print(f"    - Checking network connection")
                print(f"    - Increasing timeout (current: {tif1.get_config().get('timeout')}s)")
                print(f"    - Trying again later")
                raise

        except Exception as e:
            print(f"\n✗ Unexpected error: {type(e).__name__}: {e}")
            raise

    raise RuntimeError("Failed to load session")

# Example usage
try:
    session = load_with_resilience(2021, "Belgian Grand Prix", "Race")
    print(f"\nSession loaded successfully!")
    print(f"Drivers: {session.laps['Driver'].unique().tolist()}")
except Exception as e:
    print(f"\nFatal error: {e}")

Example 5: Custom Configuration File

Create project-specific configuration.

import tif1
from pathlib import Path
import json

def create_project_config(project_name, project_dir):
    """Create and save project-specific tif1 configuration."""
    config = tif1.get_config()
    project_path = Path(project_dir)

    # Create project-specific cache directory
    cache_dir = project_path / ".cache" / "tif1"
    cache_dir.mkdir(parents=True, exist_ok=True)

    # Configure for project
    config.set("cache_dir", str(cache_dir))
    config.set("lib", "polars")  # Use polars for performance
    config.set("timeout", 45)
    config.set("max_retries", 4)
    config.set("validate_data", True)  # Validate in development
    config.set("ultra_cold_start", False)

    # Save to project directory
    config_file = project_path / "tif1_config.json"
    config.save(config_file)

    print(f"Project Configuration Created: {project_name}")
    print(f"{'='*60}")
    print(f"Config file: {config_file}")
    print(f"Cache directory: {cache_dir}")
    print(f"\nSettings:")
    print(f"  Library: {config.get('lib')}")
    print(f"  Timeout: {config.get('timeout')}s")
    print(f"  Max retries: {config.get('max_retries')}")
    print(f"  Validation: {config.get('validate_data')}")
    print(f"\nTo use this config:")
    print(f"  1. Set TIF1_CACHE_DIR={cache_dir}")
    print(f"  2. Or load config programmatically")

    # Create README
    readme = project_path / "TIF1_README.md"
    with open(readme, "w") as f:
        f.write(f"# {project_name} - TIF1 Configuration\n\n")
        f.write(f"## Configuration\n\n")
        f.write(f"This project uses a custom tif1 configuration:\n\n")
        f.write(f"- Config file: `{config_file.name}`\n")
        f.write(f"- Cache directory: `.cache/tif1/`\n")
        f.write(f"- DataFrame library: polars\n\n")
        f.write(f"## Usage\n\n")
        f.write(f"```python\n")
        f.write(f"import tif1\n\n")
        f.write(f"# Load project config\n")
        f.write(f"config = tif1.get_config()\n")
        f.write(f"# Config is already loaded from {config_file.name}\n\n")
        f.write(f"# Use tif1 normally\n")
        f.write(f"session = tif1.get_session(2021, 'Monaco Grand Prix', 'Race')\n")
        f.write(f"```\n")

    print(f"\n✓ README created: {readme}")

    return config_file

# Create project config
project_config = create_project_config(
    project_name="F1 Analysis Project",
    project_dir="./my_f1_project"
)

Example 6: Performance Optimization

Optimize configuration for maximum performance.

import tif1
from tif1.core import clear_lap_cache
import time

def optimize_for_performance():
    """Configure tif1 for maximum performance."""
    config = tif1.get_config()

    print("Optimizing tif1 for performance...")
    print("="*60)

    # Performance settings
    optimizations = {
        "lib": "polars",  # Faster DataFrame operations
        "validate_data": False,  # Skip validation overhead
        "enable_cache": True,  # Use caching
        "cdn_use_minification": True,  # Smaller downloads
        "timeout": 30,  # Don't wait too long
        "max_retries": 3,  # Reasonable retry count
    }

    for key, value in optimizations.items():
        old_value = config.get(key)
        config.set(key, value)
        print(f"  {key}: {old_value} → {value}")

    config.save()
    print("\n✓ Configuration optimized and saved")

    # Benchmark
    print("\nRunning benchmark...")
    print("-"*60)

    # Clear cache for fair test
    cache = tif1.get_cache()
    cache.clear()
    clear_lap_cache()

    # Test load time
    start = time.time()
    session = tif1.get_session(2021, "Monaco Grand Prix", "Race")
    laps = session.laps
    first_load = time.time() - start

    print(f"First load (cold): {first_load:.2f}s")

    # Test cached load
    clear_lap_cache()  # Clear memory cache only
    start = time.time()
    session = tif1.get_session(2021, "Monaco Grand Prix", "Race")
    laps = session.laps
    cached_load = time.time() - start

    print(f"Second load (cached): {cached_load:.2f}s")
    print(f"Speedup: {first_load/cached_load:.1f}x faster")

    # Test telemetry access
    start = time.time()
    fastest = laps.pick_fastest()
    tel = fastest.get_telemetry()
    telemetry_time = time.time() - start

    print(f"Telemetry load: {telemetry_time:.3f}s")
    print(f"\n✓ Benchmark complete")

optimize_for_performance()

Example 7: Memory-Efficient Batch Processing

Process multiple sessions while managing memory usage.

import tif1
from tif1.core import clear_lap_cache
import psutil
import os

def get_memory_mb():
    """Get current process memory usage in MB."""
    process = psutil.Process(os.getpid())
    return process.memory_info().rss / 1024 / 1024

def batch_process_season(year, cache_clear_interval=5):
    """Process entire season with memory management."""
    print(f"Batch Processing Season {year}")
    print("="*60)

    # Get all events
    events = tif1.get_events(year)
    print(f"Found {len(events)} events\n")

    results = []
    initial_memory = get_memory_mb()

    for i, event in enumerate(events, 1):
        event_name = event.EventName
        print(f"[{i}/{len(events)}] Processing {event_name}...")

        try:
            # Load session
            session = tif1.get_session(year, event_name, "Race")
            laps = session.laps

            # Analyze
            fastest = laps.pick_fastest()
            fastest_time = fastest['LapTime'].iloc[0]
            fastest_driver = fastest['Driver'].iloc[0]

            result = {
                "event": event_name,
                "fastest_driver": fastest_driver,
                "fastest_time": str(fastest_time),
                "total_laps": len(laps)
            }
            results.append(result)

            current_memory = get_memory_mb()
            memory_increase = current_memory - initial_memory

            print(f"  Fastest: {fastest_driver} - {fastest_time}")
            print(f"  Memory: {current_memory:.1f} MB (+{memory_increase:.1f} MB)")

            # Clear cache periodically
            if i % cache_clear_interval == 0:
                clear_lap_cache()
                after_clear = get_memory_mb()
                freed = current_memory - after_clear
                print(f"  Cleared cache: freed {freed:.1f} MB")

        except Exception as e:
            print(f"  ✗ Error: {e}")
            results.append({
                "event": event_name,
                "error": str(e)
            })

        print()

    # Summary
    print("="*60)
    print("Batch Processing Complete")
    print(f"Processed: {len(results)} events")
    print(f"Final memory: {get_memory_mb():.1f} MB")
    print(f"Peak increase: {get_memory_mb() - initial_memory:.1f} MB")

    return results

# Run batch processing
results = batch_process_season(2021, cache_clear_interval=3)

# Print summary
print("\nFastest Laps Summary:")
for result in results:
    if "error" not in result:
        print(f"  {result['event']}: {result['fastest_driver']} - {result['fastest_time']}")

Best Practices

Guidelines for using tif1 utilities effectively in different scenarios.

Development Best Practices

Enable Debug Logging During Development
import tif1 import logging # Enable detailed logging tif1.setup_logging(logging.DEBUG)
- Helps understand data flow and identify issues
- Shows cache hits/misses for optimization
- Reveals network request patterns
- Displays parsing and validation steps
Use Pandas for Easier Debugging
config = tif1.get_config() config.set("lib", "pandas")
- More familiar API for most developers
- Better error messages
- Easier to inspect intermediate results
- Switch to polars for production
Keep Validation Enabled
config.set("validate_data", True)
- Catches data quality issues early
- Provides clear error messages
- Helps identify API changes
- ~10-15% overhead is acceptable in development
Use Shorter Timeouts
config.set("timeout", 10)
- Fail fast during development
- Do not wait for slow connections
- Iterate quickly on code changes
Clear Cache When Testing
cache = tif1.get_cache() cache.clear()
- Ensures that tests run with fresh data
- Validates cache warming logic
- Catches cache-related bugs

Production Best Practices

Use Polars for Maximum Performance
config = tif1.get_config() config.set("lib", "polars") config.save()
- 2-5x faster DataFrame operations
- Better memory efficiency
- Lower CPU usage
- Handles larger datasets
Disable Validation in Production
config.set("validate_data", False)
- Saves ~10-15% processing time
- Reduces CPU usage
- Only disable if data quality is trusted
- Keep enabled if data source is unreliable
Increase Timeout for Reliability
config.set("timeout", 60) config.set("max_retries", 5)
- Handle slow networks gracefully
- Accommodate mobile/satellite connections
- Reduce failure rate
- Balance with user experience
Configure Circuit Breaker Appropriately
config.set("circuit_breaker_threshold", 10) config.set("circuit_breaker_timeout", 120)
- Higher threshold for production resilience
- Longer timeout for recovery
- Prevents cascading failures
- Monitor circuit breaker state

Use Multiple CDNs

config.set("cdns", [
    "https://cdn.jsdelivr.net/gh/TracingInsights",
    "https://backup-cdn.example.com/TracingInsights"
])

Automatic fallback on CDN failure
Improved reliability
Geographic redundancy
Load distribution

Enable CDN Minification
config.set("cdn_use_minification", True)
- Reduces bandwidth by 20-40%
- Faster downloads
- Lower data costs
- Minimal CPU overhead

Set Up Proper Logging

import logging

# Use INFO level for production monitoring
tif1.setup_logging(logging.INFO)

# Configure file handler
handler = logging.FileHandler("/var/log/tif1.log")
handler.setFormatter(logging.Formatter(
    "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
))
logging.getLogger("tif1").addHandler(handler)

Monitor operational health
Track performance metrics
Debug production issues
Audit data access

Close Cache on Exit
import atexit import tif1 cache = tif1.get_cache() atexit.register(cache.close)
- Ensures data is flushed to disk
- Prevents corruption
- Releases file locks
- Clean shutdown

Performance Optimization

Use Ultra Cold Start for Single Lap Queries
config.set("ultra_cold_start", True)
- Faster for single lap telemetry
- Skips loading full session data
- On-demand fetching
- Not suitable for full session analysis
Clear Lap Cache Periodically
from tif1.core import clear_lap_cache # Clear after processing each session clear_lap_cache()
- Prevents memory bloat
- Important for long-running applications
- Clear between unrelated analyses
- Do not clear too frequently

Use Batch Telemetry Operations

cache = tif1.get_cache()

# Batch operation (fast)
telemetry_batch = cache.get_telemetry_batch(
    2021, "Monaco_Grand_Prix", "Race", "VER", [1, 2, 3, 4, 5]
)

# Individual operations (slow)
# for lap in [1, 2, 3, 4, 5]:
#     tel = cache.get_telemetry(2021, "Monaco_Grand_Prix", "Race", "VER", lap)

Single database query vs multiple
Much faster for multiple laps
Reduces I/O overhead

Monitor Cache Size

from pathlib import Path

cache = tif1.get_cache()
cache_db = Path(cache.cache_dir) / "cache.db"
size_mb = cache_db.stat().st_size / (1024 * 1024)

if size_mb > 5000:  # 5 GB
    print("Cache is large, consider clearing old data")

Prevent disk space issues
Set up monitoring alerts
Implement automatic cleanup
Archive old seasons

Use SSD for Cache
# Put cache on fast local SSD config.set("cache_dir", "/mnt/ssd/tif1_cache")
- Faster than HDD
- Reduces SQLite query time
- Better for concurrent access
- Avoid network drives

Error Handling

Distinguish Between Error Types

from tif1.exceptions import DataNotFoundError, NetworkError

try:
    session = tif1.get_session(2021, "Test GP", "Race")
except DataNotFoundError:
    # Data doesn't exist - don't retry
    print("Session data not available")
except NetworkError:
    # Network issue - can retry
    print("Network error, retrying...")

Do not retry DataNotFoundError
Retry NetworkError with backoff
Handle each error type appropriately

Monitor Circuit Breaker State

from tif1.retry import get_circuit_breaker

cb = get_circuit_breaker()
if cb.state == "open":
    # Alert monitoring system
    send_alert("Circuit breaker is open")

Track circuit breaker state
Alert on open state
Monitor failure patterns
Investigate root causes

Implement Exponential Backoff

import time

for attempt in range(max_retries):
    try:
        session = tif1.get_session(year, gp, session_type)
        break
    except NetworkError:
        if attempt < max_retries - 1:
            wait_time = 2 ** attempt
            time.sleep(wait_time)

Do not retry immediately
Give service time to recover
Reduce load on failing service
Increase wait time exponentially

Log Errors with Context

import logging

try:
    session = tif1.get_session(year, gp, session_type)
except Exception as e:
    logging.error(
        f"Failed to load session",
        extra={
            "year": year,
            "gp": gp,
            "session_type": session_type,
            "error": str(e),
            "circuit_breaker_state": get_circuit_breaker().state
        }
    )

Include relevant context
Track failure patterns
Enable debugging
Support monitoring

Memory Management

Clear Cache in Batch Processing

from tif1.core import clear_lap_cache

for i, event in enumerate(events):
    session = tif1.get_session(year, event, "Race")
    # Process session...

    if (i + 1) % 5 == 0:
        clear_lap_cache()

Prevent memory bloat
Clear every N iterations
Balance performance vs memory
Monitor memory usage

Use Context Managers for Cache

class CacheManager:
    def __enter__(self):
        self.cache = tif1.get_cache()
        return self.cache

    def __exit__(self, *args):
        self.cache.close()

with CacheManager() as cache:
    # Use cache...
    pass
# Cache automatically closed

Ensures proper cleanup
Prevents resource leaks
Clean shutdown

Monitor Memory Usage

import psutil
import os

def check_memory():
    process = psutil.Process(os.getpid())
    mem_mb = process.memory_info().rss / 1024 / 1024
    if mem_mb > 2000:  # 2 GB
        clear_lap_cache()

Set memory thresholds
Automatic cleanup
Prevent OOM errors
Log memory metrics

Configuration Management

Use Environment Variables for Deployment
export TIF1_LIB=polars export TIF1_CACHE_DIR=/var/cache/tif1 export TIF1_TIMEOUT=60 export TIF1_VALIDATE_DATA=false
- Easy to configure across environments
- No code changes needed
- Supports containerization
- Environment-specific settings
Save Configuration After Tuning
config = tif1.get_config() # ... tune settings ... config.save()
- Persist optimized settings
- Share configuration across team
- Version control config files
- Document configuration choices

Use Project-Specific Configuration

# Load project config
config = tif1.get_config()
config.set("cache_dir", "./project_cache")
config.save("./tif1_project_config.json")

Isolate project caches
Different settings per project
Easier to manage
Portable configuration

Testing

Clear Cache Before Tests

import pytest
import tif1

@pytest.fixture(autouse=True)
def clear_cache():
    cache = tif1.get_cache()
    cache.clear()
    yield
    cache.clear()

Ensures test isolation
Prevents cache pollution
Reproducible tests
Catches cache bugs

Mock Network Calls

from unittest.mock import patch

@patch('tif1.http_session.fetch')
def test_session_load(mock_fetch):
    mock_fetch.return_value = {"laps": []}
    session = tif1.get_session(2021, "Test", "Race")

Faster tests
No network dependency
Predictable results
Test error handling

Test Circuit Breaker Behavior

from tif1.retry import get_circuit_breaker, reset_circuit_breaker

def test_circuit_breaker():
    reset_circuit_breaker()
    cb = get_circuit_breaker()

    # Simulate failures
    for _ in range(cb.threshold):
        cb.record_failure()

    assert cb.state == "open"

Verify reliability patterns
Test failure scenarios
Ensure proper recovery
Validate configuration

Monitoring and Observability

Track Performance Metrics

import time

start = time.time()
session = tif1.get_session(year, gp, session_type)
load_time = time.time() - start

# Log metric
logging.info(f"Session load time: {load_time:.2f}s")

Monitor load times
Track cache hit rates
Identify performance regressions
Optimize based on data

Set Up Health Checks

def health_check():
    """Check tif1 system health."""
    cb = get_circuit_breaker()
    cdn = get_cdn_manager()
    cache = tif1.get_cache()

    return {
        "circuit_breaker": cb.state,
        "failures": cb.failures,
        "available_cdns": len(cdn.get_sources()),
        "cache_enabled": tif1.get_config().get("enable_cache")
    }

Regular health checks
Alert on issues
Track system state
Support debugging

Implement Structured Logging

import logging
import json

class StructuredFormatter(logging.Formatter):
    def format(self, record):
        log_data = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "message": record.getMessage(),
            "module": record.module
        }
        return json.dumps(log_data)

handler = logging.StreamHandler()
handler.setFormatter(StructuredFormatter())
logging.getLogger("tif1").addHandler(handler)

Machine-readable logs
Easy to parse and analyze
Better for log aggregation
Supports monitoring tools

Summary

Follow these best practices to:

Build reliable applications with proper error handling
Optimize performance for the specific use case
Manage resources efficiently
Debug issues quickly
Deploy confidently to production
Monitor system health effectively

Remember: The right configuration depends on the specific use case. Development, production, and testing environments should have different configurations optimized for their respective goals.

Last modified on September 3, 2026

​Overview

​Logging

​setup_logging

​Configuration

​Configuration Loading Order

​get_config

​Config Methods

​get(key: str, default: Any = None) -> Any

​set(key: str, value: Any) -> None

​save(path: Path | str | None = None) -> None

​Available Configuration Keys

​Environment Variable Configuration

​Configuration Examples

​Development Configuration

​Production Configuration

​Minimal Latency Configuration

​Custom Cache Location

​Configuration Best Practices

​Cache Management

​Cache Architecture

​get_cache

​Cache Methods

​get(key: str) -> Any | None

​set(key: str, data: Any) -> None

​get_telemetry(year: int, gp: str, session: str, driver: str, lap: int) -> Any | None

​set_telemetry(year: int, gp: str, session: str, driver: str, lap: int, data: Any) -> None

​get_telemetry_batch(year: int, gp: str, session: str, driver: str, laps: list[int]) -> dict[int, Any]

​has_session_data(year: int, gp: str, session: str) -> bool

​clear() -> None

​close() -> None

​Cache Performance Characteristics

​Cache Size Management

​Async Cache Operations

​Cache Best Practices

​Data Utilities

​to_timedelta

​to_datetime

​recursive_dict_get

​delta_time

​Reliability & Networking

​Circuit Breaker Pattern

​get_circuit_breaker

​CircuitBreaker Properties

​state: str

​failures: int

​threshold: int

​timeout: int

​last_failure_time: datetime | None

​CircuitBreaker Methods

​record_success() -> None

​record_failure() -> None

​check_and_update_state() -> tuple[bool, str]

​reset_circuit_breaker

​CDN Management

​get_cdn_manager

​CDNManager Methods

​get_sources() -> list[CDNSource]

​mark_failure(source_name: str) -> None

​mark_success(source_name: str) -> None

​reset() -> None

​Reliability Examples

​Handle Network Failures Gracefully

​Monitor Circuit Breaker Health

​CDN Fallback Testing

​Production Error Handling

​Reliability Best Practices

​Memory Management

​clear_lap_cache

​Complete Examples

​Example 1: Configure and Save Settings

​Example 2: Debug Session Loading

​Example 3: Cache Management Workflow

​Example 4: Handle Network Issues

​Example 5: Custom Configuration File

​Example 6: Performance Optimization

​Example 7: Memory-Efficient Batch Processing

​Best Practices

​Development Best Practices

​Production Best Practices

​Performance Optimization

Overview

Logging

`setup_logging`

Configuration

Configuration Loading Order

`get_config`

Config Methods

`get(key: str, default: Any = None) -> Any`

`set(key: str, value: Any) -> None`

`save(path: Path | str | None = None) -> None`

Available Configuration Keys

Environment Variable Configuration

Configuration Examples

Development Configuration

Production Configuration

Minimal Latency Configuration

Custom Cache Location

Configuration Best Practices

Cache Management

Cache Architecture

`get_cache`

Cache Methods

`get(key: str) -> Any | None`

`set(key: str, data: Any) -> None`

`get_telemetry(year: int, gp: str, session: str, driver: str, lap: int) -> Any | None`

`set_telemetry(year: int, gp: str, session: str, driver: str, lap: int, data: Any) -> None`

`get_telemetry_batch(year: int, gp: str, session: str, driver: str, laps: list[int]) -> dict[int, Any]`

`has_session_data(year: int, gp: str, session: str) -> bool`

`clear() -> None`

`close() -> None`

Cache Performance Characteristics

Cache Size Management

Async Cache Operations

Cache Best Practices

Data Utilities

`to_timedelta`

`to_datetime`

`recursive_dict_get`

`delta_time`

Reliability & Networking

Circuit Breaker Pattern

`get_circuit_breaker`

CircuitBreaker Properties

`state: str`

`failures: int`

`threshold: int`

`timeout: int`

`last_failure_time: datetime | None`

CircuitBreaker Methods

`record_success() -> None`

`record_failure() -> None`

`check_and_update_state() -> tuple[bool, str]`

`reset_circuit_breaker`

CDN Management

`get_cdn_manager`

CDNManager Methods

`get_sources() -> list[CDNSource]`

`mark_failure(source_name: str) -> None`

`mark_success(source_name: str) -> None`

`reset() -> None`

Reliability Examples

Handle Network Failures Gracefully

Monitor Circuit Breaker Health

CDN Fallback Testing

Production Error Handling

Reliability Best Practices

Memory Management

`clear_lap_cache`

Complete Examples

Example 1: Configure and Save Settings

Example 2: Debug Session Loading

Example 3: Cache Management Workflow

Example 4: Handle Network Issues

Example 5: Custom Configuration File

Example 6: Performance Optimization

Example 7: Memory-Efficient Batch Processing

Best Practices

Development Best Practices

Production Best Practices

Performance Optimization