SignalK Parquet Data Store

Vessel data Parquet file archive with automated value and geospatial triggers. History API compliant with cloud backups and queries.

Features

Core Data Management

• Smart Data Types: Intelligent Parquet schema detection preserves native data types (DOUBLE, BOOLEAN) instead of forcing everything to strings
• Daily Export Pipeline: Simplified daily export creates consolidated Parquet files directly
  • Data accumulates in SQLite buffer throughout the day
  • Single export at configurable hour (default: 4 AM UTC)
  • No separate consolidation step needed
• SQLite WAL Buffering: Crash-safe data ingestion with Write-Ahead Logging
  • Replaces in-memory buffers with persistent SQLite database
  • Automatic recovery after power loss or crashes
  • 48-hour retention for federated queries
  • Per-path tables (buffer_navigation_position, etc.) for partition-aligned access
  • buffer_tables metadata table tracks path→table mapping
• Hive-Partitioned Storage: Efficient file organization for query performance
  • Structure: tier=raw/context={ctx}/path={path}/year={year}/day={day}/
  • Aggregation tiers: raw, 5s, 60s, 1h
  • Automatic partition pruning for time-range queries
  • Bulk Aggregation: POST /api/aggregate/bulk builds all tiers from raw data in one background job
  • Post-Migration Aggregation: Migration automatically builds tiers after moving files to hive structure
  • Position Aggregation Migration: POST /api/migrate/position-aggregation re-aggregates position paths (value_latitude/value_longitude) into 5s/60s/1h tiers
    • Auto-detects position paths in the raw tier by schema
    • GPS outlier rejection via POSITION_MAX_SPEED_MPS (25 m/s ≈ 48.6 kn) to discard single-point glitches
    • Supports dryRun for previewing
• Batched Parquet Writing: Streams records to Parquet in pull-based batches instead of loading all rows into memory
  • Reduces peak memory during large exports from the SQLite buffer
• Cloud Federated Querying: Query historical data directly from S3 or Cloudflare R2 using DuckDB's native support
  • Three-tier query hierarchy: local parquet → cloud supplement → SQLite buffer
  • Automatic partition pruning reduces data transfer by 70-90%
  • Predicate pushdown filters data at source before transfer
• Auto-Discovery: Automatically configure paths when first queried
  • On-demand path configuration when History API queries unconfigured paths
  • Include/exclude glob patterns for fine-grained control
  • Optional live data requirement before configuration
• Vector Averaging: Correct aggregation of angular data (headings, bearings, wind angles)
  • Automatic detection of angular paths via SignalK metadata (units === 'rad')
  • Uses atan2(mean(sin), mean(cos)) instead of arithmetic mean
  • Lossless re-aggregation across tiers via stored sin/cos averages
• Buffer Bucketing: SQLite buffer data bucketed to match Parquet query resolution
  • Prevents raw per-second records from flooding bucketed query results
  • Supports all aggregate methods including vector averaging for angular paths
• Dynamic Regimen Control: Data subscriptions update in real-time when regimen commands are received
  • No plugin restart needed to activate/deactivate recording regimens
  • Regimen state changes via SignalK commands immediately start/stop path recording
• GPX Track Import: Load historical GPX tracks (other vessels, handhelds, archived logs) directly into the Hive-partitioned parquet store, bypassing the live SignalK subscription path
  • Drag-and-drop browser upload from the Status tab, or "Advanced" server-directory mode for USB-drive bulk imports on the host
  • Dependency-free GPX 1.0 / 1.1 parser extracts <trkpt> lat/lon/time plus optional <ele>, <speed>, <course>; <course> is converted from degrees to radians to match navigation.courseOverGroundTrue
  • Job-based with per-jobId cancellation, progress polling, and 30-minute TTL on finished jobs
  • Browser upload caps: 50 MB per file, 500 files per request

Data Validation & Schema Repair

• Schema Validation: Comprehensive validation of Parquet file schemas against SignalK metadata standards
• Automated Repair: One-click repair of schema violations with proper data type conversion
• Type Correction: Automatic conversion of incorrectly stored data types (e.g., numeric strings → DOUBLE, boolean strings → BOOLEAN)
• Metadata Integration: Uses SignalK metadata (units, types) to determine correct data types for marine measurements
• Safe Operations: Creates backups before repair and quarantines corrupted files for safety
• Progress Tracking: Real-time progress monitoring with cancellation support for large datasets

Benefits of Proper Data Types

Using correct data types in Parquet files provides significant advantages:

• Storage Efficiency: Numeric data stored as DOUBLE uses ~50% less space than string representations
• Query Performance: Native numeric operations are 5-10x faster than string parsing during analysis
• Data Integrity: Type validation prevents data corruption and ensures consistent analysis results
• Analytics Compatibility: Proper types enable advanced statistical analysis and machine learning applications
• Compression: Parquet's columnar compression works optimally with correctly typed data

Validation Process

The validation system checks each Parquet file for:

• Field Type Consistency: Ensures numeric marine data (position, speed, depth) is stored as DOUBLE
• Boolean Representation: Validates true/false values are stored as BOOLEAN, not strings
• Metadata Alignment: Compares file schemas against SignalK metadata for units like meters, volts, amperes
• Schema Standards: Enforces data best practices for long-term data integrity

Advanced Querying

• SignalK History API Compliance: Full compliance with SignalK History API specifications
  • Standard Time Parameters: All 5 standard query patterns supported
  • Time-Filtered Discovery: Paths and contexts filtered by time range using hive partition directory names (no file scanning)
  • Optional Analytics: Moving averages (EMA/SMA) available on demand
• SignalK Track API Provider (preview, v0.7.44-beta.4+): registers as a provider for the server's upcoming Track API (SignalK/signalk-server#2995) — vessel tracks as GeoJSON from the parquet store and live buffer. Inert on servers without the API; see Track API Integration
• 🌍 ISO 8601 Timestamps: All timestamps returned in server local time with offset (e.g., 2025-10-20T12:34:04-04:00)
• Flexible Time Querying: Multiple ways to specify time ranges
  • Query from now, from specific times, or between time ranges
  • Duration-based windows (1h, 30m, 2d) for easy relative queries
  • Forward and backward time querying support
• Time Alignment: Automatic alignment of data from different sensors using time bucketing
• DuckDB Integration: Direct SQL querying of Parquet files with type-safe operations
• Spatial Analysis: Advanced geographic analysis with DuckDB spatial extension
  • Track Analysis: Calculate vessel tracks, distances, and movement patterns
  • Proximity Detection: Multi-vessel distance calculations and collision risk analysis
  • Geographic Visualization: Generate movement boundaries, centroids, and spatial statistics
  • Route Planning: Historical track analysis for route optimization and performance analysis
  • Spatial Correlation: Filter any sensor data by vessel location
    • Query "wind data when vessel was within this area"
    • Bounding box (bbox) and radius filters work on all paths
    • Automatically correlates timestamps with position data
  • Spatial Context Discovery: Find vessels by geographic area
    • GET /api/history/contexts/spatial?bbox=... or radius=...
    • Single DuckDB query on navigation__position files with hive partition pruning

Management & Control

• Command Management: Register, execute, and manage SignalK commands with automatic path configuration
• Regimen-Based Data Collection: Control data collection with command-based regimens
• Multi-Vessel Support: Wildcard vessel contexts (vessels.*) with MMSI-based exclusion filtering
  • Context-Aware Path Discovery: /history/paths?context=vessels.urn:mrn:imo:mmsi:NNNNN returns paths for any tracked vessel
• Source Filtering: Filter data by SignalK source labels (bypasses server arbitration for raw data access)
• Comprehensive REST API: Full programmatic control of queries and configuration

User Interface & Integration

• Responsive Web Interface: Complete web-based management interface
• Map Explorer: Interactive spatial query and visualization tab
  • Draw bounding box or radius areas on a Leaflet map to query vessel data geographically
  • Multi-path overlay (up to 3 paths) with color-coded track, chart, and data table
  • Searchable path selector with checkbox list and removable chips
  • Display values in user-configured units via SignalK metadata (displayUnits)
  • Playback controls with scrub slider for time-series review
  • Export to CSV, GeoJSON, KML; save as SignalK waypoints, tracks, or routes
  • OpenSeaMap nautical chart overlay
  • Save/load named geographic areas with server sync via SignalK Resources API (shared with ZedDisplay)
  • High-resolution track option for maximum route fidelity when saving routes
  • Vessel name display in context dropdown with searchable filter
  • Auto-selects first data point after query for immediate detail view
• Cloud Storage Integration: Upload files to Amazon S3 or Cloudflare R2 as part of the daily export pipeline
• Context Support: Support for multiple vessel contexts with exclusion controls

Regimen System (Advanced)

• Operational Context Tracking: Define regimens for operational states (mooring, anchoring, racing, passage-making)
• Command-Based Episodes: Track state transitions using SignalK commands as regimen triggers
• Episode Boundary Detection: Sophisticated SQL-based detection of operational periods using CTEs and window functions
• Contextual Data Collection: Link SignalK paths to regimens for targeted data analysis during specific operations
• Web Interface Management: Create, edit, and manage regimens and command keywords through the web UI

Threshold Automation

• Per-Command Conditions: Each regimen/command can define one or more thresholds that watch a single SignalK path.

• True-Only Actions: On every path update the condition is evaluated; when it is true the command is set to the threshold's activateOnMatch state (ON/OFF). False evaluations leave the command untouched, so use a second threshold if you want a different level to switch it back.

• Stable Triggers: Optional hysteresis (seconds) suppresses re-firing while the condition remains true, preventing rapid toggling in noisy data.

• Multiple Thresholds Per Path: Unique monitor keys allow several thresholds to observe the same SignalK path without cancelling each other.

• Unit Handling: Threshold values must match the live SignalK units (e.g., fractional 0–1 SoC values). Angular thresholds are entered in degrees in the UI and stored as radians automatically.

• Automation State Machine: When enabling automation, command is set to OFF then all thresholds are immediately evaluated. When disabling automation, threshold monitoring stops and command state remains unchanged. Default state is hardcoded to OFF on server side.

• Custom Analysis: Create custom analysis prompts for specific operational needs

Requirements

Core Requirements

• SignalK Server v2.13+
• Node.js 22.5+ (required for node:sqlite — the built-in SQLite module used for crash-safe buffering; on Node < 22.5 the buffer falls back to in-memory LRU)

Installation

Install from GitHub

# Navigate to folder
cd ~/.signalk/node_modules/

# Install from npm (after publishing)
npm install signalk-parquet

# Or install from GitHub
npm install motamman/signalk-parquet
cd ~/.signalk/node_modules/signalk-parquet
npm run build

# Restart SignalK
sudo systemctl restart signalk

Development Setup

# Clone or copy the signalk-parquet directory
cd signalk-parquet

# Install dependencies
npm install

# Build the TypeScript code
npm run build

# Copy to SignalK plugins directory
cp -r . ~/.signalk/node_modules/signalk-parquet/

# Restart SignalK
sudo systemctl restart signalk

Production Build

# Build for production
npm run build

# The compiled JavaScript will be in the dist/ directory

Configuration

Plugin Configuration

Navigate to SignalK Admin → Server → Plugin Config → SignalK Parquet Data Store

Configure basic plugin settings (path configuration is managed separately in the web interface):

Setting	Description	Default
Buffer Size	Number of records to buffer before writing	1000
Save Interval	How often to save buffered data (seconds)	30
Output Directory	Directory to save data files	SignalK data directory
Filename Prefix	Prefix for generated filenames	signalk_data
File Format	Output format (parquet, json, csv)	parquet
Retention Days	Days to keep processed files	7
Daily Export Hour	Hour (0-23 UTC) to run daily Parquet export	4
Export Batch Size	Max records to export per cycle (1,000-200,000)	50000
Buffer Retention Hours	How long to keep exported records in SQLite (hours)	48
Enable Raw SQL	Enable /api/query endpoint for raw SQL queries	false

Auto-Discovery Configuration

Configure automatic path discovery when querying unconfigured paths:

Setting	Description	Default
Enable Auto-Discovery	Master switch for auto-discovery	false
Require Live Data	Only configure if path has live SignalK data	true
Max Auto-Configured Paths	Maximum number of auto-configured paths	100
Include Patterns	Glob patterns for paths to include (e.g., navigation.*)	[]
Exclude Patterns	Glob patterns for paths to exclude (e.g., propulsion.*)	[]

When enabled, Auto-Discovery will automatically add path configurations when:

1. A History API query requests data for an unconfigured path
2. The path matches include patterns (if specified)
3. The path doesn't match exclude patterns
4. The path has live data in SignalK (if requireLiveData is enabled)

Auto-discovered paths are marked with the autoDiscovered: true flag and have auto-generated human-readable names prefixed with [Auto].

Cloud Upload Configuration

Configure cloud storage upload in the plugin configuration. Uploads run as part of the daily export pipeline.

Setting	Description	Default
Provider	Cloud provider: none, s3, or r2	none
Bucket	Bucket name	-
Region	AWS region (S3 only)	us-east-1
Account ID	Cloudflare account ID (R2 only)	-
Key Prefix	Object key prefix	-
Access Key ID	Cloud credentials	-
Secret Access Key	Cloud credentials	-
Custom Endpoint URL	Override the S3 endpoint for self-hosted S3-compatible storage (Garage, MinIO). Include protocol and port, e.g. https://garage.example.com:3900 (S3 only)	-
Use Path-Style Addressing	Path-style bucket addressing (https://endpoint/bucket); often required by self-hosted services. Defaults to enabled when a custom endpoint is set	auto
Allow Private/Local Endpoints	Permit custom endpoints on private/loopback/link-local addresses (e.g. 192.168.x.x, localhost). Required for self-hosted storage on the boat LAN; off by default to prevent SSRF (since v0.7.44-beta.2)	false
Delete After Upload	Delete local files after upload	false

Upload timeouts (v0.7.44+): cloud requests are bounded (10 s to connect, 60 s per request) so a dead or stalled uplink fails the upload — which is retried — instead of hanging the daily export pipeline.

Path Configuration

Important: Path configuration is managed exclusively through the web interface, not in the SignalK admin interface. This provides a more intuitive interface for managing data collection paths.

Accessing Path Configuration

1. Navigate to: http://localhost:3000/plugins/signalk-parquet
2. Click the ⚙️ Path Configuration tab

Adding Data Paths

Use the web interface to configure which SignalK paths to collect:

1. Click ➕ Add New Path
2. Configure the path settings:
   • SignalK Path: The SignalK data path (e.g., navigation.position)
   • Always Enabled: Collect data regardless of regimen state
   • Regimen Control: Command name that controls collection
   • Source Filter: Only collect from specific sources
   • Context: SignalK context (vessels.self, vessels.*, or specific vessel)
   • Exclude MMSI: For vessels.* context, exclude specific MMSI numbers
3. Click ✅ Add Path

Managing Existing Paths

• Edit Path: Click ✏️ Edit button to modify path settings
• Delete Path: Click 🗑️ Remove button to delete a path
• Refresh: Click 🔄 Refresh Paths to reload configuration
• Show/Hide Commands: Toggle button to show/hide command paths in the table

Command Management

The plugin streamlines command management with automatic path configuration:

1. Register Command: Commands are automatically registered with enabled path configurations
2. Start Command: Click Start button to activate a command regimen
3. Stop Command: Click Stop button to deactivate a command regimen
4. Remove Command: Click Remove button to delete a command and its path configuration

This eliminates the previous 3-step process of registering commands, adding paths, and enabling them separately.

Path Configuration Storage

Path configurations are stored separately from plugin configuration in:

~/.signalk/signalk-parquet/webapp-config.json

This allows for:

• Independent management of path configurations
• Better separation of concerns
• Easier backup and migration of path settings
• More intuitive web-based configuration interface

Regimen-Based Control

Regimens allow you to control data collection based on SignalK commands:

Example: Weather data collection with source filtering

{
  "path": "environment.wind.angleApparent",
  "enabled": false,
  "regimen": "captureWeather",
  "source": "mqtt-weatherflow-udp",
  "context": "vessels.self"
}

Note: Source filtering accesses raw data before SignalK server arbitration, allowing collection of data from specific sources that might otherwise be filtered out.

Multi-Vessel Example: Collect navigation data from all vessels except specific MMSI numbers

{
  "path": "navigation.position",
  "enabled": true,
  "context": "vessels.*",
  "excludeMMSI": ["123456789", "987654321"]
}

Command Path: Command paths are automatically created when registering commands

{
  "path": "commands.captureWeather",
  "enabled": true,
  "context": "vessels.self"
}

This path will only collect data when the command commands.captureWeather is active.

TypeScript Architecture

Type Safety

The plugin uses comprehensive TypeScript interfaces:

interface PluginConfig {
  bufferSize: number;
  saveIntervalSeconds: number;
  outputDirectory: string;
  filenamePrefix: string;
  fileFormat: 'json' | 'csv' | 'parquet';
  retentionDays: number;
  vesselMMSI: string;
  cloudUpload: CloudUploadConfig;
  useSqliteBuffer?: boolean;
  exportBatchSize?: number;
  bufferRetentionHours?: number;
  useHivePartitioning?: boolean;
  dailyExportHour?: number;
  autoDiscovery?: AutoDiscoveryConfig;
  enableRawSql?: boolean;
}

interface CloudUploadConfig {
  provider: 'none' | 's3' | 'r2';
  bucket?: string;
  region?: string;       // S3 only
  accountId?: string;    // R2 only (Cloudflare account ID)
  keyPrefix?: string;
  accessKeyId?: string;
  secretAccessKey?: string;
  deleteAfterUpload?: boolean;
}

interface PathConfig {
  path: string;
  name?: string;
  enabled?: boolean;
  regimen?: string;
  source?: string;
  context?: string;
  excludeMMSI?: string[];
  autoDiscovered?: boolean;
}

interface DataRecord {
  received_timestamp: string;
  signalk_timestamp: string;
  context: string;
  path: string;
  value: any;
  source_label?: string;
  meta?: string;
}

Plugin State Management

The plugin maintains typed state:

interface PluginState {
  unsubscribes: Array<() => void>;
  dataBuffers: LRUCache<string, DataRecord[]>;
  activeRegimens: Set<string>;
  subscribedPaths: Set<string>;
  parquetWriter?: ParquetWriter;
  cloudClient?: any;
  currentConfig?: PluginConfig;
  sqliteBuffer?: SQLiteBufferInterface;
  exportService?: ParquetExportServiceInterface;
  commandState: CommandRegistrationState;
}

Express Router Types

API routes are fully typed:

router.get('/api/paths', 
  (_: TypedRequest, res: TypedResponse<PathsApiResponse>) => {
    // Typed request/response handling
  }
);

Data Output Structure

File Organization (Hive-Partitioned)

The plugin uses Hive-style partitioned paths for efficient querying:

output_directory/
├── tier=raw/
│   ├── context=vessels__self/
│   │   ├── path=navigation__position/
│   │   │   ├── year=2025/
│   │   │   │   ├── day=197/
│   │   │   │   │   ├── data_20250716T120000.parquet
│   │   │   │   │   └── data_20250716T130000.parquet
│   │   │   │   └── day=198/
│   │   │   │       └── data_20250717T080000.parquet
│   │   │   └── year=2024/
│   │   │       └── day=365/
│   │   └── path=navigation__speedOverGround/
│   └── context=vessels__urn-mrn-imo-mmsi-368396230/
│       └── path=navigation__position/
├── tier=5s/
│   └── [aggregated 5-second data]
├── tier=60s/
│   └── [aggregated 1-minute data]
├── tier=1h/
│   └── [aggregated hourly data]
├── buffer.db              <- SQLite WAL buffer (48h retention)
└── buffer.db-wal          <- Write-ahead log

Partition Structure:

• tier= - Aggregation level: raw, 5s, 60s, 1h
• context= - Vessel context (sanitized: . → __, : → -)
• path= - SignalK path (sanitized: . → __)
• year= - Year (e.g., 2025)
• day= - Day of year, zero-padded (e.g., 197)

Legacy flat structure: If you have data from pre-Hive versions, use the Migration API to convert to Hive partitioning. See Data Migration.

Data Migration

Migrating Legacy Files to Hive Partitioning

If you have existing data in the legacy flat structure, use the Migration API to convert to Hive partitioning:

1. Scan for migratable files:

curl -X POST http://localhost:3000/plugins/signalk-parquet/api/migrate/scan \
  -H "Content-Type: application/json" \
  -d '{"sourceDirectory": "/path/to/data"}'

Response includes:

• Total files to migrate
• Total size in bytes
• Files grouped by SignalK path
• Estimated migration time

2. Start migration:

curl -X POST http://localhost:3000/plugins/signalk-parquet/api/migrate \
  -H "Content-Type: application/json" \
  -d '{
    "sourceDirectory": "/path/to/data",
    "targetDirectory": "/path/to/data",
    "targetTier": "raw",
    "deleteSourceAfterMigration": false
  }'

3. Check progress:

curl http://localhost:3000/plugins/signalk-parquet/api/migrate/progress/{jobId}

4. Cancel if needed:

curl -X POST http://localhost:3000/plugins/signalk-parquet/api/migrate/cancel/{jobId}

Migration Options:

Option	Description	Default
sourceDirectory	Source directory to scan	Plugin data directory
targetDirectory	Target directory for Hive files	Same as source
targetTier	Target aggregation tier	raw
deleteSourceAfterMigration	Delete source files after successful migration	false

Data Schema

Each record contains:

Field	Type	Description
received_timestamp	string	When the plugin received the data
signalk_timestamp	string	Original SignalK timestamp
context	string	SignalK context (e.g., vessels.self)
path	string	SignalK path
value	DOUBLE/BOOLEAN/INT64/UTF8	Smart typed values - numbers stored as DOUBLE, booleans as BOOLEAN, etc.
value_json	string	JSON representation for complex values
source	string	Complete source information
source_label	string	Source label
source_type	string	Source type
source_pgn	number	PGN number (if applicable)
meta	string	Metadata information

Aggregated Tier Schema (5s, 60s, 1h)

Aggregated tiers use a different schema optimized for statistical queries:

Field	Type	Description
bucket_time	TIMESTAMP	Start of the aggregation time bucket
context	UTF8	SignalK context
path	UTF8	SignalK path
value_avg	DOUBLE	Average value in bucket
value_min	DOUBLE	Minimum value (NULL for angular paths)
value_max	DOUBLE	Maximum value (NULL for angular paths)
sample_count	INT64	Number of raw samples aggregated
value_sin_avg	DOUBLE	Average of sin(value) — angular paths only, for lossless re-aggregation
value_cos_avg	DOUBLE	Average of cos(value) — angular paths only, for lossless re-aggregation

Smart Data Types

The plugin now intelligently detects and preserves native data types:

• Numbers: Stored as DOUBLE (floating point) or INT64 (integers)
• Booleans: Stored as BOOLEAN
• Strings: Stored as UTF8
• Objects: Serialized to JSON and stored as UTF8
• Mixed Types: Falls back to UTF8 when a path contains multiple data types

This provides better compression, faster queries, and proper type safety for data analysis.

Web Interface

Features

• Path Configuration: Manage data collection paths with multi-vessel support
• Command Management: Streamlined command registration and control
• Data Exploration: Browse available data paths
• SQL Queries: Execute DuckDB queries against Parquet files
• History API: Query historical data using SignalK History API endpoints
• Cloud Status: Test cloud storage connectivity and configuration
• Responsive Design: Works on desktop and mobile
• MMSI Filtering: Exclude specific vessels from wildcard contexts

API Endpoints

Endpoint	Method	Description
/api/paths	GET	List available data paths
/api/files/:path	GET	List files for a path
/api/sample/:path	GET	Sample data from a path
/api/query	POST	Execute SQL query (⚠️ disabled by default; enabled when either the Enable Raw SQL plugin setting is on or the SIGNALK_PARQUET_RAW_SQL=true environment variable is set — one gate is sufficient, both are not required). Runs on a sandboxed DuckDB instance (v0.7.44+): no network access, no cloud credentials, reads restricted to the data directory. From v0.7.44-beta.4 the SQL must be a single read-only statement (ATTACH/COPY/EXPORT/SET/PRAGMA and file- or database-opening table functions are rejected with a 400 naming the reason), the sandbox configuration is locked, and results are capped at 10,000 rows with truncated: true in the response when the cap applied
/api/config/paths	GET/POST/PUT/DELETE	Manage path configurations
/api/test-cloud	POST	Test cloud storage connection
/api/health	GET	Health check
/api/version	GET	Plugin version
/api/store/stats	GET	Data store statistics
SignalK History API
/signalk/v1/history/values	GET	SignalK History API - Get historical values
/signalk/v1/history/contexts	GET	SignalK History API - Get available contexts
/api/history/contexts/spatial	GET	Get contexts with position data in bbox or radius
/signalk/v1/history/paths	GET	SignalK History API - Get available paths
/signalk/v2/api/history/*	GET	SignalK v2 API - handled by registered HistoryApi provider (spec-compliant)
Migration API
/api/migrate/scan	POST	Scan directory for migratable files
/api/migrate	POST	Start migration job
/api/migrate/progress/:jobId	GET	Get migration job progress
/api/migrate/cancel/:jobId	POST	Cancel running migration job
/api/migrate/jobs	GET	List all migration jobs
Buffer Status API
/api/buffer/stats	GET	Get SQLite buffer statistics
/api/buffer/export	POST	Force immediate export of pending records
/api/buffer/health	GET	Get buffer health status
Validation & Repair API
/api/validate-schemas	POST	Scan and validate Parquet schemas
/api/validate-schemas/progress/:jobId	GET	Get validation progress
/api/validate-schemas/cancel/:jobId	POST	Cancel validation job
/api/repair-schemas	POST	Repair schema violations
/api/repair-schemas/progress/:jobId	GET	Get repair progress
/api/repair-schemas/cancel/:jobId	POST	Cancel repair job
Vector Averaging Migration
/api/migrate/vector-averaging	POST	Migrate to vector averaging aggregation
/api/migrate/vector-averaging/:jobId	GET	Get migration progress
/api/migrate/vector-averaging/cancel/:jobId	POST	Cancel migration
Aggregation API
/api/aggregate	POST	Trigger aggregation for all tiers
/api/aggregate/:sourceTier/:targetTier	POST	Aggregate specific tier pair
Cloud Sync API
/api/cloud/compare	POST	Compare local vs cloud data
/api/cloud/compare/:jobId	GET	Get comparison progress
/api/cloud/sync	POST	Sync data to cloud
/api/cloud/sync/:jobId	GET	Get sync progress

DuckDB Integration

Query Examples

Globbing the Hive-partitioned store: paths under tier=.../context=.../path=.../ are partitioned as year=*/day=*/. Glob them with year=*/day=*/*.parquet — not **/*.parquet. The recursive ** descends into the sibling quarantine/, failed/, processed/, and repaired/ directories, and DuckDB will abort the whole query if it hits a quarantined 0-byte file (too small to be a Parquet file). The Query Database "Generate Query" button produces the correct glob automatically.

Basic Queries

-- Get latest 10 records from navigation position
SELECT * FROM read_parquet('/path/to/navigation/position/*.parquet', union_by_name=true)
ORDER BY received_timestamp DESC LIMIT 10;

-- Count total records
SELECT COUNT(*) FROM read_parquet('/path/to/navigation/position/*.parquet', union_by_name=true);

-- Filter by source
SELECT * FROM read_parquet('/path/to/environment/wind/*.parquet', union_by_name=true)
WHERE source_label = 'mqtt-weatherflow-udp'
ORDER BY received_timestamp DESC LIMIT 100;

-- Aggregate by hour
SELECT
  DATE_TRUNC('hour', received_timestamp::timestamp) as hour,
  AVG(value::double) as avg_value,
  COUNT(*) as record_count
FROM read_parquet('/path/to/data/*.parquet', union_by_name=true)
GROUP BY hour
ORDER BY hour;

🌍 Spatial Analysis Queries

-- Calculate distance traveled over time
WITH ordered_positions AS (
  SELECT
    signalk_timestamp,
    ST_Point(value_longitude, value_latitude) as position,
    LAG(ST_Point(value_longitude, value_latitude)) OVER (ORDER BY signalk_timestamp) as prev_position
  FROM read_parquet('data/vessels/urn_mrn_imo_mmsi_368396230/navigation/position/*.parquet', union_by_name=true)
  WHERE signalk_timestamp >= '2025-09-27T16:00:00Z'
    AND signalk_timestamp <= '2025-09-27T23:59:59Z'
    AND value_latitude IS NOT NULL AND value_longitude IS NOT NULL
),
distances AS (
  SELECT *,
    CASE
      WHEN prev_position IS NOT NULL
      THEN ST_Distance_Sphere(position, prev_position)
      ELSE 0
    END as distance_meters
  FROM ordered_positions
)
SELECT
  strftime(date_trunc('hour', signalk_timestamp::TIMESTAMP), '%Y-%m-%dT%H:%M:%SZ') as time_bucket,
  AVG(value_latitude) as avg_lat,
  AVG(value_longitude) as avg_lon,
  ST_AsText(ST_Centroid(ST_Collect(position))) as centroid,
  SUM(distance_meters) as total_distance_meters,
  COUNT(*) as position_records,
  ST_AsText(ST_ConvexHull(ST_Collect(position))) as movement_area
FROM distances
GROUP BY time_bucket
ORDER BY time_bucket;

-- Multi-vessel proximity analysis
SELECT
  v1.context as vessel1,
  v2.context as vessel2,
  ST_Distance_Sphere(
    ST_Point(v1.value_longitude, v1.value_latitude),
    ST_Point(v2.value_longitude, v2.value_latitude)
  ) as distance_meters,
  v1.signalk_timestamp
FROM read_parquet('data/vessels/*/navigation/position/*.parquet', union_by_name=true) v1
JOIN read_parquet('data/vessels/*/navigation/position/*.parquet', union_by_name=true) v2
  ON v1.signalk_timestamp = v2.signalk_timestamp AND v1.context != v2.context
WHERE v1.signalk_timestamp >= '2025-09-27T00:00:00Z'
  AND ST_Distance_Sphere(
    ST_Point(v1.value_longitude, v1.value_latitude),
    ST_Point(v2.value_longitude, v2.value_latitude)
  ) < 1000  -- Within 1km
ORDER BY distance_meters;

-- Advanced movement analysis with bounding boxes
WITH ordered_positions AS (
  SELECT
    signalk_timestamp,
    ST_Point(value_longitude, value_latitude) as position,
    value_latitude,
    value_longitude,
    LAG(ST_Point(value_longitude, value_latitude)) OVER (ORDER BY signalk_timestamp) as prev_position,
    strftime(date_trunc('hour', signalk_timestamp::TIMESTAMP), '%Y-%m-%dT%H:%M:%SZ') as time_bucket
  FROM read_parquet('data/vessels/urn_mrn_imo_mmsi_368396230/navigation/position/*.parquet', union_by_name=true)
  WHERE signalk_timestamp >= '2025-09-27T16:00:00Z'
    AND signalk_timestamp <= '2025-09-27T23:59:59Z'
    AND value_latitude IS NOT NULL AND value_longitude IS NOT NULL
),
distances AS (
  SELECT *,
    CASE
      WHEN prev_position IS NOT NULL
      THEN ST_Distance_Sphere(position, prev_position)
      ELSE 0
    END as distance_meters
  FROM ordered_positions
)
SELECT
  time_bucket,
  AVG(value_latitude) as avg_lat,
  AVG(value_longitude) as avg_lon,
  -- Calculate bounding box manually
  MIN(value_latitude) as min_lat,
  MAX(value_latitude) as max_lat,
  MIN(value_longitude) as min_lon,
  MAX(value_longitude) as max_lon,
  -- Distance and movement metrics
  SUM(distance_meters) as total_distance_meters,
  ROUND(SUM(distance_meters) / 1000.0, 2) as total_distance_km,
  COUNT(*) as position_records,
  -- Movement area approximation using bounding box
  (MAX(value_latitude) - MIN(value_latitude)) * 111320 *
  (MAX(value_longitude) - MIN(value_longitude)) * 111320 *
  COS(RADIANS(AVG(value_latitude))) as approx_area_m2
FROM distances
GROUP BY time_bucket
ORDER BY time_bucket;

Available Spatial Functions

• ST_Point(longitude, latitude) - Create point geometries
• ST_Distance_Sphere(point1, point2) - Calculate distances in meters
• ST_AsText(geometry) - Convert to Well-Known Text format
• ST_Centroid(ST_Collect(points)) - Find center of multiple points
• ST_ConvexHull(ST_Collect(points)) - Create movement boundary polygons

History API Integration

The plugin provides full SignalK History API compliance, allowing you to query historical data using standard SignalK API endpoints with enhanced performance and filtering capabilities.

Available Endpoints

Endpoint	Description	Parameters
/signalk/v1/history/values	Get historical values for specified paths	Standard patterns (see below) Optional: resolution, includeMovingAverages, bbox, radius
/signalk/v1/history/contexts	Get available vessel contexts for time range	Time Range: Any standard pattern (see below) ⚠️ Returns only contexts with data in specified range
/signalk/v1/history/paths	Get available SignalK paths for time range	Time Range: Any standard pattern (see below) ⚠️ Returns only paths with data in specified range
/signalk/v2/api/history/*	Spec-compliant - handled by registered HistoryApi provider	Per SignalK spec (ISO 8601 durations, no extensions)

Note: V2 routes (/signalk/v2/api/history/*) are handled by the registered HistoryApi provider (history-provider.ts) for SignalK server multi-provider support. V1 routes include signalk-parquet extensions (spatial filtering, shorthand durations, etc.) not available in V2.

⚠️ Extension: The /contexts and /paths endpoints accept time range parameters as optional. The official spec requires time parameters; without them, these endpoints return all available data (more permissive behavior).

Exact context ids (v0.7.44-beta.3+): the contexts endpoints return vessel context strings exactly as recorded — resolved from the stored data rather than reconstructed from partition directory names, whose encoding is lossy. Earlier versions mangled UUID-identified vessels (urn:mrn:signalk:uuid:…, the default when no MMSI is configured) by turning the UUID's dashes into colons.

Standard Time Range Patterns

The History API supports 5 standard SignalK time query patterns:

Pattern	Parameters	Description	Example
1	duration	Query back from now	?duration=1h
2	from + duration	Query forward from start	?from=2025-01-01T00:00:00Z&duration=1h
3	to + duration	Query backward to end	?to=2025-01-01T12:00:00Z&duration=1h
4	from	From start to now	?from=2025-01-01T00:00:00Z
5	from + to	Specific range	?from=2025-01-01T00:00:00Z&to=2025-01-02T00:00:00Z

Query Parameters

Parameter	Description	Format	Examples
Required for /values:
paths	SignalK paths with optional aggregation	path:method	navigation.speedOverGround:average
Time Range:	Use one of the 5 standard patterns above
duration	Time period (see Duration Formats below)	Multiple formats	PT1H, 3600, 1h
from	Start time (ISO 8601)	ISO datetime	2025-01-01T00:00:00Z
to	End time (ISO 8601)	ISO datetime	2025-01-01T06:00:00Z
Optional:
context	Vessel context	vessels.self or vessels.<id>	vessels.self (default)
resolution	Time bucket size in seconds	Seconds or time expression	60, 1m (1 minute buckets)

Duration Formats

Format	Example	Description
ISO 8601	PT1H, PT30M, P1D, PT1H30M	Standard ISO duration
Integer seconds	3600, 60	Plain number as seconds
Shorthand ⚠️	1h, 30m, 5s, 2d	Human-friendly format (extension)

⚠️ Shorthand format is a non-standard extension for convenience. Use ISO 8601 or integer seconds for maximum compatibility.

Resolution Parameter

BREAKING CHANGE (v0.7.0+): Resolution is now in seconds (was milliseconds).

Old (v0.6.x)	New (v0.7.0+)
?resolution=60000	?resolution=60 or ?resolution=1m
?resolution=5000	?resolution=5 or ?resolution=5s
?resolution=300000	?resolution=300 or ?resolution=5m

Aggregation Methods

Method	Description	Example
average	Average of values in bucket	path:average
min	Minimum value in bucket	path:min
max	Maximum value in bucket	path:max
first	First value in bucket	path:first
last	Last value in bucket	path:last
mid	Median value in bucket	path:mid
sma	Simple Moving Average, window default 5 samples (returns only smoothed value)	path:sma:5
ema	Exponential Moving Average, alpha default 0.2 (returns only smoothed value)	path:ema:0.2

SMA/EMA as aggregation methods (official SignalK syntax):

# SMA with window of 5 - returns ONLY the smoothed value (V1 with shorthand duration)
curl "http://localhost:3000/signalk/v1/history/values?duration=1h&paths=navigation.speedOverGround:sma:5"

# EMA with alpha of 0.3 - returns ONLY the smoothed value (V1 with shorthand duration)
curl "http://localhost:3000/signalk/v1/history/values?duration=1h&paths=environment.wind.speedApparent:ema:0.3"

Angular-aware smoothing (v0.7.43+, V2 provider routes): sma/ema first bucket values exactly like average, then apply the moving window. Paths with rad units (heading, COG, wind direction) are smoothed with a circular (vector) mean so the 0/2π wrap doesn't average 359°/1° toward 180°, and navigation.position longitude is smoothed across the ±180° antimeridian correctly (latitude linearly). Buckets with null/NaN values pass through unsmoothed rather than contaminating the moving average.

Filtering by Source

Append |<sourceRef> to a path to restrict it to a single SignalK source. This is useful when several devices publish the same path (e.g. multiple GPS receivers or heading sensors) and you want the history of just one of them. The source reference is the delta's $source value, and the filter applies after any optional aggregation method.

Format	Description	Example
path|sourceRef	Filter a path to one source	navigation.headingMagnetic|n2k-on-ve.can0.115
path:method|sourceRef	Aggregate and filter by source	navigation.speedOverGround:max|n2k-on-ve.can0.115

# Heading from a specific compass over the last hour
curl "http://localhost:3000/signalk/v1/history/values?duration=1h&paths=navigation.headingMagnetic|n2k-on-ve.can0.115"

Source filtering always reads raw data: aggregated tiers blend every source into each time bucket, so they cannot be filtered by source. Each values entry in the response echoes the sourceRef it was restricted to.

Extension Parameters (non-standard)

Parameter	Description	Format	Examples
paths ⚠️	Extended smoothing syntax: path:method:smoothing:param (returns raw AND smoothed)	Extended format	navigation.speedOverGround:average:sma:5
includeMovingAverages ⚠️	Include EMA/SMA calculations	true or 1	includeMovingAverages=true
bbox ⚠️	Bounding box filter: west,south,east,north	Coordinates	bbox=-74.5,40.2,-73.8,40.9
radius ⚠️	Radius filter: lon,lat,meters (GeoJSON convention)	Coordinates + meters	radius=-73.981,40.646,100

⚠️ Extensions: Parameters marked with ⚠️ are non-standard extensions to the SignalK History API specification. They provide additional functionality but may not be supported by other SignalK history providers.

Query Examples

Pattern 1: Duration Only (Query back from now)

# Last hour of wind data
curl "http://localhost:3000/signalk/v1/history/values?duration=1h&paths=environment.wind.speedApparent"

# Last 30 minutes with moving averages
curl "http://localhost:3000/signalk/v1/history/values?duration=30m&paths=environment.wind.speedApparent&includeMovingAverages=true"

Pattern 2: From + Duration (Query forward)

# 6 hours forward from specific time
curl "http://localhost:3000/signalk/v1/history/values?from=2025-01-01T00:00:00Z&duration=6h&paths=navigation.position"

Pattern 3: To + Duration (Query backward)

# 2 hours backward to specific time
curl "http://localhost:3000/signalk/v1/history/values?to=2025-01-01T12:00:00Z&duration=2h&paths=environment.wind.speedApparent"

Pattern 4: From Only (From start to now)

# From specific time until now
curl "http://localhost:3000/signalk/v1/history/values?from=2025-01-01T00:00:00Z&paths=navigation.speedOverGround"

Pattern 5: From + To (Specific range)

# Specific 24-hour period
curl "http://localhost:3000/signalk/v1/history/values?from=2025-01-01T00:00:00Z&to=2025-01-02T00:00:00Z&paths=navigation.position"

Advanced Query Examples

Multiple paths with time alignment:

curl "http://localhost:3000/signalk/v1/history/values?duration=6h&paths=environment.wind.angleApparent,environment.wind.speedApparent,navigation.position&resolution=1m"

Multiple aggregations of same path:

curl "http://localhost:3000/signalk/v1/history/values?from=2025-01-01T00:00:00Z&to=2025-01-01T06:00:00Z&paths=environment.wind.speedApparent:average,environment.wind.speedApparent:min,environment.wind.speedApparent:max&resolution=60"

With moving averages for trend analysis:

curl "http://localhost:3000/signalk/v1/history/values?duration=24h&paths=electrical.batteries.512.voltage&includeMovingAverages=true&resolution=5m"

Different temporal samples:

curl "http://localhost:3000/signalk/v1/history/values?duration=1h&paths=navigation.position:first,navigation.position:middle_index,navigation.position:last&resolution=1m"

Using ISO 8601 duration format:

curl "http://localhost:3000/signalk/v1/history/values?duration=PT1H30M&paths=navigation.speedOverGround&resolution=30"

Using integer seconds for duration:

curl "http://localhost:3000/signalk/v1/history/values?duration=3600&paths=navigation.speedOverGround&resolution=10s"

Context and Path Discovery

Get contexts with data in last hour:

curl "http://localhost:3000/signalk/v1/history/contexts?duration=1h"

Get contexts for specific time range:

curl "http://localhost:3000/signalk/v1/history/contexts?from=2025-01-01T00:00:00Z&to=2025-01-07T00:00:00Z"

Get vessels in a bounding box:

curl "http://localhost:3000/api/history/contexts/spatial?duration=7d&bbox=-74.1,40.5,-73.8,40.8"

Get vessels within radius (lon,lat,meters):

curl "http://localhost:3000/api/history/contexts/spatial?duration=24h&radius=-74.01,40.66,5000"

Get available paths with recent data:

curl "http://localhost:3000/signalk/v1/history/paths?duration=24h"

Get all paths (no time filter):

curl "http://localhost:3000/signalk/v1/history/paths"

Duration Formats

• 30s - 30 seconds
• 15m - 15 minutes
• 2h - 2 hours
• 1d - 1 day

Spatial Filtering

Filter data by geographic location using bounding boxes or radius queries:

Bounding Box Filter:

# Position data within a bounding box (west,south,east,north)
curl "http://localhost:3000/signalk/v1/history/values?duration=1h&paths=navigation.position&bbox=-74.5,40.2,-73.8,40.9"

Radius Filter:

# Position data within 100m of a point (lon,lat,meters — GeoJSON convention)
curl "http://localhost:3000/signalk/v1/history/values?duration=1h&paths=navigation.position&radius=-73.981,40.646,100"

Spatial Correlation (filter non-position paths by location):

# Wind data when vessel was within 100m of point
curl "http://localhost:3000/signalk/v1/history/values?duration=24h&paths=environment.wind.speedApparent&radius=-73.981,40.646,100"

# Multiple paths filtered by bounding box
curl "http://localhost:3000/signalk/v1/history/values?duration=7d&paths=environment.wind.speedApparent,environment.depth.belowKeel&bbox=-74.0,40.6,-73.9,40.7"

How Spatial Correlation Works:

• For position paths (e.g., navigation.position): Filters directly on lat/lon
• For non-position paths (e.g., environment.wind.speedApparent): First queries position data to find timestamps when vessel was within the spatial filter, then returns only data from those times
• Spatial correlation always uses navigation.position for location lookup

Query Data Hierarchy

Queries resolve data from a three-tier hierarchy, selected automatically based on date ranges:

1. Local tiered Parquet — Primary source. Uses aggregation tiers (raw, 5s, 60s, 1h) based on query resolution. Data available from first export through yesterday.
2. Cloud supplement (S3/R2) — For dates before local data starts (older than retentionDays). DuckDB queries cloud storage directly without downloading files first.
3. SQLite buffer — Today's live data, not yet exported to Parquet. Bucketed to match the query resolution.

For queries spanning multiple sources, results are combined with UNION.

Cloud Query Optimization: DuckDB's native cloud storage support provides:

• Partition pruning: Hive structure (year=/day=) allows skipping irrelevant files
• Predicate pushdown: WHERE clauses filter at Parquet level before transfer
• Projection pushdown: Only SELECT columns are transferred
• Combined effect: 70-99% reduction vs downloading full files

Requirements for cloud queries:

• Cloud upload must be configured (provider: 's3' or 'r2')
• Valid credentials configured
• Data must be uploaded using Hive partition structure

Timestamp Handling

All timestamps follow ISO 8601 conventions:

• Bare timestamps (e.g., 2025-08-13T09:00:00) are treated as server local time
• Z-suffix (e.g., 2025-08-13T09:00:00Z) is UTC
• Explicit offset (e.g., 2025-08-13T09:00:00-04:00) is parsed as-is

All response timestamps are in server local time with offset (e.g., 2025-10-20T12:34:04-04:00).

# Bare timestamp — interpreted as server local time
curl "http://localhost:3000/signalk/v1/history/values?context=vessels.self&to=2025-08-13T09:00:00&duration=1h&paths=navigation.position"

# Explicit UTC
curl "http://localhost:3000/signalk/v1/history/values?context=vessels.self&to=2025-08-13T09:00:00Z&duration=1h&paths=navigation.position"

# Explicit offset
curl "http://localhost:3000/signalk/v1/history/values?context=vessels.self&to=2025-08-13T09:00:00-04:00&duration=1h&paths=navigation.position"

Get available contexts:

curl "http://localhost:3000/signalk/v1/history/contexts"

Time Alignment and Bucketing

The History API automatically aligns data from different paths using time bucketing to solve the common problem of misaligned timestamps. This enables:

• Plotting: Data points align properly on charts
• Correlation: Compare values from different sensors at the same time
• Export: Clean, aligned datasets for analysis

Key Features:

• Smart Type Handling: Automatically handles numeric values (wind speed) and JSON objects (position)
• Robust Aggregation: Uses proper SQL type casting to prevent type errors
• Configurable Resolution: Time bucket size in seconds (default: auto-calculated based on time range)
• Multiple Aggregation Methods: average for numeric data, first for complex objects

Parameters:

• resolution - Time bucket size in seconds (default: auto-calculated)
• Aggregation methods: average, min, max, first, last, mid, middle_index

Aggregation Methods:

• average - Average value in time bucket (default for numeric data)
• min - Minimum value in time bucket
• max - Maximum value in time bucket
• first - First value in time bucket (default for objects)
• last - Last value in time bucket
• mid ⚠️ - Median value (average of middle values for even counts) - extension
• middle_index ⚠️ - Middle value by index (first of two middle values for even counts) - extension

When to Use Each Method:

• Numeric data (wind speed, voltage, etc.): Use average, min, max for statistics
• Position data: Use first, last, middle_index for specific readings
• String/object data: Avoid mid (unpredictable), prefer first, last, middle_index
• Multiple stats: Query same path with different methods (e.g., wind:average,wind:max)

Response Format

The History API returns time-aligned data in standard SignalK format.

Default Response (without moving averages)

{
  "context": "vessels.self",
  "range": {
    "from": "2025-01-01T00:00:00Z",
    "to": "2025-01-01T06:00:00Z"
  },
  "values": [
    {
      "path": "environment.wind.speedApparent",
      "method": "average"
    },
    {
      "path": "navigation.position",
      "method": "first"
    }
  ],
  "data": [
    ["2025-01-01T00:00:00Z", 12.5, {"latitude": 37.7749, "longitude": -122.4194}],
    ["2025-01-01T00:01:00Z", 13.2, {"latitude": 37.7750, "longitude": -122.4195}],
    ["2025-01-01T00:02:00Z", 11.8, {"latitude": 37.7751, "longitude": -122.4196}]
  ]
}

With Moving Averages (includeMovingAverages=true)

{
  "context": "vessels.self",
  "range": {
    "from": "2025-01-01T00:00:00Z",
    "to": "2025-01-01T06:00:00Z"
  },
  "values": [
    {
      "path": "environment.wind.speedApparent",
      "method": "average"
    },
    {
      "path": "environment.wind.speedApparent.ema",
      "method": "ema"
    },
    {
      "path": "environment.wind.speedApparent.sma",
      "method": "sma"
    },
    {
      "path": "navigation.position",
      "method": "first"
    }
  ],
  "data": [
    ["2025-01-01T00:00:00Z", 12.5, 12.5, 12.5, {"latitude": 37.7749, "longitude": -122.4194}],
    ["2025-01-01T00:01:00Z", 13.2, 12.64, 12.85, {"latitude": 37.7750, "longitude": -122.4195}],
    ["2025-01-01T00:02:00Z", 11.8, 12.45, 12.5, {"latitude": 37.7751, "longitude": -122.4196}]
  ]
}

Notes:

• Each data array element is [timestamp, value1, value2, ...] corresponding to the paths in the values array
• Moving averages (EMA/SMA) are opt-in - add includeMovingAverages=true to include them
• EMA/SMA are only calculated for numeric values; non-numeric values (objects, strings) show null for their EMA/SMA columns
• Without includeMovingAverages, response size is ~66% smaller

Response Extensions (non-standard) ⚠️

When using extension parameters, the response may include additional non-standard fields:

Field	Added by	Description
meta.autoConfigured	Auto-discovery	Indicates paths were auto-configured for recording

These fields are extensions and may not be present in responses from other SignalK history providers.

Moving Averages (EMA & SMA)

The plugin calculates Exponential Moving Average (EMA) and Simple Moving Average (SMA) for numeric values, providing enhanced trend analysis capabilities. There are two ways to enable smoothing:

Per-Path Smoothing Syntax

Apply smoothing directly in the path specification using the path:method:smoothing:param syntax:

# SMA with 5-point window
curl "http://localhost:3000/signalk/v1/history/values?duration=1h&paths=navigation.speedOverGround:average:sma:5"

# EMA with alpha=0.3
curl "http://localhost:3000/signalk/v1/history/values?duration=1h&paths=environment.wind.speedApparent:max:ema:0.3"

# Mixed: some paths with smoothing, some without
curl "http://localhost:3000/signalk/v1/history/values?duration=1h&paths=navigation.speedOverGround:average:sma:5,navigation.courseOverGround:average"

# SMA with default period (10)
curl "http://localhost:3000/signalk/v1/history/values?duration=1h&paths=navigation.speedOverGround:average:sma"

# EMA with default alpha (0.2)
curl "http://localhost:3000/signalk/v1/history/values?duration=1h&paths=navigation.speedOverGround:average:ema"

Path Syntax Format: path:aggregateMethod:smoothingType:smoothingParam

• path - SignalK path (e.g., navigation.speedOverGround)
• aggregateMethod - Aggregation method: average, min, max, first, last, mid (default: average)
• smoothingType - sma (Simple Moving Average) or ema (Exponential Moving Average)
• smoothingParam - For SMA: window size (default: 10), for EMA: alpha value 0-1 (default: 0.2)

Per-Path Response Format:

{
  "values": [
    {
      "path": "navigation.speedOverGround",
      "method": "average"
    },
    {
      "path": "navigation.speedOverGround",
      "method": "average",
      "smoothing": "sma",
      "window": 5
    }
  ],
  "data": [
    ["2025-01-01T00:00:00Z", 5.05, 5.05],
    ["2025-01-01T00:01:00Z", 5.12, 5.09],
    ["2025-01-01T00:02:00Z", 4.98, 5.05]
  ]
}

Note: With per-path smoothing, the response includes both the raw value AND the smoothed value as separate columns. Paths without smoothing specified only get a single column.

Global Moving Averages (Legacy)

History API:

# Add includeMovingAverages=true to any query
curl "http://localhost:3000/signalk/v1/history/values?duration=1h&paths=environment.wind.speedApparent&includeMovingAverages=true"

Default Behavior (v0.5.6+):

• Moving averages are opt-in - not included by default
• Reduces response size by ~66% when not needed
• Better API compliance with SignalK specification

Legacy Behavior (pre-v0.5.6):

• Moving averages were automatically included for all queries
• To maintain old behavior, add includeMovingAverages=true to all requests

Calculation Details

Exponential Moving Average (EMA)

• Period: ~10 equivalent (α = 0.2)
• Formula: EMA = α × currentValue + (1 - α) × previousEMA
• Characteristic: Responds faster to recent changes, emphasizes recent data
• Use Case: Trend detection, rapid response to data changes

Simple Moving Average (SMA)

• Period: 10 data points
• Formula: Average of the last 10 values
• Characteristic: Smooths out fluctuations, equal weight to all values in window
• Use Case: Noise reduction, general trend analysis

Data Flow & Continuity

// Initial Data Load (isIncremental: false)
Point 1: Value=5.0, EMA=5.0,   SMA=5.0
Point 2: Value=6.0, EMA=5.2,   SMA=5.5
Point 3: Value=4.0, EMA=5.0,   SMA=5.0

// Incremental Updates (isIncremental: true)
Point 4: Value=7.0, EMA=5.4,   SMA=5.5  // Continues from previous EMA
Point 5: Value=5.5, EMA=5.42,  SMA=5.5  // Rolling 10-point SMA window

Key Features

• 🎛️ Opt-In: Add includeMovingAverages=true to enable (v0.5.6+)
• ✅ Memory Efficient: SMA maintains rolling 10-point window
• ✅ Non-Numeric Handling: Non-numeric values (strings, objects) show null for EMA/SMA
• ✅ Precision: Values rounded to 3 decimal places to prevent floating-point noise
• ⚡ Performance: Smaller response sizes when not needed

Real-world Applications

Marine Data Examples:

• Wind Speed: EMA detects gusts quickly, SMA shows general wind conditions
• Battery Voltage: EMA shows charging/discharging trends, SMA indicates overall battery health
• Engine RPM: EMA responds to throttle changes, SMA shows average operating level
• Water Temperature: EMA detects thermal changes, SMA provides stable baseline

Available in:

• 📊 History API: Add includeMovingAverages=true to include EMA/SMA calculations

Track API Integration

⚠️ Preview — requires unreleased server support. The SignalK Track API is proposed in SignalK/signalk-server#2995 and is not part of a released signalk-server yet. On a release server this plugin logs one debug line at startup and skips registration; nothing else changes. On a server carrying that PR, the plugin registers as a Track API provider and the endpoints below work. The contract may still move until the PR merges.

The plugin registers a TrackProvider (src/track-provider.ts) the same way it registers the History API provider. The server owns the routes and the query parsing; the plugin answers from raw-tier navigation.position parquet federated with the live SQLite buffer, so a track is available for any window the store holds, including the hour before the plugin was installed.

Endpoints (served by the server)

Endpoint	Description
GET /signalk/v2/api/tracks	Tracks matching the query, as a GeoJSON FeatureCollection, one Feature per context
GET /signalk/v2/api/tracks/contexts	Contexts with track data in the window, without geometry
GET /signalk/v2/api/tracks/_providers	Registered providers; ?provider=signalk-parquet selects this one when several are installed

All registered providers are queried and their features concatenated; each feature carries properties.providerId, so a server running both this plugin and @signalk/tracks-plugin returns one feature from each.

Query parameters

Parameter	Description
context / contexts	Vessel(s) to return. Defaults to the own vessel. Bare ids are qualified with vessels.
from, to, duration	Time window. duration measures back from to (default now); with from as well, the later start wins. Required unless a single context is requested
bbox	west,south,east,north. Selects tracks that pass through the box during the window; a matching track is returned whole, not clipped
resolution	Minimum spacing between points (ISO 8601 or seconds). The spacing actually applied is reported in properties.resolution
maxPoints	Point budget per track; the spacing is widened until it fits. Default budget 5000
simplify, epsilon	Douglas-Peucker simplification, tolerance in metres; the applied tolerance is reported
times	Include the recording time of every point as properties.coordTimes, nested like coordinates
properties	Comma-separated paths to return alongside each position (e.g. navigation.speedOverGround), nested like coordinates under properties.values. Only paths the store holds are returned; properties.appliedProperties lists them. Values are matched to the nearest sample within a few seconds, because different talkers stamp position and speed a few hundred milliseconds apart. Angular paths use a circular mean
geometry=false	Metadata only, no coordinates

How the provider answers

• Points are thinned by time bucketing in DuckDB, keeping the first fix in each bucket rather than an average (averaging cuts corners off a track).
• A gap in recording longer than five buckets, and at least five minutes, starts a new segment of the MultiLineString, so a line is never drawn across a stretch the vessel did not travel.
• Position is read from the raw tier only, since the aggregated tiers collapse object paths; long windows are thinned by bucket size, not by tier.
• A request with no window returns the context's whole recorded history, anchored on its earliest raw partition.

# Last 24 hours of the own vessel, with times and speed
curl "http://localhost:3000/signalk/v2/api/tracks?duration=P1D&times=true&properties=navigation.speedOverGround"

# Every recorded vessel that passed through a box this week, at most 500 points each
curl "http://localhost:3000/signalk/v2/api/tracks?bbox=23.5,60.0,23.6,60.1&duration=P7D&maxPoints=500"

# Only this plugin's answer, when several providers are registered
curl "http://localhost:3000/signalk/v2/api/tracks?duration=PT1H&provider=signalk-parquet"

The Track API has no radius filter; that exists only on this plugin's own History API routes (see Spatial Filtering). A shared spatial parameter set for the v2 APIs is proposed in SignalK/signalk-server#3021.

Cloud Storage (S3 / Cloudflare R2)

Configuration

Set provider to s3 or r2 in the cloud upload configuration. Cloud uploads run automatically as part of the daily export pipeline — after Parquet files are created, they are uploaded to the configured cloud provider.

{
  "cloudUpload": {
    "provider": "s3",
    "bucket": "my-marine-data",
    "region": "us-east-1",
    "keyPrefix": "marine-data/",
    "accessKeyId": "...",
    "secretAccessKey": "..."
  }
}

For Cloudflare R2, use provider: "r2" and supply accountId instead of region.

Cloud Key Structure

With prefix marine-data/ and Hive partitioning:

marine-data/tier=raw/context=vessels__self/path=navigation__position/year=2026/day=062/signalk_data_2026-03-03T0400.parquet

Daily Export

The plugin uses a simplified daily export pipeline:

1. Data Collection: SignalK data is buffered in crash-safe SQLite WAL database
2. Daily Export: At configurable hour (default: 4 AM UTC), exports previous day's data
3. Direct Consolidation: Creates one Parquet file per context/path/day (no separate merge step)
4. Timestamped Files: Each export uses current timestamp for unique filenames
5. S3 Upload: Uploads daily files if configured

Note: The previous 5-minute interval export and separate consolidation step have been removed. This simplifies the pipeline and creates cleaner daily files directly.

Startup Sequence

When the plugin starts, it runs the following initialization steps:

1. Configuration & State — Load plugin config, vessel identity, output directory, cloud credentials
2. SQLite Buffer — Open WAL-mode database; auto-migrate legacy buffer_records table to per-path tables if needed
3. Cloud Client — Initialize S3 or R2 SDK if a cloud provider is configured
4. DuckDB Pool — Initialize connection pool; attach SQLite buffer for federated queries; register cloud credentials
5. Data Subscriptions — Subscribe to configured SignalK paths and start threshold monitoring
6. Periodic Save — Start flush interval (default: every 30s) from memory buffer to SQLite
7. Daily Export Schedule — Schedule next export at configured UTC hour (default: 4 AM); includes aggregation and cloud upload
8. Startup Catch-Up (10s delay) — Export any unexported historical data from SQLite, re-aggregate affected dates, and sync recent files to cloud (7-day lookback, raw-tier-only prefix scan)
9. History API — Register HTTP routes and SignalK HistoryApi provider
10. Auto-Discovery — Initialize service for on-demand path configuration

Performance Characteristics

• Memory Usage: Configurable buffer sizes (default 1000 records)
• Disk I/O: Efficient batch writes with configurable intervals
• CPU Usage: Minimal - mostly I/O bound operations
• Network: Optional S3 uploads with retry logic

Development

Project Structure

signalk-parquet/
├── src/
│   ├── index.ts                # Main plugin entry point and lifecycle
│   ├── commands.ts             # Command management system
│   ├── data-handler.ts         # Data processing, subscriptions, cloud upload
│   ├── api-routes.ts           # Web API endpoints
│   ├── types.ts                # TypeScript interfaces
│   ├── parquet-writer.ts       # File writing logic
│   ├── HistoryAPI.ts           # SignalK History API implementation
│   ├── HistoryAPI-types.ts     # History API type definitions
│   ├── history-provider.ts     # SignalK HistoryApi provider (v2)
│   ├── track-provider.ts       # SignalK Track API provider (preview, server PR #2995)
│   ├── services/
│   │   ├── aggregation-service.ts  # Tier aggregation (raw→5s→60s→1h)
│   │   └── parquet-export-service.ts # Daily export pipeline
│   └── utils/
│       ├── sqlite-buffer.ts    # Per-path SQLite buffer
│       ├── buffer-sql-builder.ts # SQL generation for buffer queries
│       ├── duckdb-pool.ts      # DuckDB connection pooling
│       ├── hive-path-builder.ts # Hive partition path generation
│       ├── angular-paths.ts    # Angular path detection (units=rad)
│       ├── duration-parser.ts  # ISO 8601 and shorthand duration parsing
│       ├── spatial-queries.ts  # Geographic/spatial query support
│       ├── track-geometry.ts   # Track segmentation, simplification, bbox
│       ├── lru-cache.ts        # LRU cache implementation
│       ├── path-helpers.ts     # Path utility functions
│       ├── path-discovery.ts   # Path auto-discovery
│       ├── context-discovery.ts # Context auto-discovery
│       ├── type-detector.ts    # Smart data type detection
│       ├── schema-cache.ts     # Parquet schema caching
│       └── ...                 # Additional utilities
├── dist/                       # Compiled JavaScript
├── public/
│   ├── index.html             # Web interface
│   └── parquet.png            # Plugin icon
├── tsconfig.json              # TypeScript configuration
├── package.json               # Dependencies and scripts
└── README.md                  # This file

Code Architecture

The plugin uses a modular TypeScript architecture for maintainability:

• index.ts: Plugin lifecycle, configuration, and initialization
• commands.ts: SignalK command registration, execution, and management
• data-handler.ts: Data subscriptions, buffering, and cloud upload operations
• api-routes.ts: REST API endpoints for web interface
• types.ts: Comprehensive TypeScript type definitions
• services/: Aggregation and export pipeline services
• utils/: SQLite buffer, DuckDB pool, spatial queries, path helpers, and more

Adding New Features

1. API Endpoints: Add to src/api-routes.ts
2. Data Processing: Extend src/data-handler.ts
3. Commands: Modify src/commands.ts
4. Types: Add interfaces to src/types.ts
5. Update Documentation: Update README and inline comments

Type Checking

The plugin uses strict TypeScript configuration:

{
  "compilerOptions": {
    "strict": true,
    "noImplicitAny": true,
    "noImplicitReturns": true,
    "strictNullChecks": true
  }
}

Troubleshooting

Common Issues

Build Errors

# Clean and rebuild
npm run clean
npm run build

DuckDB Not Available

• Check that @duckdb/node-api is installed
• Verify Node.js version compatibility (>=22.5.0)

Cloud Upload Failures

• Verify cloud credentials and permissions
• Check bucket exists and is accessible
• Test connection using web interface

No Data Collection

• Verify path configurations are correct
• Check if regimens are properly activated
• Review SignalK logs for subscription errors

Debug Mode

Enable debug logging in SignalK:

{
  "settings": {
    "debug": "signalk-parquet*"
  }
}

Runtime Dependencies

• @dsnp/parquetjs: Parquet file format support
• @duckdb/node-api: SQL query engine
• @aws-sdk/client-s3: S3 upload functionality
• fs-extra: Enhanced file system operations
• glob: File pattern matching
• express: Web server framework

Development Dependencies

• typescript: TypeScript compiler
• @types/node: Node.js type definitions
• @types/express: Express type definitions
• @types/fs-extra: fs-extra type definitions

License

MIT License - See LICENSE file for details.

Testing

Comprehensive testing procedures are documented in TESTING.md. The testing guide covers:

• Installation and build verification
• Plugin configuration testing
• Web interface functionality
• Data collection validation
• Regimen control testing
• File output verification
• Cloud storage integration testing
• API endpoint testing
• Performance testing
• Error handling validation

Quick Test

# Test plugin health
curl http://localhost:3000/plugins/signalk-parquet/api/health

# Test path configuration
curl http://localhost:3000/plugins/signalk-parquet/api/config/paths

# Test data collection
curl http://localhost:3000/plugins/signalk-parquet/api/paths

# Test History API
curl "http://localhost:3000/signalk/v1/history/contexts"

Legacy Notes

Upgrading to 0.7.41-beta.3: Faster Startup Sweep

The startup sweep that quarantines crash-truncated (0-byte) parquet files is now incremental, keyed to a watermark file at <dataDir>/.last-empty-sweep. The first start after upgrading does one full scan of the store to establish the watermark (on very large stores this can take a minute or two), and every start after that only inspects directories changed since the previous start — typically near-instant. No action is required; the watermark file is created automatically. See CHANGELOG.md for details.

Upgrading from 0.5.0-beta.3: Consolidation Bug Fix

If upgrading from versions prior to 0.5.0-beta.4, you may have nested processed directories from a recursive consolidation bug:

# Check for nested processed directories
find data -name "*processed*" -type d | head -20

# Remove ALL nested processed directories (RECOMMENDED)
find data -name "processed" -type d -exec rm -rf {} +

The processed directories contain legacy files from the old consolidation system. The new daily export pipeline no longer uses this directory.

Legacy Flat File Structure

Pre-Hive versions stored data in a flat directory structure:

output_directory/
├── vessels/
│   └── self/
│       ├── navigation/
│       │   └── position/
│       │       └── signalk_data_20250716T120000.parquet
└── processed/

Use the Migration API to convert legacy files to Hive partitioning.

Contributing

For new features, open an issue first to discuss the approach with the maintainer before starting work.

1. Fork the repository
2. Create a feature branch
3. Add TypeScript types for new features
4. Add unit and integration tests covering your changes, and update the documentation
5. Follow the testing procedures in TESTING.md
6. Ensure linting and tests pass locally before submitting the PR (run npm run ci for lint and format checks)
7. Submit a pull request
8. Address CodeRabbit review comments on the PR before human review

Changelog

See CHANGELOG.md for complete version history.