⏰ Background Tasks

🚀

Powering Real-Time Aviation Intelligence

SkysPy uses Celery for distributed background task processing, enabling real-time aircraft tracking, analytics computation, external data synchronization, and notification delivery without blocking the main application.

🏗️ Architecture Overview

flowchart TB
    subgraph Producers["📤 Task Producers"]
        API[🌐 Django API]
        WS[🔌 Socket.IO Handlers]
        Beat[⏱️ Celery Beat Scheduler]
    end

    subgraph Broker["🔴 Redis Broker"]
        Redis[(Redis 6.x)]
    end

    subgraph Queues["📬 Task Queues"]
        Q1[🚨 polling<br/>High Priority]
        Q2[📦 default<br/>General Purpose]
        Q3[🗄️ database<br/>DB Operations]
        Q4[🔔 notifications<br/>Alert Delivery]
        Q5[🎙️ transcription<br/>Audio Processing]
        Q6[🐢 low_priority<br/>Analytics & Cleanup]
    end

    subgraph Workers["👷 Celery Workers"]
        W1[Worker 1<br/>gevent pool]
        W2[Worker 2<br/>prefork pool]
        W3[Worker N<br/>...]
    end

    Producers --> Redis
    Redis --> Queues
    Queues --> Workers

    style Q1 fill:#ff6b6b,color:#fff
    style Q2 fill:#4ecdc4,color:#fff
    style Q3 fill:#45b7d1,color:#fff
    style Q4 fill:#f7dc6f,color:#000
    style Q5 fill:#bb8fce,color:#fff
    style Q6 fill:#85c1e9,color:#fff

🎯 Quick Reference

📊

At a Glance

Metric	Value
Total Tasks	45+
Queue Count	6
Fastest Interval	2 seconds
Slowest Interval	Weekly

📬 Queue System

flowchart LR
    subgraph Priority["Task Priority Levels"]
        direction TB
        P1["🔴 CRITICAL<br/>polling queue<br/>2-60s intervals"]
        P2["🟡 NORMAL<br/>default queue<br/>On-demand tasks"]
        P3["🟢 LOW<br/>low_priority queue<br/>Analytics & cleanup"]
    end

    P1 --> P2 --> P3

🚨 Queue Configuration

⚡

Performance Tip

For high-load deployments, run dedicated workers per queue to isolate time-critical tasks from long-running operations.

✈️ Aircraft Tasks

📍

Location: skyspy/tasks/aircraft.py

sequenceDiagram
    participant Beat as ⏱️ Celery Beat
    participant Task as 🔄 poll_aircraft
    participant ADS-B as 📡 ADS-B Receiver
    participant Cache as 💾 Redis Cache
    participant SIO as 🔌 Socket.IO

    Beat->>Task: Every 2 seconds
    Task->>ADS-B: Fetch positions
    ADS-B-->>Task: Aircraft data
    Task->>Cache: Update current_aircraft
    Task->>SIO: Broadcast updates
    Note over SIO: aircraft_update<br/>positions_update<br/>aircraft_new<br/>aircraft_remove

📊 Analytics Tasks

📍

Location: skyspy/tasks/analytics.py

flowchart TB
    subgraph RealTime["⚡ Real-Time Analytics"]
        A1[refresh_acars_stats<br/>Every 60s]
        A2[update_antenna_analytics<br/>Every 5min]
        A3[update_safety_stats<br/>Every 30s]
    end

    subgraph NearRealTime["🔄 Near Real-Time"]
        B1[refresh_flight_pattern_geographic_stats<br/>Every 2min]
        B2[refresh_tracking_quality_stats<br/>Every 2min]
        B3[refresh_engagement_stats<br/>Every 2min]
        B4[refresh_time_comparison_stats<br/>Every 5min]
    end

    subgraph Aggregation["📈 Aggregation"]
        C1[aggregate_hourly_antenna_analytics<br/>Hourly]
        C2[calculate_daily_stats<br/>Daily]
    end

    RealTime --> NearRealTime --> Aggregation

{
    "max_range_by_direction": {"0": 125.3, "30": 98.7},
    "overall_max_range": 125.3,
    "avg_range": 45.2,
    "total_positions": 15234,
    "unique_aircraft": 127,
    "avg_rssi": -12.5,
    "coverage_percentage": 91.7,
    "range_percentiles": {"p50": 35.2, "p75": 52.1, "p90": 78.4, "p95": 95.2}
}

{
    "date": "2024-01-14",
    "total_sightings": 125000,
    "unique_aircraft": 1523,
    "total_sessions": 892,
    "military_sessions": 45,
    "avg_distance": 32.5,
    "max_distance": 156.2
}

🎙️ Transcription Tasks

📍

Location: skyspy/tasks/transcription.py

flowchart LR
    Audio[🎵 Audio File] --> Queue[📋 Transcription Queue]
    Queue --> Process[🔄 process_transcription_queue]
    Process --> Transcribe[🎙️ transcribe_audio]
    Transcribe --> Extract[✈️ extract_callsigns]
    Extract --> DB[(💾 Database)]

    style Audio fill:#bb8fce,color:#fff
    style Transcribe fill:#bb8fce,color:#fff

🌍 Geographic Data Tasks

📍

Location: skyspy/tasks/geodata.py

{
    "airports": 15234,
    "navaids": 8923,
    "geojson": 127
}

🔲 Airspace Tasks

📍

Location: skyspy/tasks/airspace.py

🗂️ OpenAIP Tasks

📍

Location: skyspy/tasks/openaip.py

📜 NOTAM Tasks

📍

Location: skyspy/tasks/notams.py

🗄️ External Database Tasks

📍

Location: skyspy/tasks/external_db.py

flowchart TB
    subgraph Sources["📡 Data Sources"]
        S1[ADS-B Exchange]
        S2[tar1090]
        S3[FAA Registry]
        S4[OpenSky Network]
        S5[HexDB]
        S6[Planespotters]
    end

    subgraph Sync["🔄 Sync Tasks"]
        T1[sync_external_databases<br/>Daily 4:00 AM]
        T2[update_stale_databases<br/>Every 6 hours]
        T3[fetch_aircraft_info<br/>On-demand]
        T4[fetch_aircraft_photos<br/>On-demand]
    end

    subgraph Storage["💾 Storage"]
        DB[(PostgreSQL)]
        Cache[(Redis Cache)]
    end

    Sources --> Sync --> Storage

🧹 Cleanup Tasks

📍

Location: skyspy/tasks/cleanup.py

flowchart LR
    subgraph Master["🎯 Master Task"]
        M[run_all_cleanup_tasks<br/>Daily 3:00 AM]
    end

    subgraph Children["📋 Cleanup Tasks"]
        C1[cleanup_old_sightings<br/>30 days]
        C2[cleanup_old_sessions<br/>90 days]
        C3[cleanup_old_alert_history<br/>30 days]
        C4[cleanup_old_safety_events<br/>90 days]
        C5[cleanup_old_acars_messages<br/>7 days]
    end

    subgraph Maintenance["🔧 Maintenance"]
        V[vacuum_analyze_tables<br/>Weekly]
    end

    M --> Children
    Children --> Maintenance

🔔 Notification Tasks

📍

Location: skyspy/tasks/notifications.py

sequenceDiagram
    participant Alert as 🚨 Alert Trigger
    participant Queue as 📋 Notification Queue
    participant Task as 🔔 send_notification_task
    participant Apprise as 📤 Apprise
    participant Channels as 📱 Notification Channels

    Alert->>Queue: Queue notification
    Queue->>Task: Process
    Task->>Apprise: Send via Apprise
    Apprise->>Channels: Deliver

    alt Success
        Channels-->>Task: ✅ Delivered
    else Failure
        Channels-->>Task: ❌ Failed
        Task->>Queue: Retry with backoff
    end

🎯 Cannonball Mode Tasks

📍

Location: skyspy/tasks/cannonball.py

🚨 Real-time law enforcement aircraft detection and pattern analysis.

flowchart TB
    subgraph Detection["🎯 Detection Loop"]
        A[analyze_aircraft_patterns<br/>Every 5 seconds]
    end

    subgraph Processing["⚙️ Processing"]
        B[Pattern Analysis]
        C[Threat Scoring]
        D[Alert Generation]
    end

    subgraph Storage["💾 Storage"]
        E[(CannonballSession)]
        F[(CannonballPattern)]
        G[(CannonballAlert)]
    end

    subgraph Broadcast["📡 Broadcast"]
        H[threats_update]
        I[new_alert]
    end

    A --> B --> C --> D
    D --> E & F & G
    D --> H & I

📅 Complete Schedule Reference

⏱️

Celery Beat Schedule

All times are in UTC. The scheduler is configured in skyspy/celery.py.

⚡ High-Frequency Tasks (< 1 minute)

🔄 Medium-Frequency Tasks (1-10 minutes)

🕐 Low-Frequency Tasks (Hourly+)

📆 Daily Tasks

📅 Weekly Tasks

🔄 Retry Policy

flowchart LR
    subgraph Attempt["🎯 Task Execution"]
        A[Execute Task]
    end

    subgraph Decision{"Success?"}
        B{Check Result}
    end

    subgraph Success["✅ Success"]
        C[Complete]
    end

    subgraph Retry["🔄 Retry Logic"]
        D[Calculate Backoff]
        E[Add Jitter]
        F[Schedule Retry]
    end

    subgraph Fail["❌ Max Retries"]
        G[Mark Failed]
        H[Log to Sentry]
    end

    A --> B
    B -->|Yes| C
    B -->|No| D --> E --> F --> A
    F -->|Max Reached| G --> H

⚙️ Retry Configuration

💡

Best Practice

Use expires for time-sensitive tasks to prevent queue buildup when the system is under load.

@shared_task(
    bind=True,
    max_retries=3,
    autoretry_for=(Exception,),
    retry_backoff=True,
    retry_backoff_max=3600,
    retry_jitter=True,
)
def example_task(self):
    try:
        # Task logic
        pass
    except Exception as e:
        raise self.retry(exc=e, countdown=60)

📊 Dead Letter Handling

Failed notifications are tracked in NotificationLog with status:

📊 Monitoring Dashboard

🌸 Flower Dashboard

Deploy Flower for real-time task monitoring:

celery -A skyspy flower --port=5555

🌐

Access at http://localhost:5555

🖥️ Command-Line Monitoring

# Watch active tasks
celery -A skyspy inspect active

# View scheduled tasks
celery -A skyspy inspect scheduled

# Check worker stats
celery -A skyspy inspect stats

# View registered tasks
celery -A skyspy inspect registered

📝 Logging Configuration

LOGGING = {
    'handlers': {
        'celery': {
            'level': 'INFO',
            'class': 'logging.FileHandler',
            'filename': '/var/log/skyspy/celery.log',
            'formatter': 'verbose',
        },
    },
    'loggers': {
        'celery': {
            'handlers': ['celery'],
            'level': 'INFO',
            'propagate': False,
        },
        'skyspy.tasks': {
            'handlers': ['celery'],
            'level': 'DEBUG',
            'propagate': False,
        },
    },
}

🔍 Sentry Integration

SkysPy automatically reports task errors to Sentry when configured:

from skyspy.utils.sentry import capture_task_error

try:
    # Task logic
    pass
except Exception as e:
    capture_task_error(e, 'task_name', extra={'key': 'value'})
    raise

⚡ Performance Tuning

🎯

Optimization Guide

🏭 Worker Configuration

celery -A skyspy worker \
    --pool=gevent \
    --concurrency=100 \
    --loglevel=INFO \
    -Q polling,default,database,notifications,transcription,low_priority

celery -A skyspy worker \
    --pool=prefork \
    --concurrency=4 \
    --loglevel=DEBUG \
    -Q polling,default,database,notifications,transcription,low_priority

🎛️ Queue-Specific Workers

⚡

High-Load Tip

For high-load deployments, run dedicated workers per queue to isolate time-critical tasks.

# High-priority polling worker
celery -A skyspy worker -Q polling -c 50 --pool=gevent -n polling@%h

# Database operations worker
celery -A skyspy worker -Q database -c 10 --pool=prefork -n database@%h

# Low-priority worker
celery -A skyspy worker -Q low_priority -c 4 --pool=prefork -n lowprio@%h

🧠 Memory Optimization

# Disable prefetching for time-sensitive tasks
app.conf.worker_prefetch_multiplier = 1

# Acknowledge after completion (prevents lost tasks)
app.conf.task_acks_late = True

# Re-queue tasks if worker dies
app.conf.task_reject_on_worker_lost = True

🍓 Raspberry Pi Optimization

📦

Resource-Constrained Devices

SkysPy includes RPi-specific optimizations in settings_rpi.py.

🔴 Redis Configuration

# redis.conf - Production settings
maxmemory 512mb
maxmemory-policy allkeys-lru
appendonly yes
appendfsync everysec

🔧 Configuration Reference

📋 Django Settings

🗓️ Retention Configuration

🎛️ Feature Flags

🐳 Docker Deployment

📦 Docker Compose Configuration

services:
  redis:
    image: redis:7-alpine
    command: redis-server --appendonly yes
    volumes:
      - redis_data:/data
    ports:
      - "6379:6379"
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s
      timeout: 5s
      retries: 5

  celery-worker:
    build: ./skyspy_django
    command: celery -A skyspy worker -l INFO -Q polling,default,database,notifications,transcription,low_priority
    environment:
      - CELERY_BROKER_URL=redis://redis:6379/0
    depends_on:
      redis:
        condition: service_healthy

  celery-beat:
    build: ./skyspy_django
    command: celery -A skyspy beat -l INFO
    environment:
      - CELERY_BROKER_URL=redis://redis:6379/0
    depends_on:
      redis:
        condition: service_healthy

🔥 Troubleshooting

# Check worker is connected
celery -A skyspy inspect ping

# Verify beat is running
celery -A skyspy inspect scheduled

# Check queue length
redis-cli LLEN celery

# Purge all tasks (use cautiously!)
celery -A skyspy purge

# Restart workers periodically
celery -A skyspy control shutdown

# Test Redis connectivity
redis-cli ping

# Check Redis memory
redis-cli info memory

🏥 Health Checks

SkysPy exposes a Celery health check via the celery_heartbeat cache key:

from django.core.cache import cache

def celery_health_check():
    return cache.get('celery_heartbeat', False)

💡

This is updated every 60 seconds by update_stats_cache.

📚 Additional Resources

• Celery Documentation
• Django Celery Beat
• Redis Documentation
• Flower Monitoring