Usage#
We start by exploring the data-processing pipeline part of DAMAST.
We consider a manufactured dataset of Automatic Identification System (AIS) messages.
The data is generated for 150 boats, where the minimal length of a trajectory is 30 messages, and the maximal length is 1000
!pip install damast
import polars
import damast.domains.maritime.ais.data_generator as generator
data = generator.AISTestData(number_of_trajectories=1000, min_length=25, max_length=300)
Requirement already satisfied: damast in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (0.3.7)
Requirement already satisfied: astropy in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from damast) (6.1.7)
Requirement already satisfied: cloudpickle in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from damast) (3.1.2)
Requirement already satisfied: iso8601 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from damast) (2.1.0)
Requirement already satisfied: jinja2 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from damast) (3.1.6)
Requirement already satisfied: matplotlib in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from damast) (3.10.9)
Requirement already satisfied: networkx in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from damast) (3.4.2)
Requirement already satisfied: numba in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from damast) (0.67.0)
Requirement already satisfied: numpy>=2 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from damast) (2.2.6)
Requirement already satisfied: packaging in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from damast) (26.3)
Requirement already satisfied: pandas in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from damast) (2.3.3)
Requirement already satisfied: polars>=1.36.1 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from damast) (1.44.2)
Requirement already satisfied: polars_h3 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from damast) (0.6.4)
Requirement already satisfied: psutil in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from damast) (7.2.2)
Requirement already satisfied: pyais in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from damast) (3.2.3)
Requirement already satisfied: pyarrow in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from damast) (25.0.1)
Requirement already satisfied: pydantic>=2.0 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from damast) (2.13.5)
Requirement already satisfied: pydot in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from damast) (4.0.1)
Requirement already satisfied: ratarmount>=1.2.1 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from damast) (1.3.0)
Requirement already satisfied: typing_extensions in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from damast) (4.16.0)
Requirement already satisfied: scikit-learn in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from damast) (1.7.2)
Requirement already satisfied: tables in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from damast) (3.10.1)
Requirement already satisfied: tqdm in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from damast) (4.70.1)
Requirement already satisfied: polars-runtime-32==1.44.2 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from polars>=1.36.1->damast) (1.44.2)
Requirement already satisfied: annotated-types>=0.6.0 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from pydantic>=2.0->damast) (0.8.0)
Requirement already satisfied: pydantic-core==2.46.5 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from pydantic>=2.0->damast) (2.46.5)
Requirement already satisfied: typing-inspection>=0.4.2 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from pydantic>=2.0->damast) (0.4.4)
Requirement already satisfied: ratarmountcore~=0.11.0 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from ratarmountcore[7z,bzip2,fat,gzip,rar,xz,zip,zstd]~=0.11.0->ratarmount>=1.2.1->damast) (0.11.1)
Requirement already satisfied: mfusepy~=3.0 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from ratarmount>=1.2.1->damast) (3.1.1)
Requirement already satisfied: rarfile~=4.1 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from ratarmountcore[7z,bzip2,fat,gzip,rar,xz,zip,zstd]~=0.11.0->ratarmount>=1.2.1->damast) (4.5)
Requirement already satisfied: libarchive-c<6.0,~=5.1 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from ratarmountcore[7z,bzip2,fat,gzip,rar,xz,zip,zstd]~=0.11.0->ratarmount>=1.2.1->damast) (5.3)
Requirement already satisfied: py7zr~=1.0 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from ratarmountcore[7z,bzip2,fat,gzip,rar,xz,zip,zstd]~=0.11.0->ratarmount>=1.2.1->damast) (1.1.3)
Requirement already satisfied: rapidgzip~=0.15.0 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from ratarmountcore[7z,bzip2,fat,gzip,rar,xz,zip,zstd]~=0.11.0->ratarmount>=1.2.1->damast) (0.15.2)
Requirement already satisfied: indexed_gzip~=1.7 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from ratarmountcore[7z,bzip2,fat,gzip,rar,xz,zip,zstd]~=0.11.0->ratarmount>=1.2.1->damast) (1.10.3)
Requirement already satisfied: python-xz~=0.4.0 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from ratarmountcore[7z,bzip2,fat,gzip,rar,xz,zip,zstd]~=0.11.0->ratarmount>=1.2.1->damast) (0.4.0)
Requirement already satisfied: indexed_zstd<2.0,>=1.2.2 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from ratarmountcore[7z,bzip2,fat,gzip,rar,xz,zip,zstd]~=0.11.0->ratarmount>=1.2.1->damast) (1.7.1)
Requirement already satisfied: fast_zip_decryption in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from ratarmountcore[7z,bzip2,fat,gzip,rar,xz,zip,zstd]~=0.11.0->ratarmount>=1.2.1->damast) (3.0.0)
Requirement already satisfied: texttable in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from py7zr~=1.0->ratarmountcore[7z,bzip2,fat,gzip,rar,xz,zip,zstd]~=0.11.0->ratarmount>=1.2.1->damast) (1.7.0)
Requirement already satisfied: pycryptodomex>=3.20.0 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from py7zr~=1.0->ratarmountcore[7z,bzip2,fat,gzip,rar,xz,zip,zstd]~=0.11.0->ratarmount>=1.2.1->damast) (3.23.0)
Requirement already satisfied: brotli>=1.2.0 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from py7zr~=1.0->ratarmountcore[7z,bzip2,fat,gzip,rar,xz,zip,zstd]~=0.11.0->ratarmount>=1.2.1->damast) (1.2.0)
Requirement already satisfied: backports.zstd>=1.0.0 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from py7zr~=1.0->ratarmountcore[7z,bzip2,fat,gzip,rar,xz,zip,zstd]~=0.11.0->ratarmount>=1.2.1->damast) (1.7.0)
Requirement already satisfied: pyppmd>=1.3.1 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from py7zr~=1.0->ratarmountcore[7z,bzip2,fat,gzip,rar,xz,zip,zstd]~=0.11.0->ratarmount>=1.2.1->damast) (1.3.1)
Requirement already satisfied: pybcj>=1.0.6 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from py7zr~=1.0->ratarmountcore[7z,bzip2,fat,gzip,rar,xz,zip,zstd]~=0.11.0->ratarmount>=1.2.1->damast) (1.0.8)
Requirement already satisfied: multivolumefile>=0.2.3 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from py7zr~=1.0->ratarmountcore[7z,bzip2,fat,gzip,rar,xz,zip,zstd]~=0.11.0->ratarmount>=1.2.1->damast) (0.2.3)
Requirement already satisfied: inflate64>=1.0.4 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from py7zr~=1.0->ratarmountcore[7z,bzip2,fat,gzip,rar,xz,zip,zstd]~=0.11.0->ratarmount>=1.2.1->damast) (1.0.4)
Requirement already satisfied: pyerfa>=2.0.1.1 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from astropy->damast) (2.0.1.5)
Requirement already satisfied: astropy-iers-data>=0.2024.10.28.0.34.7 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from astropy->damast) (0.2026.9.21.0.56.25)
Requirement already satisfied: PyYAML>=3.13 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from astropy->damast) (6.0.3)
Requirement already satisfied: MarkupSafe>=2.0 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from jinja2->damast) (3.0.3)
Requirement already satisfied: contourpy>=1.0.1 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from matplotlib->damast) (1.3.2)
Requirement already satisfied: cycler>=0.10 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from matplotlib->damast) (0.12.1)
Requirement already satisfied: fonttools>=4.22.0 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from matplotlib->damast) (4.65.0)
Requirement already satisfied: kiwisolver>=1.3.1 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from matplotlib->damast) (1.5.1)
Requirement already satisfied: pillow>=8 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from matplotlib->damast) (12.3.0)
Requirement already satisfied: pyparsing>=3 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from matplotlib->damast) (3.3.3)
Requirement already satisfied: python-dateutil>=2.7 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from matplotlib->damast) (2.9.0.post0)
Requirement already satisfied: six>=1.5 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from python-dateutil>=2.7->matplotlib->damast) (1.17.0)
Requirement already satisfied: llvmlite<0.50,>=0.49.0dev0 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from numba->damast) (0.49.0)
Requirement already satisfied: pytz>=2020.1 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from pandas->damast) (2026.3.post1)
Requirement already satisfied: tzdata>=2022.7 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from pandas->damast) (2026.4)
Requirement already satisfied: attrs in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from pyais->damast) (26.1.0)
Requirement already satisfied: scipy>=1.8.0 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from scikit-learn->damast) (1.15.3)
Requirement already satisfied: joblib>=1.2.0 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from scikit-learn->damast) (1.6.0)
Requirement already satisfied: threadpoolctl>=3.1.0 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from scikit-learn->damast) (3.7.0)
Requirement already satisfied: numexpr>=2.6.2 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from tables->damast) (2.14.1)
Requirement already satisfied: py-cpuinfo in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from tables->damast) (9.0.0)
Requirement already satisfied: blosc2>=2.3.0 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from tables->damast) (4.3.3)
Requirement already satisfied: ndindex in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from blosc2>=2.3.0->tables->damast) (1.10.1)
Requirement already satisfied: msgpack in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from blosc2>=2.3.0->tables->damast) (1.2.2)
Requirement already satisfied: requests in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from blosc2>=2.3.0->tables->damast) (2.34.2)
Requirement already satisfied: charset_normalizer<4,>=2 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from requests->blosc2>=2.3.0->tables->damast) (3.5.1)
Requirement already satisfied: idna<4,>=2.5 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from requests->blosc2>=2.3.0->tables->damast) (3.20)
Requirement already satisfied: urllib3<3,>=1.26 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from requests->blosc2>=2.3.0->tables->damast) (2.8.0)
Requirement already satisfied: certifi>=2023.5.7 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from requests->blosc2>=2.3.0->tables->damast) (2026.7.22)
damast.core.polars_dataframe: POLARS_ENGINE_AFFINITY is already set to 'streaming' - keeping it instead of damast's default 'streaming'
The data is stored in a polars.LazyFrame, and we can inspect the first and last 5 messages in the dataset.
print(data.dataframe)
shape: (163_300, 11)
┌───────────┬─────────────┬────────────┬──────────────┬───┬────────────┬─────┬────────────┬────────┐
│ mmsi ┆ lon ┆ lat ┆ date_time_ut ┆ … ┆ nav_status ┆ rot ┆ message_nr ┆ source │
│ --- ┆ --- ┆ --- ┆ c ┆ ┆ --- ┆ --- ┆ --- ┆ --- │
│ i64 ┆ f64 ┆ f64 ┆ --- ┆ ┆ i64 ┆ f64 ┆ i64 ┆ str │
│ ┆ ┆ ┆ str ┆ ┆ ┆ ┆ ┆ │
╞═══════════╪═════════════╪════════════╪══════════════╪═══╪════════════╪═════╪════════════╪════════╡
│ 342577432 ┆ -118.475718 ┆ -55.843359 ┆ 1990-04-06 ┆ … ┆ 1 ┆ 0.0 ┆ 3 ┆ g │
│ ┆ ┆ ┆ 09:30:39 ┆ ┆ ┆ ┆ ┆ │
│ 316005135 ┆ 107.009228 ┆ -69.567257 ┆ 1988-10-15 ┆ … ┆ 0 ┆ 0.0 ┆ 2 ┆ g │
│ ┆ ┆ ┆ 13:24:51 ┆ ┆ ┆ ┆ ┆ │
│ 623812856 ┆ -65.778847 ┆ 37.67495 ┆ 2004-03-05 ┆ … ┆ 7 ┆ 0.0 ┆ 3 ┆ s │
│ ┆ ┆ ┆ 19:14:22 ┆ ┆ ┆ ┆ ┆ │
│ 466133739 ┆ 35.621701 ┆ -42.332705 ┆ 1987-04-04 ┆ … ┆ 1 ┆ 0.0 ┆ 1 ┆ g │
│ ┆ ┆ ┆ 02:44:46 ┆ ┆ ┆ ┆ ┆ │
│ 270644876 ┆ 39.682355 ┆ 47.933428 ┆ 2008-01-23 ┆ … ┆ 0 ┆ 0.0 ┆ 1 ┆ g │
│ ┆ ┆ ┆ 19:18:54 ┆ ┆ ┆ ┆ ┆ │
│ … ┆ … ┆ … ┆ … ┆ … ┆ … ┆ … ┆ … ┆ … │
│ 821908642 ┆ 61.155881 ┆ -82.394327 ┆ 2021-11-17 ┆ … ┆ 1 ┆ 0.0 ┆ 1 ┆ g │
│ ┆ ┆ ┆ 00:05:09 ┆ ┆ ┆ ┆ ┆ │
│ 401390029 ┆ 1.910573 ┆ 45.148034 ┆ 1992-09-29 ┆ … ┆ 7 ┆ 0.0 ┆ 2 ┆ s │
│ ┆ ┆ ┆ 02:07:10 ┆ ┆ ┆ ┆ ┆ │
│ 217833214 ┆ 56.332987 ┆ -21.998486 ┆ 2003-11-29 ┆ … ┆ 0 ┆ 0.0 ┆ 2 ┆ s │
│ ┆ ┆ ┆ 02:47:39 ┆ ┆ ┆ ┆ ┆ │
│ 215812592 ┆ 12.208488 ┆ 87.032085 ┆ 1970-09-18 ┆ … ┆ 1 ┆ 0.0 ┆ 1 ┆ g │
│ ┆ ┆ ┆ 04:56:03 ┆ ┆ ┆ ┆ ┆ │
│ 230516396 ┆ -22.317257 ┆ 54.090352 ┆ 1990-08-19 ┆ … ┆ 7 ┆ 0.0 ┆ 2 ┆ g │
│ ┆ ┆ ┆ 13:30:22 ┆ ┆ ┆ ┆ ┆ │
└───────────┴─────────────┴────────────┴──────────────┴───┴────────────┴─────┴────────────┴────────┘
The dataset consists of 11 columns, which we will go through in detail.
Data-specification#
The Maritime Mobile Service Identity (MMSI) used to identify a ship. It should be a 9 digit number whose first integer should be between 2 and 7. The data we have generated should contain some invalid numbers. Let us inspect these.
from damast.domains.maritime.data_specification import MMSI
df = data.dataframe
invalid_mmsis = df.filter((polars.col('mmsi') < MMSI.min_value) | (polars.col('mmsi') > MMSI.max_value))
invalid_mmsis
shape: (11_575, 11)
| mmsi | lon | lat | date_time_utc | sog | cog | true_heading | nav_status | rot | message_nr | source |
|---|---|---|---|---|---|---|---|---|---|---|
| i64 | f64 | f64 | str | f64 | f64 | f64 | i64 | f64 | i64 | str |
| 812680204 | 31.464409 | -79.804367 | null | 17.714986 | 3.719913 | 3.729864 | 1 | 0.0 | 2 | "g" |
| 801929911 | -124.83129 | 20.72172 | "2005-04-20 21:38:33" | -3.037462 | -1.355481 | -1.320768 | 7 | 0.0 | 3 | "s" |
| 813390024 | 91.021237 | -67.424578 | "1995-11-11 13:08:18" | -8.35026 | -2.019641 | -2.005758 | 7 | 0.0 | 1 | "s" |
| 816412230 | -65.907331 | 55.772463 | "1996-10-25 21:41:51" | -11.28018 | -1.956781 | -1.885115 | 0 | 0.0 | 1 | "s" |
| 825447227 | 160.1528 | -55.536598 | null | -13.825886 | -4.559177 | -4.487194 | 1 | 0.0 | 3 | "g" |
| … | … | … | … | … | … | … | … | … | … | … |
| 839198766 | -37.885781 | 4.879087 | "1985-03-19 19:04:14" | -0.218294 | -2.796276 | -2.770298 | 0 | 0.0 | 2 | "s" |
| 823979320 | 124.953272 | 73.960683 | null | -14.877293 | 4.550149 | 4.620098 | 0 | 0.0 | 3 | "s" |
| 811673358 | 172.890888 | -54.384613 | "1990-10-17 16:32:32" | 7.846214 | -5.508123 | -5.473253 | 1 | 0.0 | 2 | "g" |
| 828441839 | 144.717322 | 45.823718 | "1988-08-06 05:07:12" | -3.956336 | 2.928105 | 2.981869 | 1 | 0.0 | 1 | "s" |
| 821908642 | 61.155881 | -82.394327 | "2021-11-17 00:05:09" | -9.572851 | 4.543668 | 4.572558 | 1 | 0.0 | 1 | "g" |
Before sending this data to a machine learning algorithm, one would have to filter out invalid data.
We can do this by creating a damast.core.DataSpecification describing what valid output we would like in our data-frame.
from damast.core import DataSpecification, MinMax
mmsi_spec = DataSpecification(name="mmsi", description="Maritime Mobile Service Identity", representation_type=int,
value_range=MinMax(MMSI.min_value, MMSI.max_value))
We have here described what data this column is supposed to describe, how the data is represented in Python, and its minimum and maximum range.
Next, we create a damast.core.MetaData object that we can apply to the dataframe.
from damast.core import MetaData, ValidationMode
metadata = MetaData([mmsi_spec])
# metadata only describes 'mmsi', so restrict the dataframe to that column - MetaData.apply()
# requires every column present to have a matching DataSpecification.
metadata.apply(df.lazy().select(mmsi_spec.name), ValidationMode.UPDATE_DATA)
naive plan: (run LazyFrame.explain(optimized=True) to see the optimized plan)
FILTER ((col("mmsi") >= 200000000) & (col("mmsi") <= 799999999)) | col("mmsi").is_null()FROM SELECT [col("mmsi")] DF ["mmsi", "lon", "lat", "date_time_utc", ...]; PROJECT */11 COLUMNS
Of course, we do not want to do this process manually per row. Therefore, we can create a DataSpecification per row, and let the damast.core.AnnotatedDataFrame handle the validation of the data. We can choose between three ways of handling the input data with metadata, we can either use:
ValidationMode.READONLY: Reads in the data, checks it against the meta-data and throws an error if the data does not adhere to the data-specification.ValidationMode.UPDATE_METADATA: Update the metadata based on the input in the annotated data-frame. This might change the representation type, column name and valid rages of the data.ValidationMode.UPDATE_DATA: Update data so that it adheres to the meta-data.
from damast.core.dataframe import AnnotatedDataFrame
from damast.core.metadata import DataCategory
dataspec = {
"annotations": {"comment": "This is a autogenerated test data set"},
"columns": [
{"name": "mmsi", "is_optional": False, "category": DataCategory.STATIC,
"value_range":{"MinMax": {"min": MMSI.min_value, "max": MMSI.max_value}}},
{"name": "lon", "is_optional": False, "unit": "deg", "category": DataCategory.DYNAMIC},
{"name": "lat", "is_optional": False, "unit": "deg", "category": DataCategory.DYNAMIC},
{"name": "date_time_utc", "is_optional": False, "category": DataCategory.DYNAMIC},
{"name": "sog", "is_optional": False, "category": DataCategory.DYNAMIC},
{"name": "cog", "is_optional": False, "category": DataCategory.DYNAMIC},
{"name": "true_heading", "is_optional": False, "category": DataCategory.DYNAMIC},
{"name": "nav_status", "is_optional": False, "category": DataCategory.DYNAMIC},
{"name": "rot", "is_optional": False, "category": DataCategory.DYNAMIC},
{"name": "message_nr", "is_optional": False, "category": DataCategory.DYNAMIC},
{"name": "source", "is_optional": False, "category": DataCategory.DYNAMIC},
]
}
metadata = MetaData.from_dict(dataspec)
data = generator.AISTestData(number_of_trajectories=1000, min_length=25, max_length=300)
adf = AnnotatedDataFrame(data.dataframe, metadata, validation_mode=ValidationMode.UPDATE_DATA)
adf
naive plan: (run LazyFrame.explain(optimized=True) to see the optimized plan)
FILTER ((col("mmsi") >= 200000000) & (col("mmsi") <= 799999999)) | col("mmsi").is_null()FROM DF ["mmsi", "lon", "lat", "date_time_utc", ...]; PROJECT */11 COLUMNS
Data-processing#
Say we want to repeat this process on any data-set we read in. Then, we should create a damast.core.dataprocessing.DataProcessingPipeline.
A pipeline consists of pipeline-elements, that is a set of transformations on the original dataset.
We start by creating a Pipeline-element that drops all rows missing an "mmsi" entry.
from damast.core.dataprocessing import DataProcessingPipeline
from damast.data_handling.transformers.filters import DropMissingOrNan
pipeline = DataProcessingPipeline(name="Remove missing MMSI columns",
base_dir="./output_dir",
inplace_transformation=True)
pipeline.add(name="Remove MMSI column",
transformer=DropMissingOrNan(),
name_mappings={"x": "mmsi"})
transformed_adf = pipeline.transform(adf)
transformed_adf
Step : 0%| | 0/2 [00:00<?, ?it/s]
Step : 100%|██████████| 2/2 [00:00<00:00, 257.12it/s]
naive plan: (run LazyFrame.explain(optimized=True) to see the optimized plan)
FILTER (col("mmsi").is_nan() !=v true)FROM FILTER col("mmsi").is_not_null() FROM FILTER ((col("mmsi") >= 200000000) & (col("mmsi") <= 799999999)) | col("mmsi").is_null() FROM DF ["mmsi", "lon", "lat", "date_time_utc", ...]; PROJECT */11 COLUMNS
transformed_adf.collect()
shape: (150_114, 11)
| mmsi | lon | lat | date_time_utc | sog | cog | true_heading | nav_status | rot | message_nr | source |
|---|---|---|---|---|---|---|---|---|---|---|
| i64 | f64 | f64 | str | f64 | f64 | f64 | i64 | f64 | i64 | str |
| 229846815 | 32.37902 | -86.120109 | null | 2.141916 | 0.79513 | 0.852846 | 0 | 0.0 | 3 | "g" |
| 644056272 | 42.879138 | -80.953023 | "2011-09-08 20:08:13" | -13.525436 | -1.818847 | -1.789009 | 7 | 0.0 | 3 | "g" |
| 534483987 | -96.189698 | -11.408602 | "2004-07-20 05:49:11" | 2.544822 | -0.463618 | -0.460356 | 0 | 0.0 | 3 | "g" |
| 772729454 | -19.663006 | -78.382675 | "2012-05-26 04:56:41" | 2.195884 | 0.421838 | 0.466923 | 1 | 0.0 | 3 | "s" |
| 661822124 | 134.779577 | -85.348376 | "1982-10-21 05:18:05" | 3.935483 | -1.240962 | -1.168053 | 7 | 0.0 | 2 | "s" |
| … | … | … | … | … | … | … | … | … | … | … |
| 783961163 | 26.079218 | -80.234605 | "2007-07-02 04:52:57" | -0.875363 | -6.050103 | -6.035012 | 1 | 0.0 | 1 | "g" |
| 376774737 | 127.413477 | 17.682185 | "1974-02-18 21:40:58" | -8.362353 | 0.989167 | 1.068342 | 1 | 0.0 | 3 | "g" |
| 572658161 | -170.120308 | -35.520725 | "2004-05-15 16:24:19" | 4.961024 | -0.509908 | -0.42457 | 0 | 0.0 | 1 | "s" |
| 527497343 | 136.63124 | -60.652272 | "2004-05-21 19:01:21" | 1.302086 | -3.801622 | -3.767406 | 0 | 0.0 | 1 | "s" |
| 662647768 | 109.345979 | 32.322015 | "1981-11-15 20:00:04" | 34.814436 | 1.776306 | 1.868393 | 7 | 0.0 | 3 | "s" |
Data Pipelines with multiple input sources#
When multiple input sources exist and should be merged, a join operator (transformer) can be designed. The join can, but must not necessarily involve two pipelines as illustrated in the following. The pipeline is named using the ‘data_source’ argument, but this is optiona
import damast
from damast.core.transformations import PipelineElement
class JoinByTime(PipelineElement):
@damast.core.describe("Join data by timestamp")
@damast.core.input({
"timestamp": {},
"lon": {},
"lat": {},
})
@damast.core.input({
"timestamp": {},
"lat": {},
"lon": {}
}, label='other'
)
# 'other' also carries columns beyond timestamp/lat/lon (event_type, and the cycle-transform
# derived latitude_x/y, longitude_x/y computed by events_pipeline below) - declare them here
# via the '{{other:...}}' template so MetaData.apply() knows about every column the join
# brings in, not just the three used to steer the join itself.
@damast.core.output({
"{{other:event_type}}": {"representation_type": str},
"{{other:lat}}": {},
"{{other:lon}}": {},
"{{other:latitude_x}}": {},
"{{other:latitude_y}}": {},
"{{other:longitude_x}}": {},
"{{other:longitude_y}}": {},
})
def transform(self, df: AnnotatedDataFrame, other: AnnotatedDataFrame) -> AnnotatedDataFrame:
other_timestamp = self.get_name('timestamp', datasource='other')
df_timestamp = self.get_name('timestamp')
df.lazyframe = df.join(other.lazyframe, left_on=df_timestamp, right_on=other_timestamp)
# Carry over 'other's column descriptions (event_type, lat/lon and their cycle-transform
# derivatives) into the joined result - other_timestamp is consumed by the join above.
df._metadata = df._metadata.merge(other._metadata).drop(other_timestamp)
return df
event_time = data.dataframe.drop_nulls().select(polars.col('date_time_utc')).item(0,0)
events_dataframe = polars.from_dict({'latitude': [40.0, 40.1], 'longitude': [10.0,10.2], 'timestamp': [event_time, event_time], 'event_type': ["accident", "accident"]})
events_metadata = AnnotatedDataFrame.infer_annotation(events_dataframe)
events_adf = AnnotatedDataFrame(events_dataframe, metadata=events_metadata)
Extract str and categorical column metadata: 0%| | 0/4 [00:00<?, ?column/s]
Extract str and categorical column metadata: 100%|██████████| 4/4 [00:00<00:00, 2204.63column/s]
Extract numeric column metadata: 0%| | 0/2 [00:00<?, ?column/s]
Extract numeric column metadata: 100%|██████████| 2/2 [00:00<00:00, 22671.91column/s]
from damast.data_handling.transformers.cycle_transformer import CycleTransformer
events_pipeline = DataProcessingPipeline(name="events",
base_dir="./output_dir") \
.add("lat_cycle_transform", CycleTransformer(n=180), name_mappings={'x': 'latitude'}) \
.add("lon_cycle_transform", CycleTransformer(n=90), name_mappings={'x': 'longitude'}) \
pipeline = DataProcessingPipeline(name="ais_events_merge",
base_dir="./output_dir") \
.add("lat_cycle_transform", CycleTransformer(n=180), name_mappings={'x': 'lat'}) \
.add("lon_cycle_transform", CycleTransformer(n=90), name_mappings={'x': 'lon'}) \
.join("events", data_source=events_pipeline, operator=JoinByTime(),
name_mappings = {
'df': {
"timestamp": "date_time_utc",
},
'other': {
"timestamp": "timestamp",
"lon": "longitude",
"lat": "latitude"
}
},
)
To run the pipeline, all required datasource aka inputs need to be provided as arguments. While the default input is ‘df’, the datasource for the ‘join’ operator requires to be provided via the keyword of the same name, here ‘events’
joined_adf = pipeline.transform(df=adf, events=events_adf)
Step : 0%| | 0/7 [00:00<?, ?it/s]
WARNING:damast.core.metadata:'comment'
Step : 100%|██████████| 7/7 [00:00<00:00, 82.09it/s]
joined_adf.head(10)
naive plan: (run LazyFrame.explain(optimized=True) to see the optimized plan)
SLICE[offset: 0, len: 10] INNER JOIN: LEFT PLAN ON: [col("date_time_utc")] WITH_COLUMNS: [((((col("lon") * 2.0) * 3.141593)).python_udf() / 90.0).alias("lon_x"), ((((col("lon") * 2.0) * 3.141593)).python_udf() / 90.0).alias("lon_y")] WITH_COLUMNS: [((((col("lat") * 2.0) * 3.141593)).python_udf() / 180.0).alias("lat_x"), ((((col("lat") * 2.0) * 3.141593)).python_udf() / 180.0).alias("lat_y")] FILTER (col("mmsi").is_nan() !=v true) FROM FILTER col("mmsi").is_not_null() FROM FILTER ((col("mmsi") >= 200000000) & (col("mmsi") <= 799999999)) | col("mmsi").is_null() FROM DF ["mmsi", "lon", "lat", "date_time_utc", ...]; PROJECT */11 COLUMNS RIGHT PLAN ON: [col("timestamp")] WITH_COLUMNS: [((((col("longitude") * 2.0) * 3.141593)).python_udf() / 90.0).alias("longitude_x"), ((((col("longitude") * 2.0) * 3.141593)).python_udf() / 90.0).alias("longitude_y")] WITH_COLUMNS: [((((col("latitude") * 2.0) * 3.141593)).python_udf() / 180.0).alias("latitude_x"), ((((col("latitude") * 2.0) * 3.141593)).python_udf() / 180.0).alias("latitude_y")] DF ["latitude", "longitude", "timestamp", "event_type"]; PROJECT */4 COLUMNS END INNER JOIN
joined_adf.head(10).collect()
shape: (2, 22)
| mmsi | lon | lat | date_time_utc | sog | cog | true_heading | nav_status | rot | message_nr | source | lat_x | lat_y | lon_x | lon_y | latitude | longitude | event_type | latitude_x | latitude_y | longitude_x | longitude_y |
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| i64 | f64 | f64 | str | f64 | f64 | f64 | i64 | f64 | i64 | str | f64 | f64 | f64 | f64 | f64 | f64 | str | f64 | f64 | f64 | f64 |
| 644056272 | 42.879138 | -80.953023 | "2011-09-08 20:08:13" | -13.525436 | -1.818847 | -1.789009 | 7 | 0.0 | 3 | "g" | 0.001616 | 0.005315 | -0.00765 | 0.008058 | 40.0 | 10.0 | "accident" | -5.4429e-17 | 0.005556 | -2.7214e-17 | 0.011111 |
| 644056272 | 42.879138 | -80.953023 | "2011-09-08 20:08:13" | -13.525436 | -1.818847 | -1.789009 | 7 | 0.0 | 3 | "g" | 0.001616 | 0.005315 | -0.00765 | 0.008058 | 40.1 | 10.2 | "accident" | 0.003265 | 0.004495 | 0.010567 | 0.003434 |
Plugin transformers#
Not every transformer needs to be part of the damast package - or even part of an installed
Python package at all. Damast supports two ways to use PipelineElement classes that live
outside of damast itself, so that pipelines using them stay reproducible when shared with
others. Regardless of which of the two a transformer comes from, it is always resolvable the
same way in code - from damast.plugins.<package> import MyTransformer, where <package> is
the plugin’s top-level package - without needing to import (or even know) the underlying module
that defines it. Scoping by package means two plugins can provide a transformer of the same name
without clashing.
Local, ad-hoc transformers via DAMAST_PLUGIN_PATH#
For a quick transformer that does not warrant its own package, put it in a plain *.py file and
register the directory containing it under a package name via the DAMAST_PLUGIN_PATH
environment variable, as name=path (multiple entries can be separated with os.pathsep, just
like PATH). Every top-level file found there is imported once as name.<file> - so files may
use relative imports among each other - and any PipelineElement subclasses they define become
resolvable exactly like classes from an installed package. Choose a distinctive name: it is
recorded in saved pipelines, and a name that is already importable is rejected.
import os
import tempfile
from pathlib import Path
plugin_dir = Path(tempfile.mkdtemp())
(plugin_dir / "my_transformers.py").write_text("""
import polars
from damast.core.dataframe import AnnotatedDataFrame
from damast.core.decorators import describe, input, output
from damast.core.transformations import PipelineElement
class Doubler(PipelineElement):
@describe("Doubles a column")
@input({"x": {}})
@output({"{{x}}_doubled": {}})
def transform(self, df: AnnotatedDataFrame) -> AnnotatedDataFrame:
feature = self.get_name("x")
df.lazyframe = df.lazyframe.with_columns(
(polars.col(feature) * 2).alias(f"{feature}_doubled")
)
return df
""")
os.environ["DAMAST_PLUGIN_PATH"] = f"my_plugins={plugin_dir}"
from damast.core.transformations import PipelineElement # noqa: E402
# directories are only scanned once per process, so force a rescan after
# changing DAMAST_PLUGIN_PATH or adding/editing files
PipelineElement.reload_plugins()
PipelineElement.list_plugins()
{'my_plugins.Doubler': 'my_plugins.my_transformers:Doubler'}
Doubler is now resolvable via damast.plugins.my_plugins, and can be used in a pipeline like any other
transformer. damast.plugins resolves names lazily on first access - nothing beyond the
requested class is imported, and it works no matter which of the two plugin sources defines
it. Saving the pipeline records where the transformer came from, under requires, so a pipeline
loaded elsewhere fails with an actionable error (pointing at DAMAST_PLUGIN_PATH) instead of a
bare import error if the local file is missing.
from damast.core.dataprocessing import DataProcessingPipeline
from damast.plugins.my_plugins import (
Doubler, # resolved from the file we just wrote to `plugin_dir`
)
plugin_pipeline = DataProcessingPipeline(name="plugin-example", base_dir="./output_dir") \
.add("double_mmsi", Doubler(), name_mappings={"x": "mmsi"})
pipeline_path = plugin_pipeline.save("./output_dir")
print(pipeline_path.read_text())
base_dir: output_dir
description: ''
meta:
damast_version: 0.3.7
name: plugin-example
processing_graph:
edges:
- from: 940fe937-d126-4aec-99c1-79d14efe8aca
slot: df
to: 1e5f43ce-5a77-4263-91bb-db4d2b66a61a
nodes:
- name: df
transformer:
class_name: DataSource
module_name: damast.core.processing_graph
name_mappings:
df: {}
parameters: {}
requires:
distribution: damast
version: 0.3.7
uuid: 940fe937-d126-4aec-99c1-79d14efe8aca
- name: double_mmsi
transformer:
class_name: Doubler
module_name: my_plugins.my_transformers
name_mappings:
df:
x: mmsi
parameters: {}
requires:
hint: local
package: my_plugins
path: /tmp/tmpj_tp89xl
uuid: 1e5f43ce-5a77-4263-91bb-db4d2b66a61a
Installable plugin packages (entry-points)#
For a transformer that should be pip-installable and reused across projects, a package can
advertise its PipelineElement subclasses via the damast.transformers entry-point group in its
own pyproject.toml:
[project.entry-points."damast.transformers"]
MyTransformer = "acme_pkg.transformers:MyTransformer"
Instead of one entry per class, an entry may also name a whole module, e.g.
acme_pkg = "acme_pkg.transformers": every PipelineElement defined in that module - or, for a
package, in its top-level submodules - is then registered.
Once installed, it is discovered the same way as a local plugin - PipelineElement.list_plugins()
merges both sources - and pipelines saved with it additionally record the distribution name and
version under requires, so a version mismatch on reload is logged as a warning rather than
silently changing behavior. Run damast plugins from the command line to list everything that is
currently discoverable, from either source, without writing any Python. In code, it resolves the
same way as a local plugin too - from damast.plugins.acme_pkg import MyTransformer.
Artifacts#
Not every pipeline step produces its result as dataframe columns - some write a file as a
side effect instead, e.g. exporting to another format, or saving a plot. Such a file is
called an artifact. damast.core.artifacts is a decorator - used alongside describe/
input/output on transform - that declares which artifact(s) a step is expected to
produce, so a bug that silently fails to write the file is caught immediately instead of
surfacing later as a missing file somewhere downstream.
artifacts takes a dict mapping a descriptive name to either an absolute path, or a path
pattern interpreted relative to the pipeline’s base_dir and matched with
pathlib.Path.glob - e.g. {"csv_export": "*.csv"}. After transform runs, every pattern
must match at least one file, or a RuntimeError is raised naming the missing artifact.
Because a relative pattern resolves against parent_pipeline.base_dir, the transformer needs
to actually be part of a DataProcessingPipeline (added via .add(), or reconstructed from a
saved pipeline) for this to work - it cannot be used standalone.
mini_df = polars.DataFrame({"mmsi": [235096758, 982351234, 111235123]})
mini_adf = damast.core.AnnotatedDataFrame(
mini_df, damast.core.AnnotatedDataFrame.infer_annotation(mini_df)
)
class ExportCSV(PipelineElement):
"""
Writes the dataframe to a CSV file as a side effect - the file *is* this step's actual
output; the dataframe itself passes through unchanged.
"""
def __init__(self, filename: str):
self.filename = filename
@damast.core.input({})
@damast.core.artifacts({"csv_export": "*.csv"})
def transform(self, df: damast.core.AnnotatedDataFrame) -> damast.core.AnnotatedDataFrame:
self.parent_pipeline.base_dir.mkdir(parents=True, exist_ok=True)
df.lazyframe.collect().write_csv(self.parent_pipeline.base_dir / self.filename)
return df
artifact_pipeline = DataProcessingPipeline(name="export-example", base_dir="./output_dir") \
.add("export mmsi", ExportCSV("mmsi-export.csv"))
artifact_pipeline.transform(df=mini_adf)
sorted(Path("./output_dir").glob("*.csv"))
Extract str and categorical column metadata: 0%| | 0/1 [00:00<?, ?column/s]
Extract str and categorical column metadata: 100%|██████████| 1/1 [00:00<00:00, 9300.01column/s]
Extract numeric column metadata: 0%| | 0/1 [00:00<?, ?column/s]
Extract numeric column metadata: 100%|██████████| 1/1 [00:00<00:00, 9731.56column/s]
Step : 0%| | 0/2 [00:00<?, ?it/s]
Step : 100%|██████████| 2/2 [00:00<00:00, 1016.80it/s]
[PosixPath('output_dir/mmsi-export.csv')]
A step that claims an artifact but does not actually produce it fails loudly, right where the mistake was made:
class BrokenExport(PipelineElement):
"""Forgets to write the file it claims to produce - @artifacts should catch this."""
@damast.core.input({})
@damast.core.artifacts({"csv_export": "*.does-not-exist"})
def transform(self, df: damast.core.AnnotatedDataFrame) -> damast.core.AnnotatedDataFrame:
return df
broken_pipeline = DataProcessingPipeline(name="broken-export-example", base_dir="./output_dir") \
.add("export mmsi", BrokenExport())
try:
broken_pipeline.transform(df=mini_adf)
except RuntimeError as e:
print(e)
Step : 0%| | 0/2 [00:00<?, ?it/s]
Step : 50%|█████ | 1/2 [00:00<00:00, 531.66it/s]
Step #2 in pipeline (name=export mmsi BrokenExport (uuid=784e6750-83e3-49d5-801d-f5e3cd04e03f)) failed: name_mappings: {'df': {}}
Traceback (most recent call last):
File "/home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages/damast/core/decorators.py", line 280, in check
required_artifact_specs.validate(
File "/home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages/damast/core/metadata.py", line 102, in validate
raise RuntimeError(
RuntimeError: ArtifactSpecification.validate: no artifact matching *.does-not-exist found in 'output_dir'
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages/damast/core/dataprocessing.py", line 697, in _run
node.result = self.processing_graph.execute(node)
File "/home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages/damast/core/processing_graph.py", line 359, in execute
return node.transformer.fit_transform(**kwargs)
File "/home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages/damast/core/transformations.py", line 536, in fit_transform
return self.transform(df=df)
File "/home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages/damast/core/decorators.py", line 143, in check
return func(*args, **kwargs)
File "/home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages/damast/core/decorators.py", line 284, in check
raise RuntimeError(
RuntimeError: artifacts: <function BrokenExport.transform at 0x7f75c02b7520> is expected to generate an artifact: ArtifactSpecification.validate: no artifact matching *.does-not-exist found in 'output_dir' Pipeline element ran as part of pipeline: 'broken-export-example'
input 'df':
shape: (1, 1)
┌───────────┐
│ mmsi │
│ --- │
│ i64 │
╞═══════════╡
│ 235096758 │
└───────────┘