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.3)
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: 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.66.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.43.2)
Requirement already satisfied: polars_h3 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from damast) (0.6.1)
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.0)
Requirement already satisfied: pyarrow in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from damast) (25.0.0)
Requirement already satisfied: pydantic>=2.0 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from damast) (2.13.4)
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: 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.0)
Requirement already satisfied: polars-runtime-32==1.43.2 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from polars>=1.36.1->damast) (1.43.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.4 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from pydantic>=2.0->damast) (2.46.4)
Requirement already satisfied: typing-extensions>=4.14.1 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from pydantic>=2.0->damast) (4.16.0)
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.2)
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: 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: 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: 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: 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: 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: 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: 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.6.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.8.3.0.53.6)
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: 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.63.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.0)
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.2)
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.49,>=0.48.0dev0 in /home/runner/work/damast/damast/.tox/build_docs/lib/python3.10/site-packages (from numba->damast) (0.48.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.3)
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.5.3)
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.6.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.1)
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.4.9)
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.18)
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.7.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: (159_415, 11)
┌───────────┬─────────────┬────────────┬──────────────┬───┬────────────┬─────┬────────────┬────────┐
│ mmsi ┆ lon ┆ lat ┆ date_time_ut ┆ … ┆ nav_status ┆ rot ┆ message_nr ┆ source │
│ --- ┆ --- ┆ --- ┆ c ┆ ┆ --- ┆ --- ┆ --- ┆ --- │
│ i64 ┆ f64 ┆ f64 ┆ --- ┆ ┆ i64 ┆ f64 ┆ i64 ┆ str │
│ ┆ ┆ ┆ str ┆ ┆ ┆ ┆ ┆ │
╞═══════════╪═════════════╪════════════╪══════════════╪═══╪════════════╪═════╪════════════╪════════╡
│ 542215483 ┆ -137.599835 ┆ -25.394456 ┆ 2011-08-08 ┆ … ┆ 7 ┆ 0.0 ┆ 2 ┆ s │
│ ┆ ┆ ┆ 01:50:10 ┆ ┆ ┆ ┆ ┆ │
│ 736182029 ┆ -164.547314 ┆ -1.575633 ┆ 1971-10-26 ┆ … ┆ 0 ┆ 0.0 ┆ 3 ┆ g │
│ ┆ ┆ ┆ 15:19:43 ┆ ┆ ┆ ┆ ┆ │
│ 239534028 ┆ -162.337749 ┆ -89.481813 ┆ 2017-04-13 ┆ … ┆ 0 ┆ 0.0 ┆ 1 ┆ g │
│ ┆ ┆ ┆ 14:22:33 ┆ ┆ ┆ ┆ ┆ │
│ 679866937 ┆ -74.213448 ┆ 83.763916 ┆ 1997-11-30 ┆ … ┆ 0 ┆ 0.0 ┆ 3 ┆ g │
│ ┆ ┆ ┆ 14:06:45 ┆ ┆ ┆ ┆ ┆ │
│ 251451475 ┆ 17.389807 ┆ 45.864855 ┆ 2006-06-29 ┆ … ┆ 7 ┆ 0.0 ┆ 2 ┆ s │
│ ┆ ┆ ┆ 17:12:51 ┆ ┆ ┆ ┆ ┆ │
│ … ┆ … ┆ … ┆ … ┆ … ┆ … ┆ … ┆ … ┆ … │
│ 564118973 ┆ -129.684367 ┆ 86.209579 ┆ 2002-01-12 ┆ … ┆ 1 ┆ 0.0 ┆ 3 ┆ g │
│ ┆ ┆ ┆ 02:22:25 ┆ ┆ ┆ ┆ ┆ │
│ 455260195 ┆ -66.141805 ┆ -62.545182 ┆ 2010-02-27 ┆ … ┆ 1 ┆ 0.0 ┆ 2 ┆ s │
│ ┆ ┆ ┆ 11:33:57 ┆ ┆ ┆ ┆ ┆ │
│ 237920884 ┆ 53.691753 ┆ -28.465175 ┆ 1997-10-19 ┆ … ┆ 0 ┆ 0.0 ┆ 3 ┆ s │
│ ┆ ┆ ┆ 20:48:38 ┆ ┆ ┆ ┆ ┆ │
│ 791425241 ┆ -33.807152 ┆ 67.165186 ┆ 1991-10-26 ┆ … ┆ 7 ┆ 0.0 ┆ 3 ┆ g │
│ ┆ ┆ ┆ 23:28:23 ┆ ┆ ┆ ┆ ┆ │
│ 516122163 ┆ -32.170851 ┆ 41.255736 ┆ 1981-10-10 ┆ … ┆ 7 ┆ 0.0 ┆ 3 ┆ g │
│ ┆ ┆ ┆ 22:02:03 ┆ ┆ ┆ ┆ ┆ │
└───────────┴─────────────┴────────────┴──────────────┴───┴────────────┴─────┴────────────┴────────┘
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: (8_890, 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 |
| 834274217 | 67.946593 | -89.521388 | "1997-06-29 18:04:38" | -5.686281 | 1.023195 | 1.097587 | 0 | 0.0 | 3 | "g" |
| 800861098 | -65.152475 | 41.671351 | "1984-05-26 01:50:30" | 2.775006 | -4.645878 | -4.596816 | 7 | 0.0 | 1 | "s" |
| 805200489 | 54.561686 | 48.984988 | "2020-03-12 05:51:37" | -9.603851 | -1.647586 | -1.58416 | 1 | 0.0 | 3 | "s" |
| 826066981 | -146.616383 | 60.736464 | "1975-05-02 22:01:27" | -6.334956 | 1.690634 | 1.788756 | 0 | 0.0 | 2 | "s" |
| 195769281 | 56.077732 | -86.782308 | "1979-07-02 01:34:23" | 0.045738 | 2.118046 | 2.134311 | 1 | 0.0 | 1 | "s" |
| … | … | … | … | … | … | … | … | … | … | … |
| 831804110 | 175.131972 | -52.91488 | "1995-05-03 20:27:13" | -10.590274 | 1.3484 | 1.35584 | 7 | 0.0 | 3 | "g" |
| 804545530 | 89.622459 | 82.253685 | null | -4.200654 | 1.327063 | 1.400069 | 7 | 0.0 | 3 | "g" |
| 199238357 | -148.170576 | -82.162412 | "2021-03-24 17:21:15" | 12.459666 | -2.373441 | -2.360546 | 7 | 0.0 | 2 | "s" |
| 820765126 | -135.284538 | -39.161457 | null | 11.183032 | 2.128049 | 2.188734 | 1 | 0.0 | 1 | "s" |
| 801855206 | 167.04203 | -62.856694 | "1995-12-21 06:53:22" | -4.316075 | 3.333872 | 3.354648 | 7 | 0.0 | 3 | "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.metadata import DataCategory
from damast.core.dataframe import AnnotatedDataFrame
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.data_handling.transformers.filters import DropMissingOrNan
from damast.core.dataprocessing import DataProcessingPipeline
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, 279.07it/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: (151_612, 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 |
| 380121378 | 29.2714 | 44.932179 | "2015-12-22 09:04:09" | -3.003905 | 2.899003 | 2.991889 | 7 | 0.0 | 1 | "s" |
| 729205050 | -49.142089 | 76.903083 | "2013-02-15 02:38:10" | -5.121596 | -1.222903 | -1.142481 | 7 | 0.0 | 2 | "g" |
| 726139934 | -98.260442 | 81.390309 | "2022-08-10 22:11:09" | -0.640847 | -0.984218 | -0.886147 | 1 | 0.0 | 2 | "s" |
| 737787646 | 79.008118 | 26.723266 | "2004-11-29 08:33:13" | 4.155393 | -13.371644 | -13.275093 | 0 | 0.0 | 3 | "s" |
| 596263325 | -98.255897 | -35.40071 | "2005-11-09 16:37:12" | 9.744332 | -1.224419 | -1.167695 | 0 | 0.0 | 2 | "g" |
| … | … | … | … | … | … | … | … | … | … | … |
| 760316521 | -55.197746 | 84.95063 | "1989-02-06 13:02:54" | 8.40412 | -3.316311 | -3.269531 | 7 | 0.0 | 2 | "s" |
| 286674356 | 14.576202 | -8.226751 | "2012-09-19 01:18:32" | 15.178092 | -2.697753 | -2.623583 | 1 | 0.0 | 2 | "g" |
| 517377155 | -96.5238 | 85.273718 | "1981-07-03 03:24:23" | 10.528098 | -0.334419 | -0.242993 | 0 | 0.0 | 3 | "s" |
| 760441444 | -156.118917 | 57.575855 | "1981-03-03 15:12:50" | -35.368782 | -2.450755 | -2.39692 | 0 | 0.0 | 3 | "g" |
| 512767632 | 177.385443 | 49.252229 | "1971-01-18 03:37:04" | -5.02452 | -0.577512 | -0.565943 | 1 | 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<?, ?it/s]
Extract str and categorical column metadata: 100%|██████████| 4/4 [00:00<00:00, 1947.22it/s]
Extract numeric column metadata: 0%| | 0/2 [00:00<?, ?it/s]
Extract numeric column metadata: 100%|██████████| 2/2 [00:00<00:00, 14614.30it/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]
Step : 86%|████████▌ | 6/7 [00:00<00:00, 58.97it/s]
WARNING:damast.core.metadata:'comment'
Step : 100%|██████████| 7/7 [00:00<00:00, 39.05it/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 |
| 380121378 | 29.2714 | 44.932179 | "2015-12-22 09:04:09" | -3.003905 | 2.899003 | 2.991889 | 7 | 0.0 | 1 | "s" | -0.002296 | 0.005059 | 0.011011 | -0.001489 | 40.0 | 10.0 | "accident" | -5.4429e-17 | 0.005556 | -2.7214e-17 | 0.011111 |
| 380121378 | 29.2714 | 44.932179 | "2015-12-22 09:04:09" | -3.003905 | 2.899003 | 2.991889 | 7 | 0.0 | 1 | "s" | -0.002296 | 0.005059 | 0.011011 | -0.001489 | 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 import MyTransformer - without needing to import (or
even know) the underlying module or package that defines it.
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
point the DAMAST_PLUGIN_PATH environment variable at the directory containing it (multiple
directories can be separated with os.pathsep, just like PATH). Every top-level file found
there is imported once - using its filename stem as the module name - so any PipelineElement
subclasses it defines become resolvable exactly like classes from an installed package.
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"] = str(plugin_dir)
from damast.core.transformations import PipelineElement
# 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()
{'Doubler': 'my_transformers:Doubler'}
Doubler is now resolvable via damast.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.plugins import Doubler # resolved from the file we just wrote to `plugin_dir`
from damast.core.dataprocessing import DataProcessingPipeline
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.3
name: plugin-example
processing_graph:
edges:
- from: 6a2cf056-25ec-4674-8312-b04a6483b58d
to: abc3176a-bb4f-4bfe-b092-fcc7d3d9f12b
nodes:
- name: df
transformer:
class_name: DataSource
module_name: damast.core.processing_graph
name_mappings:
df: {}
parameters: {}
requires:
distribution: damast
version: 0.3.3
uuid: 6a2cf056-25ec-4674-8312-b04a6483b58d
- name: double_mmsi
transformer:
class_name: Doubler
module_name: my_transformers
name_mappings:
df:
x: mmsi
parameters: {}
requires:
hint: local
path: /tmp/tmpcz7g15b4/my_transformers.py
uuid: abc3176a-bb4f-4bfe-b092-fcc7d3d9f12b
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"
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 import MyTransformer.