Simple DASH AIS graphical interface

Dockerfile

FROM ubuntu:22.04
ENV DEB_PYTHON_INSTALL_LAYOUT=deb_system

RUN apt-get update && \
    apt-get install -q -y python3-pip

RUN python3 -m pip install --upgrade pip setuptools

RUN python3 -m pip install vaex dash numpy

WORKDIR /root/script
COPY january_ais.hdf5 .
COPY mmsi_vesseltype.hdf5 .
COPY main.py .
COPY web_application.py .
EXPOSE 8888/tcp
ENTRYPOINT ["python3", "main.py", "--file=/root/script/january_ais.hdf5", "--mmsi-file=/root/script/mmsi_vesseltype.hdf5", "--port=8888"]

main.py

# Copyright (C) 2023 Jørgen S. Dokken
#
# SPDX-License-Identifier:     BSD 3-Clause

import argparse
import pathlib

import vaex
from web_application import AISApp

if __name__ == "__main__":
    parser = argparse.ArgumentParser(formatter_class=argparse.ArgumentDefaultsHelpFormatter)
    parser.add_argument("--file", dest="csv_file", required=True, type=str, help="Input csv file")
    parser.add_argument("--port", dest="port", required=True, type=int, help="Port for dash server")
    parser.add_argument("--mmsi-file", dest="mmsi_file", required=True, type=str, help="File identifying vessels")
    parser.add_argument("--delimiter", dest="delimiter", required=False, type=str, default=";",
                        help="Delimiter used in csv file")
    args = parser.parse_args()

    # We start by opening the DataFrame used in the application
    path = pathlib.Path(args.csv_file)
    mmsi_path = pathlib.Path(args.mmsi_file)
    if path.suffix == ".csv":
        vaex_df = vaex.read_csv(path, delimiter=args.delimiter)
    elif path.suffix == ".hdf5":
        vaex_df = vaex.open(path)
    else:
        raise ValueError("Unknown file format for input AIS data")

    if mmsi_path.suffix == ".csv":
        mmsi_df = vaex.read_csv(mmsi_path, delimiter=args.delimiter)
    elif mmsi_path.suffix == ".hdf5":
        mmsi_df = vaex.open(mmsi_path)
    else:
        raise ValueError("Unknown file format for input MMSI vessel types")

    # Only consider fishing boats
    unique_boats = mmsi_df[mmsi_df["vessel_type"] == "fishing"]["MMSI"].unique()
    boat_df = vaex_df[vaex_df["mmsi"].isin(unique_boats)]

    # Create and launch web-application
    web_app = AISApp(boat_df, port=args.port)
    web_app.generate_layout()
    web_app.run_server()

web_application.py

# Copyright (C) 2023 Jørgen S. Dokken
#
# SPDX-License-Identifier:     BSD 3-Clause

import dash
import vaex
from typing import Dict, List, Any
import webbrowser
import plotly.express as px
import plotly.graph_objs as go
import numpy as np
import numpy.typing as npt


def sort_by(df: vaex.DataFrame, key: str) -> vaex.DataFrame:
    """Sort input dataframe by entries in given column"""
    return df.sort(df[key])


def transform_value(value):
    return 10 ** value


def plot_boat_trajectory(data_df: vaex.DataFrame) -> go.Figure:
    """
    Extract (lat, long) coordinates from dataframe and group them by MMSI.
    NOTE: Assumes that the input data is sorted by in time
    """
    input = {"lat": data_df["lat"].evaluate(),
             "lon": data_df["lon"].evaluate(),
             "mmsi": data_df["mmsi"].evaluate()}
    fig = px.line_mapbox(input,
                         lat="lat", lon="lon",
                         color="mmsi")

    fig2 = px.density_mapbox(input, lat='lat', lon='lon', z='mmsi', radius=5)

    fig.add_trace(fig2.data[0])
    fig.update_coloraxes(showscale=False)

    fig.update_layout(mapbox_style="open-street-map", mapbox_zoom=4)

    return fig


def count_number_of_messages(data: vaex.DataFrame) -> vaex.DataFrame:
    """Given a set of AIS messages, accumulate number of messages per mmsi identifier"""
    return data.groupby("mmsi", agg="count")


class WebApplication():
    """Base class for Dash web applications
    """

    _app: dash.Dash
    _data: Dict[str, vaex.DataFrame]
    _layout: List[Any]
    _port: str
    _server: str

    def __init__(self, df: Dict[str, vaex.DataFrame], header: str,
                 server: str = "0.0.0.0", port: str = "8888"):
        self._data = df
        self._app = dash.Dash(__name__)
        self._port = port
        self._server = server
        self._layout = []
        self.set_header(header)

    def set_header(self, header: str):
        self._layout.append(dash.html.Div(
            children=[dash.html.H1(children=header)]))

    def run_server(self, debug: bool = False):
        self._app.layout = dash.html.Div(self._layout)
        self._app.run(debug=debug, host=self._server, port=self._port)

    def open_browser(self):
        webbrowser.open(f"http://{self._host}:{self._port}")

    @property
    def app(self) -> dash.Dash:
        return self._app


class AISApp(WebApplication):

    _messages_per_mmsi: vaex.DataFrame

    def __init__(self, ais: vaex.DataFrame, port: str = "8888"):

        super().__init__({"ais": ais}, "Map of MMSI messages", port=port)
        self._messages_per_mmsi = count_number_of_messages(self._data["ais"])
        self.callbacks()

    def filter_boats_by_mmsi(self, MMSI: List[int]) -> vaex.DataFrame:
        """Filter boats by MMSI identifier"""
        return self._data["ais"][self._data["ais"]["mmsi"].isin(MMSI)]

    def boats_with_min_messages(self, min_messages: float) -> npt.NDArray[np.int32]:
        """Get the boats (MMSI-identifiers) that has more than `min_messages`
        messages in database

        Args:
            min_messages (np.float64): Minimal number of messages in database

        Returns:
            _type_: MMSI identifiers
        """

        messages = int(transform_value(min_messages))
        min_message_boats = self._messages_per_mmsi[self._messages_per_mmsi["count"]
                                                    >= messages]["mmsi"]
        return min_message_boats.unique()

    def update_map(self, boat_ids: List[np.int32]) -> go.Figure:
        """
        Plot input boats in map (sorted by time-stamp)

        Args:
            boat_ids (List[np.int32]): List of boat identifiers

        Returns:
            go.Figure: Figure with heatmap and group per MMSI
        """
        if boat_ids is None:
            boat_ids = []
        data = self.filter_boats_by_mmsi(boat_ids)
        sorted_data = sort_by(data, "date_time_utc")
        return plot_boat_trajectory(sorted_data)

    def callbacks(self):
        """Define input/output of the different dash applications
        """

        self._app.callback(dash.dependencies.Output("dropdown", "options"),
                           dash.dependencies.Input("num_messages", "value"))(self.boats_with_min_messages)
        self._app.callback(dash.dependencies.Output("plot_map", "figure"),
                           dash.dependencies.Input("dropdown", "value"), prevent_initial_call=True)(self.update_map)

    def generate_layout(self):
        """Generate dashboard layout
        """

        # Generate slider for filtering min number of messages per MMSi
        min_messages = np.log(
            self._messages_per_mmsi["count"].min())/np.log(10)
        max_messages = np.log(
            self._messages_per_mmsi["count"].max())/np.log(10)
        markers = np.linspace(min_messages, max_messages, 10, endpoint=True)
        min_filter = [dash.html.H2(children="Minimum number of MMSI messages"),
                      dash.dcc.Slider(id='num_messages',
                                      min=min_messages, max=max_messages,
                                      value=max_messages//2, marks={i: f"{int(10**i)}" for i in
                                                                    markers}),
                      dash.html.Div(children=["Boat identifiers (MMSI)",
                                              dash.dcc.Dropdown(id="dropdown", multi=True)]),
                      dash.html.Div(id='num_messages-output-container', style={'margin-top': 20})]
        self._layout += min_filter

        # Add map
        map = [dash.dcc.Graph(id="plot_map", figure=self.update_map([]))]

        self._layout += map

Run command: docker run -t -p 8888:8888 --rm test_ais