Command Line Interface

The command line interface offers a number of workflow simplifications that are encapsulated in sub-commands:

inspect

check metadadata and dataset properties

convert

convert from (zipped) csv, netcdf to parquet (default) or hdf5 (deprecated)

annotate

create metadata file and update dataframe with metadata

process

apply a data pipeline to a dataset

plugins

list transformer plugins registered by installed packages or via DAMAST_PLUGIN_PATH

watch

watch directories for completed files and run a configured command on each

Inspect

usage: damast inspect [-h] [-w WORKDIR] [-v]
                      [--log-level {CRITICAL,FATAL,ERROR,WARN,WARNING,INFO,DEBUG,NOTSET}]
                      [--log-file LOG_FILE] [--version] -f FILES [FILES ...]
                      [--filter FILTER [FILTER ...]] [--head HEAD]
                      [--tail TAIL] [--column-count COLUMN_COUNT]
                      [--column-width COLUMN_WIDTH]
                      [--columns COLUMNS [COLUMNS ...]]
                      [--validation-mode {ignore,readonly,update_data,update_metadata}]
                      [-s SAVE_AS] [-e EXPORT_METADATA]
                      {} ...

damast inspect - data inspection subcommand called

positional arguments:
  {}                    sub-command help

options:
  -h, --help            show this help message and exit
  -w WORKDIR, --workdir WORKDIR
  -v, --verbose
  --log-level {CRITICAL,FATAL,ERROR,WARN,WARNING,INFO,DEBUG,NOTSET}
                        Set loglevel to display
  --log-file LOG_FILE   Set file for saving log (default prints to terminal)
  --version             Show current version of damast
  -f FILES [FILES ...], --files FILES [FILES ...]
                        Files or patterns of the (annotated) data file that
                        should be inspected (space separated)
  --filter FILTER [FILTER ...]
                        Filter based on column data, e.g., mmsi==120123,
                        allowed operators !=,<,>,==,<=,>=,<> or =~ for a
                        regex, e.g.,'name =~ '^SAR*'
  --head HEAD           First this number of rows, default is 10
  --tail TAIL           Print number of rows from the end, default is 10
  --column-count COLUMN_COUNT
                        Number of columns to show
  --column-width COLUMN_WIDTH
                        Column width to show
  --columns COLUMNS [COLUMNS ...]
                        Show/Select these columns
  --validation-mode {ignore,readonly,update_data,update_metadata}
                        Define the validation mode
  -s SAVE_AS, --save-as SAVE_AS
                        If filters are being used, it will save the result in
                        the given file (all columns being used). A plain path
                        saves one file; use
                        'time:<column>+<interval>:<template>',
                        'column:<column>:<template>', or
                        'time+column:<column>+<interval>+<column>:<template>'
                        to instead save one file per partition - see
                        damast.core.partitioning.SaveAs.parse
  -e EXPORT_METADATA, --export-metadata EXPORT_METADATA
                        Export the (updated) metadata spec file to the given
                        path

Inspect allows to identify columns and properties of columns in a given dataset. The dataset can consist of one or more (zipped) files, either given as list of filenames or using file pattern.

$ damast inspect -f 1.zip

Subparser: DataInspectParser
Loading dataframe (1 files) of total size: 0.0 MB
Creating offset dictionary for /tmp/damast-example/datasets/1.zip ...
Creating offset dictionary for /tmp/damast-example/datasets/1.zip took 0.00s
Created mount point at: /tmp/damast-mountqigwlx74/1.zip
INFO:damast.core.dataframe:Loading parquet: files=[PosixPath('/tmp/damast-mountqigwlx74/1.zip/dataset-1.zst.parquet')]
WARNING:damast.core.dataframe:/tmp/damast-mountqigwlx74/1.zip/dataset-1.zst.parquet has no (damast) annotations
INFO:damast.core.dataframe:No metadata provided or found in files - searching now for an existing spec file
INFO:damast.core.dataframe:Found no candidate for a spec file
INFO:damast.core.dataframe:Metadata is not available and not required, so inferring annotation
Extract str and categorical column metadata: 100%|██████████████████████████████████████████████████████████████████████████████████████████████████████████| 32/32 [00:00<00:00, 1092.51it/s]
Extract numeric column metadata: 100%|█████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 19/19 [00:00<00:00, 12767.03it/s]
INFO:damast.core.dataframe:Metadata inferring completed
Annotations:
    accuracy:
        is_optional: False
        representation_type: Boolean
    call_sign:
        is_optional: False
        representation_type: String
        value_range: {'ListOfValues': [None, '', 'SIDF9', 'SABD4', 'STDL5', 'STJE3', 'SKCY7', 'XAGBE']}
    cog:
        is_optional: False
        representation_type: Float32
        value_stats: {'mean': 142.0380096435547, 'stddev': 117.50126647949219, 'total_count': 1234, 'null_count': 745}
    corrupted:
        is_optional: False
        representation_type: Boolean
    corrupted_right:
        is_optional: False
        representation_type: Boolean
    destination:
        is_optional: False
        representation_type: String
        value_range: {'ListOfValues': [None, '', 'VILA', 'ES SUR', 'ESICL', 'EBAL>EDGA', 'IT-SEP', 'PLATF ROMA', 'ITL-BREG']}
    dimension_to_bow:
        is_optional: False
        representation_type: UInt16

 ...
     sog:
     is_optional: False
     representation_type: Float32
     value_stats: {'mean': 2.0780696868896484, 'stddev': 4.677201271057129, 'total_count': 1979, 'null_count': 0}
 version:
     is_optional: False
     representation_type: Int64
     value_range: {'MinMax': {'min': 3, 'max': 3, 'allow_missing': True}}
     value_stats: {'mean': 3.0, 'stddev': 0.0, 'total_count': 1979, 'null_count': 0}


 First 10 and last 10 rows:
 shape: (10, 32)
 ┌───────────┬─────────────────────┬──────────┬───────────┬──────┬───┬────────────┬────────────────────┬──────────────────┬───────────────────────┬─────────┐
 │ mmsi      ┆ reception_date      ┆ lon      ┆ lat       ┆ rot  ┆ … ┆ eta        ┆ message_type_right ┆ satellite_static ┆ reception_date_static ┆ version │
 │ ---       ┆ ---                 ┆ ---      ┆ ---       ┆ ---  ┆   ┆ ---        ┆ ---                ┆ ---              ┆ ---                   ┆ ---     │
 │ i32       ┆ datetime[ms]        ┆ f64      ┆ f64       ┆ f32  ┆   ┆ i64        ┆ i64                ┆ str              ┆ datetime[ms]          ┆ i64     │
 ╞═══════════╪═════════════════════╪══════════╪═══════════╪══════╪═══╪════════════╪════════════════════╪══════════════════╪═══════════════════════╪═════════╡
 │ 345080000 ┆ 2020-11-18 33:00:18 ┆ 0.783398 ┆ 40.483513 ┆ 0.0  ┆ … ┆ null       ┆ null               ┆ null             ┆ null                  ┆ 3       │
 │ 334015340 ┆ 2020-11-18 33:00:33 ┆ 0.435345 ┆ 40.414097 ┆ null ┆ … ┆ 1735889800 ┆ 5                  ┆ SAT-AA_037       ┆ 2020-11-18 33:07:35   ┆ 3       │
 │ 334088470 ┆ 2020-11-18 33:00:37 ┆ 0.403745 ┆ 40.358495 ┆ null ┆ … ┆ null       ┆ null               ┆ null             ┆ null                  ┆ 3       │
 │ 334098970 ┆ 2020-11-18 33:00:39 ┆ 0.88999  ┆ 40.389833 ┆ null ┆ … ┆ 1783310700 ┆ 5                  ┆ SAT-AA_038       ┆ 2020-11-18 33:04:13   ┆ 3       │
 │ 333019738 ┆ 2020-11-18 33:01:18 ┆ 0.80045  ┆ 40.819483 ┆ null ┆ … ┆ null       ┆ 34                 ┆ SAT-AA_038       ┆ 2020-11-18 33:33:51   ┆ 3       │
 │ 353003075 ┆ 2020-11-18 33:01:38 ┆ 0.550948 ┆ 40.571973 ┆ 0.0  ┆ … ┆ null       ┆ 5                  ┆ SAT-AA_037       ┆ 2020-11-18 33:01:13   ┆ 3       │
 │ 345080000 ┆ 2020-11-18 33:01:37 ┆ 0.759477 ┆ 40.481487 ┆ 0.0  ┆ … ┆ null       ┆ null               ┆ null             ┆ null                  ┆ 3       │
 │ 334015340 ┆ 2020-11-18 33:01:33 ┆ 0.435338 ┆ 40.414093 ┆ null ┆ … ┆ 1735889800 ┆ 5                  ┆ SAT-AA_037       ┆ 2020-11-18 33:07:35   ┆ 3       │
 │ 334088470 ┆ 2020-11-18 33:01:37 ┆ 0.403743 ┆ 40.358513 ┆ null ┆ … ┆ null       ┆ null               ┆ null             ┆ null                  ┆ 3       │
 │ 334098970 ┆ 2020-11-18 33:03:18 ┆ 0.890313 ┆ 40.370833 ┆ null ┆ … ┆ 1783310700 ┆ 5                  ┆ SAT-AA_038       ┆ 2020-11-18 33:04:13   ┆ 3       │
 └───────────┴─────────────────────┴──────────┴───────────┴──────┴───┴────────────┴────────────────────┴──────────────────┴───────────────────────┴─────────┘
 shape: (10, 32)
 ┌───────────┬─────────────────────┬──────────┬───────────┬──────┬───┬────────────┬────────────────────┬──────────────────┬───────────────────────┬─────────┐
 │ mmsi      ┆ reception_date      ┆ lon      ┆ lat       ┆ rot  ┆ … ┆ eta        ┆ message_type_right ┆ satellite_static ┆ reception_date_static ┆ version │
 │ ---       ┆ ---                 ┆ ---      ┆ ---       ┆ ---  ┆   ┆ ---        ┆ ---                ┆ ---              ┆ ---                   ┆ ---     │
 │ i32       ┆ datetime[ms]        ┆ f64      ┆ f64       ┆ f32  ┆   ┆ i64        ┆ i64                ┆ str              ┆ datetime[ms]          ┆ i64     │
 ╞═══════════╪═════════════════════╪══════════╪═══════════╪══════╪═══╪════════════╪════════════════════╪══════════════════╪═══════════════════════╪═════════╡
 │ 335990004 ┆ 2020-11-19 01:59:00 ┆ 0.849883 ┆ 40.937813 ┆ null ┆ … ┆ null       ┆ 34                 ┆ SAT-AA_038       ┆ 2020-11-19 01:31:43   ┆ 3       │
 │ 334015340 ┆ 2020-11-19 03:00:13 ┆ 0.435335 ┆ 40.414083 ┆ null ┆ … ┆ 1735889800 ┆ 5                  ┆ SAT-AA_037       ┆ 2020-11-19 01:19:34   ┆ 3       │
 │ 334088470 ┆ 2020-11-19 03:00:19 ┆ 0.40377  ┆ 40.358493 ┆ null ┆ … ┆ null       ┆ null               ┆ null             ┆ null                  ┆ 3       │
 │ 333049539 ┆ 2020-11-19 03:00:31 ┆ 0.80088  ┆ 40.819835 ┆ null ┆ … ┆ null       ┆ 34                 ┆ SAT-AA_038       ┆ 2020-11-19 01:03:01   ┆ 3       │
 │ 334018830 ┆ 2020-11-19 03:00:35 ┆ 0.895348 ┆ 40.897835 ┆ null ┆ … ┆ 1735889800 ┆ 5                  ┆ SAT-AA_038       ┆ 2020-11-19 01:07:38   ┆ 3       │
 │ 333058871 ┆ 2020-11-19 03:00:31 ┆ 0.800105 ┆ 40.819735 ┆ null ┆ … ┆ null       ┆ 34                 ┆ SAT-AA_037       ┆ 2020-11-19 00:59:00   ┆ 3       │
 │ 334098970 ┆ 2020-11-19 03:00:37 ┆ 0.891373 ┆ 40.403033 ┆ null ┆ … ┆ 1783310700 ┆ 5                  ┆ SAT-AA_038       ┆ 2020-11-19 01:04:13   ┆ 3       │
 │ 345080000 ┆ 2020-11-19 03:00:38 ┆ 0.373085 ┆ 40.10578  ┆ 0.0  ┆ … ┆ null       ┆ null               ┆ null             ┆ null                  ┆ 3       │
 │ 333041379 ┆ 2020-11-19 03:00:48 ┆ 0.801778 ┆ 40.818448 ┆ null ┆ … ┆ null       ┆ null               ┆ null             ┆ null                  ┆ 3       │
 │ 333048134 ┆ 2020-11-19 03:00:54 ┆ 0.803097 ┆ 40.830813 ┆ null ┆ … ┆ null       ┆ 34                 ┆ SAT-AA_037       ┆ 2020-11-19 01:08:30   ┆ 3       │
 └───────────┴─────────────────────┴──────────┴───────────┴──────┴───┴────────────┴────────────────────┴──────────────────┴───────────────────────┴─────────┘

By default, inspect runs in READONLY validation mode: it only checks metadata against the data, without computing or changing anything. If a file’s stored metadata is missing value_range/value_stats, then pass --validation-mode update_metadata to have them computed from the actual data instead:

$ damast inspect -f data.parquet --validation-mode update_metadata

To make the information persistent in the parquet file:

$ damast annotate -f data.parquet --apply --inplace

Examples

Individual columns can be filtered using a python expression that is compliant with the backend (here: polars) being used.

For instance to extract:

• the time-series for a particular id (mmsi):

damast inspect -f 1.zip --filter 'mmsi == 335990004'

• all data in a time interval:

damast inspect -f 1.zip --filter 'reception_date >= dt.datetime.fromisoformat("2020-11-19 00:00:00")' --filter 'reception_date <= dt.datetime.fromisoformat("2020-11-20 00:00:00")'

--save-as (only active together with --filter) saves the filtered result. A plain path saves it as a single file:

damast inspect -f 1.zip --filter 'mmsi == 335990004' -save-as filtered.parquet

To instead save one file per partition, use a <strategy>:<spec>:<template> string - time:<timestamp_column>+<interval>:<template> (one file per time bucket, <interval> is hourly/daily/weekly/monthly or any raw polars.Expr.dt.truncate interval, e.g. "3h"), column:<column>:<template> (one file per distinct value of <column>), or time+column:<timestamp_column>+<interval>+<column>:<template> (one file per (time bucket, column value) pair). <template> is the filename stem, combining literal text - including / for a nested output directory - with strftime codes (%Y, %m, …) and {<column>} placeholders:

damast inspect -f 1.zip --filter 'reception_date >= dt.datetime.fromisoformat("2020-11-19 00:00:00")' -save-as "time:reception_date+daily:export/AIS_%Y_%m_%d"
damast inspect -f 1.zip --filter 'mmsi > 0' -save-as "column:mmsi:export/vessel_{mmsi}"
damast inspect -f 1.zip --filter 'mmsi > 0' -save-as "time+column:reception_date+daily+mmsi:export/{mmsi}/AIS_%Y_%m_%d"

See damast.core.partitioning.SaveAs.parse for the full grammar.

Convert

usage: damast convert [-h] [-w WORKDIR] [-v]
                      [--log-level {CRITICAL,FATAL,ERROR,WARN,WARNING,INFO,DEBUG,NOTSET}]
                      [--log-file LOG_FILE] [--version] -f FILES [FILES ...]
                      [-m METADATA_INPUT] [-o OUTPUT_FILE]
                      [--output-dir OUTPUT_DIR] [--output-type OUTPUT_TYPE]
                      [-s SAVE_AS] [-c {none,snappy,gzip,brotli,lz4,zstd,bz2}]
                      [-l COMPRESSION_LEVEL]
                      [--validation-mode {ignore,readonly,update_data,update_metadata}]
                      {} ...

damast convert - data conversion subcommand called

positional arguments:
  {}                    sub-command help

options:
  -h, --help            show this help message and exit
  -w WORKDIR, --workdir WORKDIR
  -v, --verbose
  --log-level {CRITICAL,FATAL,ERROR,WARN,WARNING,INFO,DEBUG,NOTSET}
                        Set loglevel to display
  --log-file LOG_FILE   Set file for saving log (default prints to terminal)
  --version             Show current version of damast
  -f FILES [FILES ...], --files FILES [FILES ...]
                        Files or patterns of the (annotated) data file that
                        should be converted
  -m METADATA_INPUT, --metadata-input METADATA_INPUT
                        The metadata input file
  -o OUTPUT_FILE, --output-file OUTPUT_FILE
                        The output file either: .parquet, .hdf5 DEPRECATED:
                        use --save-as instead
  --output-dir OUTPUT_DIR
                        The output directory. DEPRECATED: use --save-as
                        instead
  --output-type OUTPUT_TYPE
                        The output file type: .parquet or .hdf5 (default:
                        .parquet)
  -s SAVE_AS, --save-as SAVE_AS
                        Combine all input files and save the result according
                        to this spec, instead of --output-file/--output-dir. A
                        plain path saves one file; use
                        'time:<column>+<interval>:<template>',
                        'column:<column>:<template>', or
                        'time+column:<column>+<interval>+<column>:<template>'
                        to instead save one file per partition - see
                        damast.core.partitioning.SaveAs.parse
  -c {none,snappy,gzip,brotli,lz4,zstd,bz2}, --compression-type {none,snappy,gzip,brotli,lz4,zstd,bz2}
  -l COMPRESSION_LEVEL, --compression-level COMPRESSION_LEVEL
  --validation-mode {ignore,readonly,update_data,update_metadata}
                        Define the validation mode (default: update_data)

Examples

• convert one or more files to parquet (N:N)

damast convert -f 1.zip --output-dir export --output-type .parquet

• convert one or more files to a single parquet file (N:1)

damast convert -f 1.zip --output-file data-1.parquet --output-type .parquet

• combine one or more files and split the result into partitions (N:M), instead of --output-file/--output-dir - see damast.core.partitioning.SaveAs.parse for the full <strategy>:<spec>:<template> grammar:

damast convert -f 1.zip -save-as "time:reception_date+daily:export/AIS_%Y_%m_%d"
damast convert -f 1.zip -save-as "column:mmsi:export/vessel_{mmsi}"

Annotate

usage: damast annotate [-h] [-w WORKDIR] [-v]
                       [--log-level {CRITICAL,FATAL,ERROR,WARN,WARNING,INFO,DEBUG,NOTSET}]
                       [--log-file LOG_FILE] [--version] -f FILES [FILES ...]
                       [-m METADATA_INPUT] [-o OUTPUT_DIR]
                       [--output-spec-file OUTPUT_SPEC_FILE]
                       [--set-abbreviation COLUMN:VALUE [COLUMN:VALUE ...]]
                       [--set-category COLUMN:VALUE [COLUMN:VALUE ...]]
                       [--set-description COLUMN:VALUE [COLUMN:VALUE ...]]
                       [--set-is_optional COLUMN:VALUE [COLUMN:VALUE ...]]
                       [--set-precision COLUMN:VALUE [COLUMN:VALUE ...]]
                       [--set-representation_type COLUMN:VALUE [COLUMN:VALUE ...]]
                       [--set-unit COLUMN:VALUE [COLUMN:VALUE ...]]
                       [--inplace] [--apply]
                       {} ...

damast annotate - extract (default) or apply annotation to dataset

positional arguments:
  {}                    sub-command help

options:
  -h, --help            show this help message and exit
  -w WORKDIR, --workdir WORKDIR
  -v, --verbose
  --log-level {CRITICAL,FATAL,ERROR,WARN,WARNING,INFO,DEBUG,NOTSET}
                        Set loglevel to display
  --log-file LOG_FILE   Set file for saving log (default prints to terminal)
  --version             Show current version of damast
  -f FILES [FILES ...], --files FILES [FILES ...]
                        Files or patterns of the (annotated) data file that
                        should be annotated
  -m METADATA_INPUT, --metadata-input METADATA_INPUT
                        The metadata that is applied to the data
  -o OUTPUT_DIR, --output-dir OUTPUT_DIR
                        Output directory
  --output-spec-file OUTPUT_SPEC_FILE
                        The spec file name - if provided with path, it will
                        override output-dir
  --set-abbreviation COLUMN:VALUE [COLUMN:VALUE ...]
                        Set abbreviation in spec for a column and value
  --set-category COLUMN:VALUE [COLUMN:VALUE ...]
                        Set category in spec for a column and value
  --set-description COLUMN:VALUE [COLUMN:VALUE ...]
                        Set description in spec for a column and value
  --set-is_optional COLUMN:VALUE [COLUMN:VALUE ...]
                        Set is_optional in spec for a column and value
  --set-precision COLUMN:VALUE [COLUMN:VALUE ...]
                        Set precision in spec for a column and value
  --set-representation_type COLUMN:VALUE [COLUMN:VALUE ...]
                        Set representation_type in spec for a column and value
  --set-unit COLUMN:VALUE [COLUMN:VALUE ...]
                        Set unit in spec for a column and value
  --inplace             Update the dataset inplace (only possible for a single
                        file)
  --apply               Update the annotation, infer and rewrite the metadata
                        to the dataset (implicitly used when using --set-X
                        option)

Examples

• set the unit for two columns, here lat and lon to deg, and creating a new file in the subfolder export

damast annotate -f input.parquet --set-unit lon:deg lat:deg --output-dir export

• set the unit for two columns, here lat and lon to deg, inplace, i.e., change the existing file

damast annotate -f input.parquet --set-unit lon:deg lat:deg --inplace

Process

usage: damast process [-h] [-w WORKDIR] [-v]
                      [--log-level {CRITICAL,FATAL,ERROR,WARN,WARNING,INFO,DEBUG,NOTSET}]
                      [--log-file LOG_FILE] [--version]
                      [--input-data INPUT_DATA [INPUT_DATA ...]] --pipeline
                      PIPELINE [--output-file OUTPUT_FILE]
                      [--base-dir BASE_DIR] [--describe]
                      [--export {html,mermaid,svg}]
                      [--mermaid-templates MERMAID_TEMPLATES]
                      {} ...

damast process - apply an existing pipeline

positional arguments:
  {}                    sub-command help

options:
  -h, --help            show this help message and exit
  -w WORKDIR, --workdir WORKDIR
  -v, --verbose
  --log-level {CRITICAL,FATAL,ERROR,WARN,WARNING,INFO,DEBUG,NOTSET}
                        Set loglevel to display
  --log-file LOG_FILE   Set file for saving log (default prints to terminal)
  --version             Show current version of damast
  --input-data INPUT_DATA [INPUT_DATA ...]
                        Input file(s) for a datasource - 'FILE...' for a
                        single-datasource pipeline, or repeated '--input-data
                        NAME=FILE...' (one per datasource, e.g.
                        'osint_events=1.parquet 2.parquet') for a pipeline
                        that requires more than one. Not required if
                        --describe is given.
  --pipeline PIPELINE   Pipeline (*.damast.ppl) file to apply to the data
  --output-file OUTPUT_FILE
                        Save the result of a pipeline in the given (*.parquet)
                        file
  --base-dir BASE_DIR   Save pipeline artifacts relative to the given base
                        directory (default: .)
  --describe            Print the pipeline's interface (required datasources
                        and their columns) and steps, then exit - no --input-
                        data needed
  --export {html,mermaid,svg}
                        Render the pipeline's steps, dataflow and interfaces
                        to the given format and exit - no --input-data needed.
                        Written to --output-file, or '<base-dir>/<pipeline-
                        name>' with the format's default extension if that's
                        not given. 'svg' requires the Graphviz 'dot'
                        executable; 'html'/'mermaid' need no external tool
  --mermaid-templates MERMAID_TEMPLATES
                        Directory of .j2 templates overriding the default
                        Mermaid rendering for '--export html'/'--export
                        mermaid' - only the files present there are overridden
                        (e.g. a single 'datasource_block.j2'), everything else
                        keeps using the shipped default

Once a DataProcessPipeline has been exported and saved, e.g., in the following example as my-pipeline.damast.ppl, it can be reapplied to an existing data set. The dataset needs to comply with the required input columns and metadata requirements, such as units, so that the pipeline can successfully run. Damast will check these requirements and raise an exception if these requirements are not satisfied.

from pathlib import Path

from damast.core import DataProcessingPipeline
from damast.data_handling.transformers import AddDeltaTime
from damast.domains.maritime.transformers.features import DeltaDistance, Speed


class MyPipeline(DataProcessingPipeline):
    def __init__(self,
                 workdir: str | Path,
                 name: str = "my-pipeline",
                 name_mappings: dict[str, str] = {}):
        super().__init__(name=name,
                         base_dir=workdir,
                         name_mappings=name_mappings)

        self.add("Delta Time",
                 AddDeltaTime(),
                 name_mappings={
                     "group": "mmsi",
                     "time_column": "reception_date"
                })

        self.add("Delta Distance",
                 DeltaDistance(x_shift=True, y_shift=True),
                 name_mappings={
                     "group": "mmsi",
                     "sort": "reception_date",
                     "x": "lat",
                     "y": "lon",
                     "out": "delta_distance",
                })

        self.add("Speed",
                 Speed(),
                 name_mappings={
                     "delta_distance": "delta_distance",
                     "delta_time": "delta_time",
                })

pipeline = MyPipeline(workdir=".")
pipeline.save("pipelines")

Examples

damast process --input-data input.parquet --pipeline pipelines/my-pipeline.damast.ppl

Multiple input datasources

A pipeline is not limited to a single input - damast.core.dataprocessing.DataProcessingPipeline.join() lets a pipeline declare additional, named datasources, e.g., here to fuse AIS ship-position pings with OSINT (open-source intelligence) event reports by timestamp. Using damast process with --input-data allows to specify a datasource with a prefix, e.g., --input-data osint_events=1.parquet 2.parquet. A pipeline with only the default datasource (the common case) can omit the prefix --input-data 1.parquet name needed.

The following trims down the pattern used in damast-examples/hozint-ais/ais_osint_fusion.py to a minimal, self-contained example - a JoinByTimestamp plugin transformer (see Plugins below) that joins AIS pings (the default df datasource) with OSINT events (a second datasource, osint_events) wherever their timestamps match:

import damast
from damast.core.dataframe import AnnotatedDataFrame
from damast.core.transformations import PipelineElement


class JoinByTimestamp(PipelineElement):
    """Join two datasources on a shared timestamp column - a minimal join operator."""

    @damast.core.describe("Join by timestamp")
    @damast.core.input({"timestamp": {}, "lon": {}, "lat": {}})
    @damast.core.input({"timestamp": {}, "lat": {}, "lon": {}}, label="other")
    @damast.core.output({
        "{{other:event_type}}": {"representation_type": str},
        "{{other:lat}}": {},
        "{{other:lon}}": {},
    })
    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)
        df._metadata = df._metadata.merge(other._metadata).drop(other_timestamp)
        return df

"""
Build and save a pipeline with two named datasources - 'df' for AIS pings, 'osint_events' for
OSINT reports - joined by timestamp. Inspired by damast-examples/hozint-ais/ais_osint_fusion.py,
trimmed to a self-contained example using only the damast core API.

Usage:
    python docs/examples/damast-osint-ais-generate-data.py
    python docs/examples/damast-osint-ais-pipeline.py
    damast process --pipeline pipelines/osint_ais_preparation.damast.ppl \\
        --input-data df=docs/examples/data/ais.parquet \\
        --input-data osint_events=docs/examples/data/osint.parquet \\
        --output-file output.parquet
"""
import os
from pathlib import Path

# JoinByTimestamp lives in its own package/file in a real project - see docs/cli.rst >
# Plugins. Here it's a local plugin file, resolved the same way.
os.environ["DAMAST_PLUGIN_PATH"] = f"ais_osint={Path(__file__).parent / 'plugins'}"

from damast.core.dataprocessing import DataProcessingPipeline
from damast.plugins.ais_osint import JoinByTimestamp

pipeline = DataProcessingPipeline(name="osint_ais_preparation",
                                  description="Join AIS pings with OSINT events by timestamp",
                                  base_dir=".") \
    .join("osint_events", JoinByTimestamp(),
          name_mappings={
              "df": {"timestamp": "date_time_utc"},
              "other": {"timestamp": "timestamp", "lat": "latitude", "lon": "longitude"},
          })

pipeline_path = pipeline.save("pipelines")
print(f"Saved pipeline to {pipeline_path}")

Generating small synthetic datasets to try it on:

"""
Generate small synthetic AIS and OSINT datasets for the multi-datasource pipeline example -
see damast-osint-ais-pipeline.py and docs/cli.rst > Process > Multiple input datasources.

Inspired by damast-examples/hozint-ais/ais_osint_fusion.py, trimmed to what's needed here and
using only generated/synthetic data - no real datasets are shipped with damast.
"""
from pathlib import Path

import polars

import damast.domains.maritime.ais.data_generator as generator

output_dir = Path(__file__).parent / "data"
output_dir.mkdir(parents=True, exist_ok=True)

ais_data = generator.AISTestData(number_of_trajectories=5, min_length=10, max_length=20)
ais_df = ais_data.dataframe
ais_df.write_parquet(output_dir / "ais.parquet")

# two OSINT "events" derived from real AIS pings, so the join in the example pipeline has
# actual matches to show, plus one deliberately unmatched event
sample = ais_df.filter(polars.col("date_time_utc").is_not_null()).sample(2, seed=1)
matched_events = sample.select(
    timestamp="date_time_utc", latitude="lat", longitude="lon",
).with_columns(event_type=polars.Series(["sighting", "distress-call"]))

unrelated_event = polars.DataFrame({
    "timestamp": ["1999-01-01 00:00:00"],
    "latitude": [0.0],
    "longitude": [0.0],
    "event_type": ["unrelated"],
})

osint_df = polars.concat([matched_events, unrelated_event])
osint_df.write_parquet(output_dir / "osint.parquet")

print(f"Wrote {output_dir / 'ais.parquet'} ({len(ais_df)} rows)")
print(f"Wrote {output_dir / 'osint.parquet'} ({len(osint_df)} rows)")

python docs/examples/damast-osint-ais-generate-data.py
python docs/examples/damast-osint-ais-pipeline.py

# JoinByTimestamp is a local plugin transformer (see Plugins below) - needed both to
# build the pipeline above and to load it back for damast process
export DAMAST_PLUGIN_PATH=ais_osint=docs/examples/plugins
damast process --pipeline pipelines/osint_ais_preparation.damast.ppl \
    --input-data df=docs/examples/data/ais.parquet \
    --input-data osint_events=docs/examples/data/osint.parquet \
    --output-file output.parquet

Describe a pipeline

--describe prints a saved pipeline’s interface: every datasource it requires and the columns each one must provide, followed by every processing step. Noe input data or running the pipeline is needed. Loading the pipeline still needs its transformers to be resolvable, i.e., DAMAST_PLUGIN_PATH must be set for a local plugin, while installed plugins will automatically be discovered:

export DAMAST_PLUGIN_PATH=ais_osint=docs/examples/plugins
damast process --pipeline pipelines/osint_ais_preparation.damast.ppl --describe

Subparser: DataProcessingParser
Pipeline: osint_ais_preparation
Description: Join AIS pings with OSINT events by timestamp
Base dir: .

Interface:
  df:
        date_time_utc
        lon
        lat
  osint_events:
        timestamp
        latitude
        longitude

Steps:
#1 df
    DataSource
            description: Node for marking a plain data entry
            input:
            output:
#2 osint_events
    DataSource
            description: Node for marking a plain data entry
            input:
            output:
#3 osint_events
    JoinByTimestamp
            description: Join by timestamp
            input:
                other:
                    timestamp
                    latitude
                    longitude
                df:
                    date_time_utc
                    lon
                    lat
            output:
                    event_type
                    latitude
                    longitude

The Interface section is specifically the requirement for each datasource’s first consuming step - once inside the pipeline, later steps consume columns the pipeline itself has already produced, not the raw datasource, so this is the actual external contract to satisfy when supplying --input-data.

Experiment tracking

damast.integrations.mlflow_tracker.track_pipeline permits a pipeline to use [mlflow](https://mlflow.org) for reporting. Part of the reporting is the output AnnotatedDataFrame’s metadata contract, i.e., units, and representation_type as well as runtime stats for per-step timing and row counts.

The feature requires mlflow, which will be installed with the extra ml: (uv pip install damast[ml]).

from pathlib import Path

import polars
from astropy import units

from damast.core.data_description import MinMax
from damast.core.dataframe import AnnotatedDataFrame
from damast.core.dataprocessing import DataProcessingPipeline
from damast.core.metadata import DataSpecification, MetaData, ValidationMode
from damast.integrations.mlflow_tracker import track_pipeline

pipeline = DataProcessingPipeline.load(Path("pipelines") / "my-pipeline.damast.ppl")

df = polars.DataFrame({"lat": [59.9], "lon": [10.7]})
metadata = MetaData(columns=[
    DataSpecification(name="lat", unit=units.deg, value_range=MinMax(-90.0, 90.0)),
    DataSpecification(name="lon", unit=units.deg, value_range=MinMax(-180.0, 180.0)),
])
adf = AnnotatedDataFrame(df, metadata, validation_mode=ValidationMode.UPDATE_METADATA)

# tracking_uri/experiment_name default to mlflow's own defaults (a local ./mlruns dir) if omitted
with track_pipeline(pipeline, run_name="my-pipeline-run") as tracker:
    result = pipeline.transform(adf)
    tracker.log_result(result)

A damast.ml.experiments.Experiment (see the Experiments notebook) combines such a pipeline with model training and evaluation. Experiment.run() performs all three internally, so the pipeline-level contract and the per-epoch training metrics need two separate hooks in the same run: wrap the call in track_pipeline for the pipeline (as above), and additionally call mlflow.keras.autolog() beforehand.

The experiment report that Experiment.run() writes (training parameters, per-model evaluation results) isn’t part of any AnnotatedDataFrame’s metadata, so it is logged separately, straight onto the tracker:

import mlflow.keras
import yaml

from damast.integrations.mlflow_tracker import track_pipeline
from damast.ml.experiments import Experiment

experiment = Experiment.from_file("experiment.yaml")

mlflow.keras.autolog()  # per-epoch metrics + the trained model, straight from model.fit()

with track_pipeline(experiment.learning_task.pipeline, run_name=experiment.label) as tracker:
    report_path = experiment.run()  # computes features, trains, evaluates, writes the yaml report

    with open(report_path) as f:
        report = yaml.safe_load(f)

    # training_parameters/evaluation aren't part of any AnnotatedDataFrame's metadata, so log
    # them directly rather than through tracker.log_result()
    tracker.log_params({
        f"training.{k}": v for k, v in report["learning_task"]["training_parameters"].items()
    })
    for model_name, results in report["evaluation"].items():
        tracker.log_metrics({f"eval.{model_name}.{k}": v for k, v in results.items()})
    tracker.log_artifact(report_path)

If other trackers, e.g. W&B, shall be used, an integration can be based on damast.integrations.tracking.flatten_metadata/flatten_step_stats and the damast.integrations.tracking.ExperimentTracker interface, both are backend-agnostic,

Watch

usage: damast watch [-h] [-w WORKDIR] [-v]
                    [--log-level {CRITICAL,FATAL,ERROR,WARN,WARNING,INFO,DEBUG,NOTSET}]
                    [--log-file LOG_FILE] [--version] [-c CONFIG] [--job JOB]
                    [--dry-run] [--create-config]
                    {} ...

damast watch - watch directories for completed files and act on them

positional arguments:
  {}                    sub-command help

options:
  -h, --help            show this help message and exit
  -w WORKDIR, --workdir WORKDIR
  -v, --verbose
  --log-level {CRITICAL,FATAL,ERROR,WARN,WARNING,INFO,DEBUG,NOTSET}
                        Set loglevel to display
  --log-file LOG_FILE   Set file for saving log (default prints to terminal)
  --version             Show current version of damast
  -c CONFIG, --config CONFIG
                        YAML watch config file, defining one or more jobs
  --job JOB             Restrict to job(s) with this name (repeatable);
                        default: run every job in the config
  --dry-run             List files that are ready to be processed, without
                        running anything
  --create-config       Interactively create the watch config file given via
                        --config, instead of running a scan

Some data sources are collected incrementally, e.g. one file per day, appended to for hours before it is complete. watch scans one or more directories for such files and, once a file has not been modified for a configurable quiet period, runs a configured command on it - typically damast convert with a metadata spec, or damast process with a pipeline.

watch performs a single scan-and-exit; it is not a daemon. Schedule it periodically with cron or a systemd timer (see Deployment below). On success a source file is moved into the job’s processed_dir, on failure into failed_dir alongside a <file>.error.log - this also means a file is only ever handled once, even across repeated invocations.

Configuration

A watch config is a YAML file listing one or more jobs:

jobs:
  - name: ais-daily                 # optional; default: source_dir's basename
    source_dir: /data/incoming/ais
    target_dir: /data/processed/ais # optional, default: source_dir; exposed to command as {output_dir}
    pattern: "*.csv"                # optional, default "*.csv"
    quiet_period: 1800              # optional, seconds, default 1800
    processed_dir: /data/incoming/ais/processed   # optional, default {source_dir}/processed
    failed_dir: /data/incoming/ais/failed          # optional, default {source_dir}/failed
    command:                        # required, argv list - no shell
      - damast
      - process
      - --pipeline
      - /pipelines/ais.damast.ppl
      - --input-data
      - "{input}"
      - --output-file
      - "{output_dir}/{stem}.parquet"

  - name: osint-daily
    source_dir: /data/incoming/osint
    command:
      - damast
      - convert
      - -f
      - "{input}"
      - -m
      - /specs/osint.spec.yaml
      - -o
      - "{output_dir}/{stem}.parquet"

Each token of command is substituted with str.format, so a job’s command can use these placeholders:

--create-config builds a config interactively instead of hand-writing the YAML, prompting for each job’s fields (blank answers accept the default shown in brackets) and writing the result to the file given via --config:

$ damast watch --config watch.yaml --create-config
--- job 1 ---
Source directory to watch: /data/incoming/ais
Job name [ais]:
Target directory (available to the command as {output_dir}) [/data/incoming/ais]: /data/processed/ais
File pattern [*.csv]:
Quiet period in seconds before a file is considered complete [1800]:
Command to run on each ready file (use {input}/{output_dir}/{stem}/{name}): damast process --pipeline /pipelines/ais.damast.ppl --input-data {input} --output-file {output_dir}/{stem}.parquet
Add another job? [y/N]: n
Wrote 1 job(s) to 'watch.yaml'

The command line is split the same way a shell would (via shlex.split) into the argv list the config stores - no shell is ever invoked. Running --create-config against a file that already exists asks whether to append the new job(s) to it or overwrite it.

Environment variables and home directory

source_dir, target_dir, processed_dir, failed_dir and every token of command accept ${VARNAME} and ~/{home}, expanded to the named environment variable and the current user’s home directory respectively:

jobs:
  - name: ais-daily
    source_dir: ${DATA_ROOT}/incoming/ais
    target_dir: ~/damast-watch/ais
    command:
      - damast
      - convert
      - -f
      - "{input}"
      - -m
      - "{home}/specs/ais.spec.yaml"
      - -o
      - "{output_dir}/{stem}.parquet"

Expansions run at different times, which matters for how a missing variable shows up:

• source_dir/target_dir/processed_dir/failed_dir are expanded once, when the job is loaded - raising an exception when encountering an unset variable

• command tokens are expanded for every ready file.

Progress

While a job runs, watch shows a live tqdm progress bar naming the job, the file currently being processed, and the log to check on it:

[ais-daily] 2026-08-30.csv:  45%|####5     | 5/11 [00:03<00:04,  1.32file/s, log: /data/processed/ais/2026-08-30.log]

The command’s own stdout/stderr is streamed live into that <target_dir>/<stem>.log file - tail -f it to follow a long-running command - and mirrored to the damast.core.watch logger at DEBUG level. Neither is printed to the console at the default INFO level, so the progress bar stays a single, clean line; pass --log-level DEBUG to see that output inline instead. On failure, the same log path is named in the raised error and, for the moved file, duplicated into <failed_dir>/<file>.error.log alongside the traceback.

Examples

damast watch --config watch.yaml

# restrict to one job, e.g. for a tighter cron schedule
damast watch --config watch.yaml --job ais-daily

# list which files are ready without running anything
damast watch --config watch.yaml --dry-run

A non-zero exit code means at least one job or file failed - useful for cron/systemd failure alerting. watch does not interpret the configured command’s output; check a job’s failed_dir for the moved input and its .error.log (captured stdout/stderr and exit code).

Deployment

Since damast watch performs a single scan, recurring execution is the deployment’s responsibility. A crontab entry, scanning every 15 minutes:

*/15 * * * * damast watch --config /etc/damast/watch.yaml

Or a systemd service/timer pair:

# /etc/systemd/system/damast-watch.service
[Unit]
Description=damast watch

[Service]
Type=oneshot
ExecStart=/usr/local/bin/damast watch --config /etc/damast/watch.yaml

# /etc/systemd/system/damast-watch.timer
[Unit]
Description=Run damast-watch periodically

[Timer]
OnCalendar=*:0/15
Persistent=true

[Install]
WantedBy=timers.target

Enable with systemctl enable --now damast-watch.timer; OnFailure= on the service can point at an alerting unit, since the non-zero exit code on any job/file failure is preserved.

Plugins

usage: damast plugins [-h] [-w WORKDIR] [-v]
                      [--log-level {CRITICAL,FATAL,ERROR,WARN,WARNING,INFO,DEBUG,NOTSET}]
                      [--log-file LOG_FILE] [--version]
                      {} ...

damast plugins - list transformer plugins registered by installed packages
(entry-point group 'damast.transformers') or via the 'DAMAST_PLUGIN_PATH'
environment variable

positional arguments:
  {}                    sub-command help

options:
  -h, --help            show this help message and exit
  -w WORKDIR, --workdir WORKDIR
  -v, --verbose
  --log-level {CRITICAL,FATAL,ERROR,WARN,WARNING,INFO,DEBUG,NOTSET}
                        Set loglevel to display
  --log-file LOG_FILE   Set file for saving log (default prints to terminal)
  --version             Show current version of damast

A pipeline can use transformers that are not part of the damast package itself. Two ways of registering such plugin transformers are supported - see damast.core.transformations.PluginManager for the full API:

• installable packages that declare their damast.core.transformations.PipelineElement subclasses via the damast.transformers entry-point group in their own pyproject.toml - either one entry per class, or one entry per module, which registers every transformer defined in that module (or, for a package, in its top-level submodules):

  [project.entry-points."damast.transformers"]
  MyTransformer = "acme_pkg.transformers:MyTransformer"
  acme_pkg = "acme_pkg.transformers"
  

• local directories that are not part of any installed package, made discoverable via the DAMAST_PLUGIN_PATH environment variable (several entries separated with os.pathsep). An entry name=path loads the directory as a package called name: its top-level *.py files become name.<file>, and may use relative imports (from .helpers import x), also into subdirectories. The same can be done in code via plugin_manager.register_plugin_package(name, path). Choose a distinctive name - one that is already importable is rejected. A bare path entry (deprecated) loads each file as a flat module named after the file, without relative imports.

The package name of a local directory is recorded in saved pipelines (e.g. module_name: my_plugins.my_transformers), so a pipeline can only be replayed with the directory registered under the same name - or with an installed package of that name, e.g. once the directory has been turned into an installable package with a module entry-point.

Regardless of which of the two a transformer comes from, it is resolvable in code the same way, via damast.plugins.<package> - where <package> is the top-level package of the module defining it, i.e. acme_pkg for acme_pkg.transformers:MyTransformer, or the name a local directory was registered under:

from damast.plugins.acme_pkg import MyTransformer

Scoping by package means two plugins can provide a transformer of the same name without clashing. damast.plugins resolves names lazily on first access, so nothing beyond the requested class is ever imported - see damast.plugins for details.

Example: a local plugin transformer

A local, ad-hoc transformer is just a damast.core.transformations.PipelineElement subclass in a loose *.py file, written like any other transformer:

import polars

from damast.core.dataframe import AnnotatedDataFrame
from damast.core.decorators import describe, input, output
from damast.core.transformations import PipelineElement


class MyTripler(PipelineElement):
    """Multiplies a column by 3 - a minimal example of a local plugin transformer."""

    @describe("Triples a column")
    @input({"x": {}})
    @output({"{{x}}_tripled": {}})
    def transform(self, df: AnnotatedDataFrame) -> AnnotatedDataFrame:
        feature = self.get_name("x")
        result = self.get_name("{{x}}_tripled")
        df.lazyframe = df.lazyframe.with_columns(
            (polars.col(feature) * 3).alias(result)
        )
        return df

With DAMAST_PLUGIN_PATH registering the directory containing that file (here as package my_plugins), damast plugins lists it without requiring any further Python code:

$ export DAMAST_PLUGIN_PATH=my_plugins=./examples/plugins
$ damast plugins

my_plugins (local: examples/plugins)
    JoinByTimestamp  .osint_ais_transformers
    MyTripler        .my_transformers

MyTripler is now resolvable via damast.plugins.my_plugins and can be used in a pipeline like any other transformer:

import os
from pathlib import Path

# DAMAST_PLUGIN_PATH registers the directory holding my_transformers.py as package
# 'my_plugins', so that 'MyTripler' becomes resolvable via damast.plugins.my_plugins - see
# docs/examples/plugins/.
os.environ["DAMAST_PLUGIN_PATH"] = f"my_plugins={Path(__file__).parent / 'plugins'}"

from damast.core import DataProcessingPipeline
from damast.plugins.my_plugins import MyTripler

pipeline = DataProcessingPipeline(name="my-plugin-pipeline", base_dir=".")
pipeline.add("Triple mmsi",
             MyTripler(),
             name_mappings={"x": "mmsi"})

pipeline.save("pipelines")

The resulting pipeline can be applied like any other, e.g. via damast process (see Process above), as long as DAMAST_PLUGIN_PATH still registers the directory containing my_transformers.py as my_plugins:

damast process --input-data data.parquet --pipeline pipelines/my-plugin-pipeline.damast.ppl

Pipelines saved with a plugin transformer record where it came from under requires (the installed distribution and version, or the local package name and directory), so that loading the pipeline elsewhere fails with an actionable message - naming the missing package to pip install, or the DAMAST_PLUGIN_PATH entry to add - instead of a bare import error.