DEV Community

Matthias
Matthias

Posted on • Originally published at Medium

From the Strait of Hormuz to Singapore: Building a Real-Time Ship Tracking Analyser with Flask and AISStream

A practical story about building a live vessel-tracking dashboard, hitting real-world data limitations, and turning a failed first location into a better geospatial prototype.

I started this project with a simple idea:

Can I build a small web app that watches ship movements in a strategic maritime area and turns live AIS data into something useful?

The first target was the Strait of Hormuz.

That choice made sense on paper. It is one of the most important maritime chokepoints in the world, and it looked like the perfect place to analyse tanker activity, slow traffic, vessel movement, and possible AIS reporting gaps.

But then reality showed up.

I could not get reliable live AIS signals for the area I selected.

Maybe it was my bounding box. Maybe it was provider coverage. Maybe it was filtering. Maybe it was the exact message types I requested. Maybe it was a mix of everything.

That is where the project became more interesting.

Instead of faking the data or forcing the original idea, I moved the prototype to the Singapore Strait, where live vessel traffic was easier to work with. That small pivot turned the project from β€œa ship map” into a useful lesson about live data systems:

Sometimes the hardest part of a data application is not the code. It is getting dependable data in the first place.

What I Built

The result is a small real-time ship tracking analyser built with:

  • Python
  • Flask
  • AISStream
  • WebSockets
  • Leaflet
  • OpenStreetMap tiles

The app runs locally. Flask connects to AISStream on the server side, keeps recent vessel positions in memory, and exposes a simple JSON endpoint for the browser dashboard.

The browser never connects directly to AISStream.

You can find the full source code here:

πŸ‘‰ GitHub repository: dataai01/ship-analyser

That was intentional.

I did not want the API key inside frontend JavaScript, so the architecture looks like this:

AISStream WebSocket
        |
        v
Flask background listener
        |
        v
In-memory vessel cache
        |
        v
Flask JSON API
        |
        v
Leaflet dashboard
Enter fullscreen mode Exit fullscreen mode

This keeps the first version small, understandable, and local-first.

The Current Monitoring Zone

The current default zone is the Singapore Strait:

Latitude:  1.00 to 1.60
Longitude: 103.40 to 104.20
Enter fullscreen mode Exit fullscreen mode

The dashboard lets me change the bounding box from the browser, restart the stream, clear the cache, and tune thresholds for stale vessels, AIS gaps, and slow-moving ships.

This was important because I did not want the application to be locked to one region. Singapore is the current default, but the same approach can be reused for other maritime zones.

Main Features

1. Live vessel map

The dashboard displays vessels on a Leaflet map.

Each vessel is plotted by latitude and longitude, and the marker heading is based on course over ground. The active bounding box is also drawn on the map, so I can immediately see which area the stream is supposed to cover.

This sounds simple, but it is very useful when debugging live geospatial data. If nothing appears, the first question is always:

Am I even looking at the right area?

2. Vessel classification

The analyser groups vessels into simple categories:

  • tankers
  • cargo/container vessels
  • service vessels
  • other vessels

The classification is intentionally basic. It uses AIS ship type data when available and handles both numeric AIS type codes and text labels.

It is not a perfect maritime classification system, but it is enough for a first dashboard and makes the traffic summary much easier to understand.

3. Traffic summary cards

The dashboard shows quick metrics for the monitored zone:

  • active vessels in the zone
  • active tankers
  • cargo/container vessels
  • slow or stopped vessels
  • AIS gap alerts
  • average active speed
  • estimated vessel directions
  • hourly traffic trend

A vessel is considered active when it is inside the selected zone and has reported recently enough to pass the stale threshold.

4. Slow or stopped vessel detection

Ships below a configurable speed threshold are marked as slow or stopped.

The default threshold is:

3 knots
Enter fullscreen mode Exit fullscreen mode

This is not meant to be a final suspicious-behaviour detector. A vessel may be slow because it is anchoring, waiting, manoeuvring, entering port, or following local traffic rules.

But as a first signal, it is useful.

A better version could compare speed against:

  • vessel type
  • anchorage zones
  • port boundaries
  • shipping lanes
  • historical behaviour
  • nearby vessel density

5. AIS gap alerts

For each MMSI, the app keeps a short in-memory track history.

If the time between two received positions is greater than the configured gap threshold, the dashboard flags that vessel with an AIS gap alert.

This does not automatically mean suspicious activity.

AIS gaps can happen because of:

  • reception coverage
  • satellite or terrestrial receiver limitations
  • provider filtering
  • transponder behaviour
  • network issues
  • message type filtering

But gaps are still worth inspecting, especially when combined with speed, location, vessel type, and historical context.

How the Flask Backend Works

The backend is where most of the interesting work happens.

The Flask app has three main responsibilities:

  1. keep the AISStream API key on the server
  2. maintain one WebSocket connection to AISStream
  3. expose clean dashboard data through local API endpoints

Keeping the API key out of the browser

The first rule was simple: the browser should never see the AISStream API key.

So the backend reads the key from either an environment variable or a local key file:

def read_api_key() -> str:
    env_key = os.getenv("AISSTREAM_API_KEY", "").strip()
    if env_key:
        return env_key

    for filename in (
        "my_key_file.txt",
        "aisstream_api_key.txt",
        "api_key.txt",
        ".aisstream_key",
        "aisstream.key",
    ):
        path = APP_DIR / filename
        if path.exists() and path.is_file():
            text = path.read_text(encoding="utf-8", errors="replace").strip()
            if text:
                return text.splitlines()[0].strip()

    raise RuntimeError("No AISStream API key found.")
Enter fullscreen mode Exit fullscreen mode

The important design choice is not the exact filename list. The important part is that the key stays server-side.

The browser only talks to my Flask app.

Subscribing to a geographic bounding box

AISStream lets the app subscribe to a bounding box. I use that to define the monitored region.

def make_subscription(api_key: str, zone: Dict[str, float]) -> Dict[str, Any]:
    return {
        "APIKey": api_key,
        "BoundingBoxes": [
            [
                [zone["min_lat"], zone["min_lon"]],
                [zone["max_lat"], zone["max_lon"]],
            ]
        ],
        "FilterMessageTypes": [
            "PositionReport",
            "StandardClassBPositionReport",
            "ExtendedClassBPositionReport",
            "ShipStaticData",
        ],
    }
Enter fullscreen mode Exit fullscreen mode

This is what makes the app reusable. The Singapore Strait is just the default zone. The dashboard can send new latitude and longitude values, and the stream can be restarted for a different area.

Running the AIS WebSocket in the background

Flask handles HTTP requests, but the AISStream connection is a long-running WebSocket.

To keep the dashboard responsive, I run the stream listener in a background thread. Inside that thread, an async loop connects to AISStream and receives messages.

async def aisstream_loop(zone: Dict[str, float], stop_event: threading.Event) -> None:
    api_key = read_api_key()
    subscription = make_subscription(api_key, zone)

    while not stop_event.is_set():
        try:
            async with websockets.connect(AISSTREAM_URL, ping_interval=20, ping_timeout=20) as websocket:
                await websocket.send(json.dumps(subscription))
                update_status(connected=True, last_error=None)

                while not stop_event.is_set():
                    raw_message = await asyncio.wait_for(websocket.recv(), timeout=5)
                    message = json.loads(raw_message)
                    record_stream_message()
                    handle_aisstream_message(message)

        except Exception as exc:
            update_status(connected=False, last_error=str(exc))
            await asyncio.sleep(5)
Enter fullscreen mode Exit fullscreen mode

This part also explains why the dashboard can show connection state. Every reconnect, error, and received message updates the stream status dictionary.

Storing recent positions in memory

For the first version, I did not use a database.

Instead, I store recent vessel positions in memory by MMSI:

VESSEL_TRACKS: Dict[str, Deque[AisPoint]] = defaultdict(
    lambda: deque(maxlen=MAX_POINTS_PER_MMSI)
)
Enter fullscreen mode Exit fullscreen mode

Each vessel position becomes an AisPoint:

@dataclass
class AisPoint:
    mmsi: str
    timestamp: datetime
    lat: float
    lon: float
    sog: float
    cog: float
    ship_type: str
    name: str
    destination: str = ""
Enter fullscreen mode Exit fullscreen mode

When a position report arrives, the app normalises the data and appends it to that vessel's track:

def store_position(report: Dict[str, Any]) -> None:
    mmsi = str(report.get("UserID", "")).strip()
    lat = to_float(report.get("Latitude"), default=float("nan"))
    lon = to_float(report.get("Longitude"), default=float("nan"))

    if not mmsi or lat != lat or lon != lon:
        return

    with DATA_LOCK:
        point = AisPoint(
            mmsi=mmsi,
            timestamp=utc_now(),
            lat=lat,
            lon=lon,
            sog=to_float(report.get("Sog", report.get("SpeedOverGround", 0.0))),
            cog=to_float(report.get("Cog", report.get("CourseOverGround", 0.0))),
            ship_type=normalise_ship_type(report.get("ShipType", "Unknown")),
            name=f"MMSI {mmsi}",
        )
        VESSEL_TRACKS[mmsi].append(point)
Enter fullscreen mode Exit fullscreen mode

This is deliberately simple. It is not meant to be permanent storage. It is just enough to power a live prototype.

Enriching positions with static vessel data

AIS position reports do not always include friendly vessel names or destinations.

That information often arrives separately as static vessel data, so the app stores it in another dictionary:

VESSEL_STATIC: Dict[str, Dict[str, Any]] = {}
Enter fullscreen mode Exit fullscreen mode

When static data arrives, the latest known vessel position can be backfilled with a better name, type, and destination.

def store_static_data(static_message: Dict[str, Any]) -> None:
    mmsi = str(static_message.get("UserID", static_message.get("Mmsi", ""))).strip()
    if not mmsi:
        return

    VESSEL_STATIC[mmsi] = {
        "name": str(static_message.get("Name") or f"MMSI {mmsi}").strip(),
        "ship_type": normalise_ship_type(static_message.get("ShipType") or "Unknown"),
        "destination": str(static_message.get("Destination") or "").strip(),
    }
Enter fullscreen mode Exit fullscreen mode

This makes the dashboard easier to read over time. A vessel may first appear as only an MMSI, then later gain a name and destination when static data is received.

Turning raw positions into dashboard metrics

The dashboard does not receive raw WebSocket messages.

It receives analysed data from Flask.

The analyse() function checks which vessels are inside the zone, which ones are stale, which ones are slow, and which ones have reporting gaps.

active_vessels = [v for v in vessels if v["active"]]
slow_vessels = [v for v in active_vessels if v["slow"]]
vessels_with_gaps = [v for v in vessels if v["gap_alerts"]]

category_counts = Counter(v["category"] for v in active_vessels)
direction_counts = Counter(v["direction"] for v in active_vessels)
Enter fullscreen mode Exit fullscreen mode

The final response is shaped for the frontend:

return {
    "summary": {
        "active_vessels_in_zone": len(active_vessels),
        "tankers_active": category_counts.get("Tankers", 0),
        "slow_or_stopped": len(slow_vessels),
        "ais_gap_alerts": len(vessels_with_gaps),
        "average_active_speed_knots": round(avg_speed, 2),
    },
    "counts": {
        "categories": dict(category_counts),
        "directions": dict(direction_counts),
    },
    "vessels": vessels,
}
Enter fullscreen mode Exit fullscreen mode

This keeps the browser code simple. It only needs to render the data, not understand AIS message formats.

Exposing local API endpoints

The Flask routes are intentionally small:

@app.get("/api/data")
def api_data():
    zone = get_zone_from_request()
    stale_minutes = int(to_float(request.args.get("stale_minutes"), DEFAULT_STALE_MINUTES))
    gap_minutes = int(to_float(request.args.get("gap_minutes"), DEFAULT_GAP_MINUTES))
    slow_knots = to_float(request.args.get("slow_knots"), DEFAULT_SLOW_KNOTS)

    return jsonify(analyse(zone, stale_minutes, gap_minutes, slow_knots))
Enter fullscreen mode Exit fullscreen mode

Starting the stream is also just an HTTP request:

@app.post("/api/stream/start")
def api_stream_start():
    zone = get_zone_from_request()
    status = start_stream(zone)
    return jsonify({"ok": True, "stream": status})
Enter fullscreen mode Exit fullscreen mode

This gives the dashboard a clean control surface:

  • fetch analysed data
  • start or restart the stream
  • stop the stream
  • clear the cache
  • check health

Polling from the browser

The frontend does not need a direct WebSocket connection.

It simply polls the Flask API every few seconds:

async function refreshData() {
  const res = await fetch(`/api/data?${queryParams()}`);
  const data = await res.json();
  render(data);
}

setInterval(refreshData, 10000);
Enter fullscreen mode Exit fullscreen mode

For this prototype, polling is good enough. It also keeps the architecture easier to debug because there is only one external WebSocket connection: Flask to AISStream.

A production version could use Server-Sent Events or a browser WebSocket, but I did not need that complexity for the first version.

The Most Useful Lesson: Live Data Is Messy

With static data, you can run the same input again and again.

With live AIS data, everything depends on the real world:

  • Is there vessel activity right now?
  • Does the provider have coverage there?
  • Is the bounding box correct?
  • Are the message types correct?
  • Is the WebSocket connected?
  • Are vessels transmitting frequently?
  • Are some reports delayed or filtered?

When nothing appears on the dashboard, it does not automatically mean the code is broken.

That was the biggest lesson from moving from Hormuz to Singapore.

A good live-data application needs to make uncertainty visible.

For this project, that meant showing stream status, last message time, positions stored, active zone, stale thresholds, and error messages directly in the dashboard.

What I Would Improve Next

This project is still a prototype, but there are several directions I would like to explore next.

Add persistent storage

Right now, vessel tracks are stored in memory.

That keeps the app simple, but it also means everything disappears when the process restarts.

A natural next step would be:

  • SQLite for a simple local version
  • PostGIS for serious geospatial querying
  • time-series storage for historical trend analysis

Add historical playback

Live dashboards are useful, but historical playback would make the analyser much more interesting.

For example, I could select a time window and replay traffic movement through the strait.

That would also make testing easier because I could compare changes in the analysis logic against the same historical track data.

Improve maritime context

A slow tanker in open water means something different from a slow vessel near a port or anchorage.

To reduce false signals, the app should understand more context:

  • port areas
  • anchorage zones
  • traffic separation schemes
  • shipping lanes
  • restricted areas
  • typical speed ranges by vessel type

Support multiple regions

At the moment, the app monitors one bounding box at a time.

A more advanced version could support multiple named regions, such as:

  • Singapore Strait
  • Strait of Hormuz
  • Suez Canal approach
  • Malacca Strait
  • English Channel
  • Panama Canal approach

That would make the app more useful for comparing traffic patterns across regions.

Add better reliability metrics

The first version already tracks basic stream status, but I would like to improve reconnect and data-quality visibility.

Useful metrics could include:

  • reconnect count
  • messages per minute
  • positions per minute
  • static messages received
  • provider errors
  • empty periods by region
  • average reporting interval by MMSI

What This Project Taught Me

The main lesson was not about Flask, Leaflet, or WebSockets.

It was this:

Real-time dashboards are only as good as the data pipeline behind them.

The frontend can look great. The backend can be clean. The map can render perfectly.

But if the live data is sparse, delayed, filtered, or missing, the application has to handle that honestly.

That is why I am glad the first version did not work perfectly in Hormuz.

It forced me to design the app around uncertainty instead of pretending that live data is always available and clean.

Final Thoughts

This was a small project, but it touched many practical areas:

  • WebSocket streaming
  • Flask backend design
  • API key handling
  • geospatial bounding boxes
  • Leaflet maps
  • in-memory caching
  • live dashboard polling
  • vessel classification
  • AIS gap detection
  • data availability problems

And that is exactly why I enjoyed building it.

It started as a ship map.

It became a lesson in building software around real-world data.

Full Code

The complete project is available on GitHub:

πŸ‘‰ github.com/dataai01/ship-analyser

The repository includes the Flask app, AISStream WebSocket listener, dashboard code, and local setup notes.

Question for the Community

If you were improving this ship tracking analyser, what would you focus on first?

Would you add persistent storage, improve the data pipeline, build better maritime analytics, enhance the map UI, or add another AIS/data source to make the signals more reliable?

I would especially like to hear from people who have worked with AIS data, maritime analytics, real-time dashboards, or geospatial applications.

Top comments (0)