# Project export: ATCMonitor

This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small.

## Project metadata

- Hackathon: UC Berkeley AI Hackathon 2025
- Tagline: ATCMonitor analyzes thousands of past flights to define safe corridors, then continuously monitors live flights/frequencies for outliers to catch anomalies long before they become emergencies.
- Devpost: https://devpost.com/software/atcmonitor
- GitHub: https://github.com/twangodev/atcmonitor
- Team: 3 GitHub contributor(s) — James Ding (7 commits), hhassan14 (2 commits), Bryan Yu (1 commits)

## Devpost submission (written by the team)

### Inspiration

Various airline accidents happen due to difficulty in coordinating between pilots and air traffic control. We wanted to make communication easier and more automated to aid air traffic controllers in spotting abnormalities in air traffic earlier.

### What it does

By ingesting thousands of historical flights, we define safe corridors through machine learning. We then continuously monitor live flights and radio frequencies, ensuring that outliers are caught before they shape into actual emergencies.

### How we built it

ADSB Flight Data and VHF voice data requires ingesting massive amounts of data, especially when scaled to multiple regions. To ensure this is scalable, we use Kafka as a backbone for our data processing pipeline. To quickly iterate and prototype, we chose to wrote this project in Python, but should consider rewriting it in a higher performance programming language, which comes with benefits such as parallel processing.

### Challenges we ran into

To transcribe air traffic control speech, our team utilized a fine-tuned model of whisper. Although this version was better than the original whisper model, air traffic control radio is often filled with static and hard-to-hear speakers, so transcribing was a challenge. To aid whisper, we chunked the data into sections where someone was the speaker and static, so whisper could focus on transcribing the most important parts of the audio.

### Accomplishments we're proud of

Being able to get interactive visualizations of flight paths and to have a clustering algorithm to find common flight paths and abnormalities was great. We are also happy to be able to not only transcribe air traffic control data, but also predict if the tower or pilot was speaking based on the context of the words.

### What we learned

Throughout our experience, we learned how complex flight data could be, with a combination of slight deviations in flight paths based on the situation and difficult-to-hear air traffic control. However, with this complex system also comes appreciation for the various people and systems that make air travel one of the safest modes of transportation. Through these systems, we learned more about how to use multiple libraries in conjunction with each other, such as whisper with a text classifier and ultimately with Kafka.

### What's next

ATCMonitor would benefit from being fully connected at every part of the workflow, from taking path data to clustering and processing audio data. When combining these data together, we can get a powerful prediction of when things go wrong and are able to alert the proper controllers to resolve the situation and keep the skies safe. By having this system run live in airports across the world, we can help air traffic controllers catch mistakes that may slip and ensure the safety of all flights.

## README (from the GitHub repository)

# atcmonitor

ATCMonitor is a tool for creating visual maps of what air traffic controllers are doing. It is designed to help visualize the flow of air traffic and the activities of air traffic controllers in a given area.

## Features

- Visualize air traffic controller activities
- Create maps of air traffic control operations
- Log and analyze air traffic data, including VHF and ADS-B data
- Support for multiple data sources
- Validate and process air traffic data, alerting controllers to potential issues

## Screenshots

### KSFO (San Francisco International Airport)

#### Tracks
Visual representation of aircraft tracks around San Francisco International Airport (KSFO).

![sfo-tracks.png](assets/sfo-tracks.png)

Darker lines indicate commonly used flight paths, while lighter lines represent less frequently used paths.

#### Clustering

To determine when an aircraft is on a common flight path, the tool clusters aircraft tracks. This helps in identifying patterns and common routes taken by aircraft.

![sfo-clustering.png](assets/sfo-cluster.png)

Blue indicates high precision, while red indicates low precision but very frequent paths. Generally, red regions signal departure and arrive coordination by air traffic controllers or pilots, while blue regions are used for high precision, en route traffic.

Rules can then be written to alert controllers when aircraft deviate from these common paths, helping to maintain safety and efficiency in air traffic control operations. By default, aircraft within 5-mile radius of an airport must be within a valid cluster; otherwise an alert is generated.

## Credits

This repository uses [dump1090](https://github.com/flightaware/dump1090) for decoding live ADS-B data and [RTLSDR-Airband](https://github.com/rtl-airband/RTLSDR-Airband) for decoding live VHF data.

Historical data is provided by the [OpenSky Network](https://opensky-network.org/), which offers a large dataset of air traffic control data.



## Detected evidence (automated analysis)

Indexed codebase: 19 recognized source files, 58 KB.
- Python (language) — detected in the code
- Docker (technology) — claimed on Devpost, not found in the code

## Codebase structure (from repository index)

### Files (27 of 27)

```
adsb-producer/main.py
adsb-producer/Pipefile
adsb-producer/Pipfile
adsb-producer/Pipfile.lock
audio/main.py
audio/Pipfile
audio/Pipfile.lock
audio/vad.py
audio/whisper_enhanced.py
audio/whisper1.py
docker-compose.adsb.yml
docker-compose.radio.yml
docker-compose.yml
opensky-producer/3dcluster.py
opensky-producer/cache.py
opensky-producer/cluster.py
opensky-producer/dataset.py
opensky-producer/flight_envelopes.geojson
opensky-producer/geojson.py
opensky-producer/locationAnomaly.py
opensky-producer/locationAnomaly3d.py
opensky-producer/main.py
opensky-producer/Pipfile
opensky-producer/Pipfile.lock
opensky-producer/s3_source.py
opensky-producer/visualize.py
README.md
```

### Dependencies

No dependency index available.

### Recent commits (newest first)

- Add ADSB and Radio docker-compose files
- Update flight_envelopes.geojson for KSFO
- Update README.md to include links for dump1090 and RTLSDR-Airband
- Update README.md with project description, features, and usage examples
- Add docker-compose configuration for Zookeeper and Kafka services
- Add opensky-producer
- Add adsb-producer
- Move audio into seperate directory
- audio processing
- Initial commit

## Key source files (fetched from GitHub, selected and truncated for size)

### docker-compose.yml

```yaml
services:

  zookeeper:
    image: confluentinc/cp-zookeeper:7.4.0
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  broker:
    image: confluentinc/cp-kafka:7.4.0
    container_name: broker
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1

```

### audio/main.py

```python
import re
from pathlib import Path
from typing import List

from vad import chunk
from whisper_enhanced import file_to_speaker

_CHUNK_SEG_RE = re.compile(r'chunk_(\d+)_seg_(\d+)\.mp3$', re.IGNORECASE)

def get_chunk_segments(chunk_path: str) -> List[str]:
    """
    Return a list of all .mp3 files named chunk_<x>_seg_<y>.mp3,
    sorted by x (chunk) then y (segment).
    """
    p = Path(chunk_path)
    files = [
        f for f in p.glob('*.mp3')
        if f.is_file() and _CHUNK_SEG_RE.match(f.name)
    ]

    def sort_key(f: Path):
        m = _CHUNK_SEG_RE.match(f.name)
        # safe to unwrap because we filtered above
        chunk_num, seg_num = map(int, m.groups())
        return (chunk_num, seg_num)

    files.sort(key=sort_key)
    return [str(f) for f in files]

def run_whisper():
    for path in get_chunk_segments("./chunks"):
        file_to_speaker(path)

if __name__ == "__main__":
    # chunk()
    run_whisper()

```

### opensky-producer/main.py

```python
import json
from os import environ

import pandas as pd
from kafka import KafkaProducer
from tqdm import tqdm

from dataset import Dataset, df_near_coordinates
from s3_source import get_tar_references

bootstrap_servers = environ.get("KAFKA_BOOTSTRAP_SERVERS", "localhost:9092").split(",")

PRODUCER_TOPIC = "adsb.historical"

center = (37.6191, -122.3816)  # Example coordinates for San Francisco International Airport

def main(
    should_sum_dfs: bool = False,
    should_send_to_kafka: bool = True,
):

    producer = KafkaProducer(
        bootstrap_servers=bootstrap_servers,
        value_serializer=lambda v: json.dumps(v).encode("utf-8")
    )

    tars = get_tar_references()
    datasets = [Dataset(tar_tuple) for tar_tuple in tars][:50] # TODO remove slice for production/larger compute

    dfs = []
    for dataset in tqdm(datasets, desc="Processing Datasets", unit="dataset"):
        df = dataset.request_df()
        print(f"Processed dataset with {len(df)} rows.")

        df_near_sfo = df_near_coordinates(df, center, radius_miles=100)

        df_clean = df_near_sfo.astype(object).where(df_near_sfo.notnull(), None)

        if should_send_to_kafka:
            for record in tqdm(df_clean.to_dict(orient="records"), desc="Sending Records to Kafka", unit="record"):
                producer.send(PRODUCER_TOPIC, value=record)

        if should_sum_dfs:
            dfs.append(df_clean)

        producer.flush()

    if should_sum_dfs:
        summed_df = pd.concat(dfs, ignore_index=True)
        print(f"Summed DataFrame has {len(summed_df)} rows.")
        return summed_df.sort_values(['icao24','time'])

    return None

if __name__ == "__main__":
    main()

```

### adsb-producer/main.py

```python
import subprocess
from enum import IntEnum
import json
from datetime import datetime
from os import environ
from kafka import KafkaProducer
import pandas as pd 
import csv


TESTMODE = 1


hostname = "192.168.128.229"
port = 30003
bootstrap_servers = environ.get("KAFKA_BOOTSTRAP_SERVERS", "localhost:9092").split(",")
PRODUCER_TOPIC = "adsb.live" 
producer = KafkaProducer(
    bootstrap_servers=bootstrap_servers,
    value_serializer=lambda v: json.dumps(v).encode("utf-8")
)

roles = {
    "MSG1": [1,1,1,1,1,1,1,1,1,1,1,0,0,0,0,0,0,0,0,0,0,0],
    "MSG2": [1,1,1,1,1,1,1,1,1,1,0,1,1,1,1,1,0,0,0,0,0,1],
    "MSG3": [1,1,1,1,1,1,1,1,1,1,0,1,0,0,1,1,0,0,1,1,1,1],
    "MSG4": [1,1,1,1,1,1,1,1,1,1,0,0,1,1,0,0,1,0,0,0,0,0],
    "MSG5": [1,1,1,1,1,1,1,1,1,1,0,1,0,0,0,0,0,0,1,0,1,1],
    "MSG6": [1,1,1,1,1,1,1,1,1,1,0,1,0,0,0,0,0,1,1,1,1,1],
    "MSG7": [1,1,1,1,1,1,1,1,1,1,0,1,0,0,0,0,0,0,0,0,0,1],
    "MSG8": [1,1,1,1,1,1,1,1,1,1,0,0,0,0,0,0,0,0,0,0,0,1],
    "SEL":  [1,0,1,1,1,1,1,1,1,1,1,0,0,0,0,0,0,0,0,0,0,0],
    "ID":   [1,0,1,1,1,1,1,1,1,1,1,0,0,0,0,0,0,0,0,0,0,0],
    "AIR":  [1,0,1,1,1,1,1,1,1,1,0,0,0,0,0,0,0,0,0,0,0,0],
    "STA":  [1,0,1,1,1,1,1,1,1,1,0,0,0,0,0,0,0,0,0,0,0,0],
    "CLK":  [1,0,1,0,0,0,1,1,1,1,0,0,0,0,0,0,0,0,0,0,0,0]
}

class Message(IntEnum):
    messageType = 0
    transmissionType = 1
    sessionID = 2
    aircraftID = 3
    hexIdent = 4
    flightID = 5
    dataMessageGenerated = 6
    timeMessageGenerated = 7

    dateMessageLogged = 8
    timeMessageLogged = 9
    
    #non standard messages
    callsign = 10
    altitude = 11
    groundSpeed = 12
    track = 13
    lat = 14
    lon = 15
    verticalRate = 16
    squawk = 17
    alert = 18
    emergency = 19
    SPI = 20
    isOnGround = 21



def createJson(data, flag):

    message = {}
    #### non flagged
    dt = datetime.strptime(data[Message.dataMessageGenerated] + " " + data[Message.timeMessageGenerated], "%Y/%m/%d %H:%M:%S.%f")
    message['time'] = dt.timestamp()

    
    message['icao24'] = data[Message.hexIdent]
    ####

    message['lat'] = float(data[Message.lat]) if (flag[Message.lat] and data[Message.lat] != '') else None
    message['lon'] = float(data[Message.lon]) if (flag[Message.lon] and data[Message.lon] != '') else None
    message['velocity'] = float(data[Message.groundSpeed]) if (flag[Message.groundSpeed] and data[Message.groundSpeed] != '') else None
    
    message['heading'] = float(data[Message.track]) if (flag[Message.track] and data[Message.track] != '') else None

    message['vertrate'] = float(data[Message.verticalRate]) if (flag[Message.verticalRate] and data[Message.verticalRate] != '') else None
    message['callsign'] = data[Message.callsign] if (flag[Message.callsign] and data[Message.callsign] != '') else None
    message['onground'] = bool(int(data[Message.isOnGround])) if (flag[Message.isOnGround] and data[Message.isOnGround] != '') else None
    message['alert'] = bool(int(data[Message.alert])) if (flag[Message.alert] and data[Message.alert] != '') else None
    message['spi'] = bool(int(data[Message.SPI])) if (flag[Message.SPI] and data[Message.SPI] != '') else None
    message['squawk'] = float(data[Message.squawk]) if (flag[Message.squawk] and data[Message.squawk] != '') else None

    message['baroaltitude'] = None
    message['geoaltitude'] = float(data[Message.altitude]) if (flag[Message.altitude] and data[Message.altitude] != '') else None 

    
    
    dt = datetime.strptime(data[Message.dateMessageLogged] + " " + data[Message.timeMessageLogged], "%Y/%m/%d %H:%M:%S.%f")

    message['lastposupdate'] = dt.timestamp() if (flag[Message.lon] and data[Message.lon] != '' and flag[Message.lat] and data[Message.lat] != '') else None
    message['lastcontact'] = dt.timestamp()
    return(json.dumps(message))




def main():

    nc_command = ["nc", hostname, str(port)]

    # Create a Popen object, redirecting stdout to a pipe
    process = subprocess.Popen(nc_command, stdout=subprocess.PIPE)

    # Read data line by line from Netcat's stdout
    for line in process.stdout:
        #print(f"Received: {line.decode().strip()}")

        try:
            decodedLine = line.decode().strip()
            messagesList = decodedLine.split(",")
            messageType = messagesList[Message.messageType]
            transmissionType = messagesList[Message.transmissionType] 

            json = createJson(messagesList, roles[(messageType+transmissionType)])
            producer.send(PRODUCER_TOPIC, value=json)
        except:
            continue
    
    #print(str(messagesList[Message.groundSpeed])+ '\n')
    producer.flush()
    process.wait()

def test():
    with open('datafeed.csv') as f:
        a = [{k: v for k, v in row.items()}
            for row in csv.DictReader(f, skipinitialspace=True)]

    for row in a:
        jsonResult = json.dumps(row)
        print(jsonResult)
        producer.send(PRODUCER_TOPIC, value=jsonResult)

    producer.flush()


if __name__ == "__main__":
    if TESTMODE:
        test()
    else:
        main()
```

### docker-compose.radio.yml

```yaml
services:

  rtlsdr_airband:
    image: fredclausen/rtlsdrairband
    tty: true
    container_name: rtlsdrairband
    restart: always
    devices:
      - /dev/bus/usb:/dev/bus/usb
    ports:
      - 8000:8000
    environment:
      - RTLSDRAIRBAND_FREQS=121.8

```

### docker-compose.adsb.yml

```yaml
services:

  dump1090:
    image: jraviles/dump1090
    ports:
      - 8080:8080
      - 30001:30001
      - 30002:30002
      - 30003:30003
      - 30004:30004
      - 30005:30005
      - 30104:30104
    devices:
      - /dev/bus/usb
    restart: unless-stopped

```

### opensky-producer/cache.py

```python
import hashlib
from pathlib import Path
from typing import Union

cache_dir = Path("./cache")
cache_dir.mkdir(parents=True, exist_ok=True)

def _hash_key(key: str) -> str:
    """Return a filesystem-safe filename for any string key."""
    h = hashlib.sha256(key.encode("utf-8")).hexdigest()
    return h

def read_cache(key: str) -> bytes | None:
    """
    Read cached data by key.
    Returns the raw string, or None if not found.
    """
    fname = cache_dir / _hash_key(key)
    try:
        return fname.read_bytes()
    except FileNotFoundError:
        return None

def write_cache(key: str, data: Union[str, bytes], extension: str = "") -> None:
    """
    Write data to cache under its hashed name.
    Accepts either bytes (any binary) or str (will be UTF-8 encoded).
    """
    file_hash = _hash_key(key)
    if extension:
        fname = cache_dir / f"{file_hash}.{extension}"
    else:
        fname = cache_dir / file_hash
    fname.parent.mkdir(parents=True, exist_ok=True)

    fname.write_bytes(data)
```

### opensky-producer/s3_source.py

```python
import io
import tarfile

import boto3
from botocore import UNSIGNED
from botocore.config import Config
from tqdm import tqdm

from cache import read_cache, write_cache

def get_tar_references(
        bucket="data-samples",
        prefix="states/",
        endpoint="https://s3.opensky-network.org",
):
    s3 = boto3.client(
        "s3",
        endpoint_url=endpoint,
        config=Config(signature_version=UNSIGNED),
    )

    paginator = s3.get_paginator("list_objects_v2")
    pages = paginator.paginate(Bucket=bucket, Prefix=prefix)
    csv_tars = [
        obj["Key"]
        for page in tqdm(pages, desc="Listing S3 Pages", unit="file")
        for obj in page.get("Contents", [])
        if (
            obj["Key"].lower().endswith(".tar")
            and obj["Size"] > 1024 ** 2  # at least 1 MB
            and "csv" in obj["Key"].lower()
        )
    ]

    handlers = []
    for key in csv_tars:
        def make_handler(k):
            def fetch_and_extract():
                data = read_cache(k)
                if data is None:
                    resp = s3.get_object(Bucket=bucket, Key=k)
                    data = resp["Body"].read()
                    write_cache(k, data)

                buf = io.BytesIO(data)
                result = {}
                with tarfile.open(fileobj=buf, mode="r:*") as tar:
                    for member in tar.getmembers():
                        if member.isreg():
                            f = tar.extractfile(member)
                            result[member.name] = f.read()
                return result

            return fetch_and_extract

        handlers.append((key, make_handler(key)))

    return handlers

```

### opensky-producer/dataset.py

```python
import gzip
from io import BytesIO

import numpy as np
import pandas
import pandas as pd
from sklearn.neighbors import BallTree

class Dataset:

    def __init__(self, tar_tuple: tuple[str, callable]):
        self.tar_key, self.fetch_handler = tar_tuple

    def request_df(self):
        file = self.fetch_handler()

        for name, content in file.items():
            if name.lower().endswith(".csv.gz"):
                with gzip.GzipFile(fileobj=BytesIO(content)) as gz:
                    csv_bytes = gz.read()
                    df = pd.read_csv(
                        BytesIO(csv_bytes),
                    )
                    return df

        raise FileNotFoundError("No CSV file found in the tar archive.")

def df_near_coordinates(df: pandas.DataFrame, center: tuple[float, float], radius_miles: float) -> pandas.DataFrame:
    """
    Find all rows in the DataFrame within a given radius of a center coordinate.

    :param df: DataFrame containing 'lat' and 'lon' columns.
    :param center: Tuple of (latitude, longitude) for the center point.
    :param radius_miles: Radius in miles to search around the center.
    :return: DataFrame with rows within the specified radius.
    """
    df = df.copy()

    df["lat"] = pd.to_numeric(df["lat"], errors="coerce")
    df["lon"] = pd.to_numeric(df["lon"], errors="coerce")

    df = df.dropna(subset=["lat", "lon"]).reset_index(drop=True)

    coords = np.deg2rad(df[["lat", "lon"]].values)
    tree = BallTree(coords, metric="haversine")

    center_rad = np.deg2rad([center])
    earth_radius_miles = 3958.8
    radius_radians = radius_miles / earth_radius_miles

    indices = tree.query_radius(center_rad, r=radius_radians)[0]

    return df.iloc[indices].reset_index(drop=True)

```

### audio/whisper_enhanced.py

```python
from faster_whisper import WhisperModel
from transformers import pipeline
from typing import Dict, List, Tuple, Union


def file_to_speaker(audio_path: str) -> Union[Dict[str, object], None]:
    """
    Transcribe an audio file and classify its speaker role.

    Returns a dict with:
      - "values": List of (timestamp_seconds, word) tuples
      - "category": "Pilot" or "Tower"
    """
    # 1) Load your Whisper model
    asr = WhisperModel(
        "jacktol/whisper-medium.en-fine-tuned-for-ATC-faster-whisper",
        device="cuda",
        compute_type="float32",
    )

    # 2) Transcribe with word-level timestamps
    segments, _ = asr.transcribe(
        audio_path,
        language="en",
        condition_on_previous_text=True,
        beam_size=10,
        # no_speech_threshold=0.1,
        log_prob_threshold=-0.5,
        temperature=[0.0],
        chunk_length=60,
        vad_filter=True,
        word_timestamps=True,
    )

    # 3) Flatten out (timestamp, word) pairs
    values: List[Tuple[float, str]] = []
    for seg in segments:
        for w in seg.words:
            values.append((w.start, w.word))

    # 4) Build the full transcript for classification
    # Use recognized words instead of seg.text to ensure transcript is not empty
    full_text = " ".join(tup[1].strip() for tup in values if tup[1].strip())

    if not full_text:
        return None


    # 5) Load and run your speaker-role classifier
    role_clf = pipeline(
        "text-classification",
        model="jacktol/atc-pilot-speaker-role-classification-model",
        tokenizer="jacktol/atc-pilot-speaker-role-classification-model",
    )
    pred = role_clf(full_text)[0]
    raw_lbl = pred["label"]

    # print(f"[DEBUG] Raw label from classifier: {raw_lbl}")

    # 6) Map to human-friendly category
    label_map = {
        "PILOT":   "Pilot",
        "ATC":     "Tower",
        "LABEL_0": "Tower",   # fallback if generic
        "LABEL_1": "Pilot",
    }
    category = label_map.get(raw_lbl.upper(), raw_lbl)

    # print(f"[{audio_path}] [{category}]: {full_text}")
    print(f"[{category}]: {full_text}")
    # 7) Return the results
    return {
        "values": values,
        "category": category,
    }
```

[8 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]