From cb8e2522b3060ed7a7259aed30687b03b0fb9311 Mon Sep 17 00:00:00 2001 From: Ragnarok Date: Fri, 25 Sep 2026 12:14:36 +0300 Subject: [PATCH] feat(signalk): kinavis-signalk, Signal K deltas in, CPA/TCPA and the COLREGs ruling out - New crate kinavis-signalk 0.1.0 (std): a picture of own vessel and every target from Signal K deltas, SI units converted at the boundary; a watch that assesses each target with kinavis-traffic and, for a dangerous or developing approach, rules with kinavis-colregs. - Out: navigation.closestApproach in each target's context, and notifications.navigation.closestApproach. on own vessel when a target's level changes. - Own vessel is the vessel whose position comes from a sensor other than AIS, or named explicitly. The clock is the caller's: sources stamp by different clocks. Below half a knot a vessel is at rest; no ruling for own vessel at rest. Warnings only within a horizon. - Checked on a Signal K server 2.33 through the signalk-kinavis WASM plugin. --- Cargo.lock | 12 + README.md | 3 + ci/allowed-deps.toml | 6 + crates/kinavis-signalk/Cargo.toml | 28 ++ crates/kinavis-signalk/LICENSE-APACHE | 1 + crates/kinavis-signalk/LICENSE-MIT | 1 + crates/kinavis-signalk/README.md | 44 +++ crates/kinavis-signalk/src/delta.rs | 87 +++++ crates/kinavis-signalk/src/lib.rs | 60 ++++ crates/kinavis-signalk/src/picture.rs | 262 +++++++++++++++ crates/kinavis-signalk/src/time.rs | 92 ++++++ crates/kinavis-signalk/src/watch.rs | 459 ++++++++++++++++++++++++++ crates/kinavis-signalk/tests/watch.rs | 343 +++++++++++++++++++ crates/kinavis/README.md | 1 + 14 files changed, 1399 insertions(+) create mode 100644 crates/kinavis-signalk/Cargo.toml create mode 120000 crates/kinavis-signalk/LICENSE-APACHE create mode 120000 crates/kinavis-signalk/LICENSE-MIT create mode 100644 crates/kinavis-signalk/README.md create mode 100644 crates/kinavis-signalk/src/delta.rs create mode 100644 crates/kinavis-signalk/src/lib.rs create mode 100644 crates/kinavis-signalk/src/picture.rs create mode 100644 crates/kinavis-signalk/src/time.rs create mode 100644 crates/kinavis-signalk/src/watch.rs create mode 100644 crates/kinavis-signalk/tests/watch.rs diff --git a/Cargo.lock b/Cargo.lock index e6cfcf8..9e18ef9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -129,6 +129,18 @@ dependencies = [ "serde_json", ] +[[package]] +name = "kinavis-signalk" +version = "0.1.0" +dependencies = [ + "kinavis", + "kinavis-colregs", + "kinavis-kernel", + "kinavis-traffic", + "serde", + "serde_json", +] + [[package]] name = "kinavis-traffic" version = "0.1.0" diff --git a/README.md b/README.md index ca6b040..c87dcef 100644 --- a/README.md +++ b/README.md @@ -80,6 +80,8 @@ graph BT traffic[kinavis-traffic] --> kinavis colregs[kinavis-colregs] --> kinavis alerts[kinavis-alerts] --> traffic + signalk[kinavis-signalk] --> traffic + signalk --> colregs ``` | Crate | What it does | @@ -94,6 +96,7 @@ graph BT | [`kinavis-traffic`](https://docs.rs/kinavis-traffic) | target tracking from radar and AIS, CPA and TCPA, the avoiding manoeuvre | | [`kinavis-colregs`](https://docs.rs/kinavis-colregs) | the steering and sailing rules of the COLREGs as data | | [`kinavis-alerts`](https://docs.rs/kinavis-alerts) | bridge alert management: alarms, warnings and cautions, with acknowledgement | +| [`kinavis-signalk`](https://docs.rs/kinavis-signalk) | Signal K deltas in, CPA, TCPA and the COLREGs ruling out; the core of the [Signal K plugin](https://github.com/KINAVIS/signalk-kinavis) (`std`) | An adapter depends on the kernel alone and never pulls in the algorithms; take only the crates you need. diff --git a/ci/allowed-deps.toml b/ci/allowed-deps.toml index d926da5..d59cefb 100644 --- a/ci/allowed-deps.toml +++ b/ci/allowed-deps.toml @@ -67,3 +67,9 @@ allow = ["kinavis-kernel", "serde"] # Unpublished demonstrations. Every KINAVIS crate they use is a # dev-dependency, so the shipped graph of this package is empty. allow = [] + +[kinavis-signalk] +# An adapter that sits above the rules: Signal K deltas become own ship and +# targets, and the assessment and the ruling go back as deltas. JSON needs +# `serde_json`; nothing below depends on this crate. +allow = ["kinavis-kernel", "kinavis", "kinavis-traffic", "kinavis-colregs", "serde", "serde_json"] diff --git a/crates/kinavis-signalk/Cargo.toml b/crates/kinavis-signalk/Cargo.toml new file mode 100644 index 0000000..22d6583 --- /dev/null +++ b/crates/kinavis-signalk/Cargo.toml @@ -0,0 +1,28 @@ +[package] +name = "kinavis-signalk" +description = "Signal K deltas in and out of the KINAVIS navigation crates: the traffic picture from a Signal K server, and CPA, TCPA and the COLREGs ruling for every target back as deltas and notifications." +documentation = "https://docs.rs/kinavis-signalk" +readme = "README.md" +keywords = ["signalk", "ais", "colregs", "cpa", "marine"] +categories = ["science", "parser-implementations"] +# Satellite crate, versioned independently of the kernel and `kinavis`. +version = "0.1.0" +authors.workspace = true +edition.workspace = true +rust-version.workspace = true +repository.workspace = true +homepage.workspace = true +license.workspace = true + +[dependencies] +# Signal K is JSON over a server: this adapter needs the standard library, as +# the crates it feeds do not. +kinavis-kernel = { workspace = true, features = ["std"] } +kinavis = { workspace = true, features = ["std"] } +kinavis-traffic = { version = "0.1.0", path = "../kinavis-traffic" } +kinavis-colregs = { version = "0.1.0", path = "../kinavis-colregs" } +serde = { workspace = true } +serde_json = { workspace = true } + +[lints] +workspace = true diff --git a/crates/kinavis-signalk/LICENSE-APACHE b/crates/kinavis-signalk/LICENSE-APACHE new file mode 120000 index 0000000..1cd601d --- /dev/null +++ b/crates/kinavis-signalk/LICENSE-APACHE @@ -0,0 +1 @@ +../../LICENSE-APACHE \ No newline at end of file diff --git a/crates/kinavis-signalk/LICENSE-MIT b/crates/kinavis-signalk/LICENSE-MIT new file mode 120000 index 0000000..b2cfbdc --- /dev/null +++ b/crates/kinavis-signalk/LICENSE-MIT @@ -0,0 +1 @@ +../../LICENSE-MIT \ No newline at end of file diff --git a/crates/kinavis-signalk/README.md b/crates/kinavis-signalk/README.md new file mode 100644 index 0000000..ee0cdff --- /dev/null +++ b/crates/kinavis-signalk/README.md @@ -0,0 +1,44 @@ +# kinavis-signalk + +Signal K deltas in and out of the [KINAVIS](https://crates.io/crates/kinavis) +navigation crates. + +A [Signal K](https://signalk.org) server merges a vessel's NMEA 0183, +NMEA 2000 and AIS into one JSON model and streams changes as deltas. This +crate reads them into a picture of own vessel and every target, and assesses +it: CPA and TCPA of each target, and for a dangerous or developing approach +the COLREGs ruling — the encounter, who gives way, and what each may do. The +results go back as `navigation.closestApproach` deltas in each target's +context and as notifications on own vessel: + +```text +notifications.navigation.closestApproach.urn:mrn:imo:mmsi:244000001 + state: alarm + message: ANNA: CPA 0.00 NM in 15:00; crossing, target on starboard, give way (Rule 15): alter to starboard, or slow down +``` + +- Units converted at the boundary: Signal K's radians and metres per second + in, KINAVIS types inside. +- COLREGs category from `navigation.state`, else the AIS ship type; a vessel + at anchor, moored or aground gets a CPA but no ruling. +- An alarm for a dangerous CPA within the TCPA limit, a warning for one within + the warning horizon; a notification is sent when a target's level changes, + and cleared when it is no longer dangerous, stale or gone. +- No ruling while own vessel is at rest: the steering rules are for vessels + under way. + +An aid to the watch, not a substitute for it: a proper lookout (COLREGs +Rule 5) comes first, the ruling takes every target to be what AIS says it is, +and Rules 9 and 10 are not applied. + +Requires the standard library. The Signal K server plugin built on it is +[`signalk-kinavis`](https://github.com/KINAVIS/signalk-kinavis). + +The navigation algorithms and the list of the other KINAVIS crates are in +[`kinavis`](https://crates.io/crates/kinavis). + +## License + +Licensed under either of [Apache License, Version 2.0](LICENSE-APACHE) or [MIT license](LICENSE-MIT) at your option. + +Unless you explicitly state otherwise, any contribution intentionally submitted for inclusion in this crate by you, as defined in the Apache-2.0 license, shall be dual licensed as above, without any additional terms or conditions. diff --git a/crates/kinavis-signalk/src/delta.rs b/crates/kinavis-signalk/src/delta.rs new file mode 100644 index 0000000..8684266 --- /dev/null +++ b/crates/kinavis-signalk/src/delta.rs @@ -0,0 +1,87 @@ +//! The Signal K delta: a context, and updates of paths within it. +//! +//! Only the fields this crate reads or writes are modelled; anything else in +//! a delta is ignored on reading. + +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +/// One delta message. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct Delta { + /// Vessel the updates are about, `vessels.`; absent means own vessel. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub context: Option, + /// Updates, each from one source at one time. + #[serde(default)] + pub updates: Vec, +} + +/// Values from one source at one time. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct Update { + /// RFC 3339 time of the values; the server fills it in on emission. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub timestamp: Option, + /// Where the values came from: for a device, `type` with `sentence` + /// (NMEA 0183) or `pgn` (NMEA 2000). The server fills it in. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub source: Option, + /// Path and value pairs. + #[serde(default)] + pub values: Vec, +} + +/// A value at a path. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct PathValue { + /// Dotted path, `navigation.position`; empty for the vessel's root + /// properties (`name`, `mmsi`). + pub path: String, + /// JSON value; `null` clears it. + pub value: Value, +} + +impl Delta { + /// Delta of one value at one path in a context. + #[must_use] + pub fn single(context: Option, path: &str, value: Value) -> Self { + Self { + context, + updates: vec![Update { + timestamp: None, + source: None, + values: vec![PathValue { + path: path.to_owned(), + value, + }], + }], + } + } +} + +/// NMEA 2000 parameter groups that carry other vessels: the AIS messages. +const AIS_PGNS: [u64; 11] = [ + 129_038, 129_039, 129_040, 129_041, 129_793, 129_794, 129_798, 129_801, 129_802, 129_809, + 129_810, +]; + +impl Update { + /// Whether the values come from own vessel's own sensors: an NMEA 0183 + /// sentence other than AIS `VDM`, or an NMEA 2000 parameter group other + /// than AIS. Signal K puts such values in own vessel's context; only AIS + /// describes other vessels. `false` when the source does not say. + #[must_use] + pub fn is_own_sensor(&self) -> bool { + let Some(source) = &self.source else { + return false; + }; + if let Some(sentence) = source.get("sentence").and_then(Value::as_str) { + return sentence != "VDM"; + } + source + .get("pgn") + .and_then(Value::as_u64) + .is_some_and(|pgn| !AIS_PGNS.contains(&pgn)) + } +} diff --git a/crates/kinavis-signalk/src/lib.rs b/crates/kinavis-signalk/src/lib.rs new file mode 100644 index 0000000..3524eb1 --- /dev/null +++ b/crates/kinavis-signalk/src/lib.rs @@ -0,0 +1,60 @@ +//! Signal K deltas in and out of the KINAVIS navigation crates. +//! +//! A [Signal K](https://signalk.org) server merges a vessel's NMEA 0183, +//! NMEA 2000 and AIS into one JSON model and streams changes as deltas. This +//! crate reads those deltas into a [`Picture`] of own vessel and every +//! target, and a [`Watch`] assesses the picture: CPA and TCPA of each target +//! (`kinavis-traffic`), and for a dangerous or developing approach the +//! COLREGs ruling (`kinavis-colregs`): the encounter, who gives way and what +//! each may do. The results go back to the server as +//! `navigation.closestApproach` deltas in each target's context and as +//! notifications on own vessel. +//! +//! Signal K units are SI: radians, metres, metres per second. The crate +//! converts at the boundary; everything inside is KINAVIS types. +//! +//! ```rust +//! use kinavis_signalk::{parse_timestamp, Emission, Level, Watch, WatchConfig}; +//! +//! let mut watch = Watch::new(WatchConfig::default()); +//! // Own vessel steering north at 6 knots (3.087 m/s)... +//! watch.ingest(r#"{"context":"vessels.self","updates":[{"timestamp":"2026-09-25T12:00:00Z","values":[ +//! {"path":"navigation.position","value":{"latitude":53.0,"longitude":5.0}}, +//! {"path":"navigation.courseOverGroundTrue","value":0.0}, +//! {"path":"navigation.speedOverGround","value":3.087}]}]}"#)?; +//! // ...and a motor vessel a mile and a half north and east, steering west at 6 knots. +//! watch.ingest(r#"{"context":"vessels.urn:mrn:imo:mmsi:244000001","updates":[{"timestamp":"2026-09-25T12:00:00Z","values":[ +//! {"path":"","value":{"name":"ANNA"}}, +//! {"path":"navigation.position","value":{"latitude":53.025,"longitude":5.04155}}, +//! {"path":"navigation.courseOverGroundTrue","value":4.712389}, +//! {"path":"navigation.speedOverGround","value":3.087}, +//! {"path":"navigation.state","value":"motoring"}]}]}"#)?; +//! +//! let now = parse_timestamp("2026-09-25T12:00:01Z").unwrap(); +//! let (reports, emissions) = watch.assess(now); +//! assert_eq!(reports[0].level, Level::Alarm); +//! let Some(Emission::Notification { path, value }) = emissions.last() else { panic!() }; +//! assert_eq!(path, "notifications.navigation.closestApproach.urn:mrn:imo:mmsi:244000001"); +//! assert_eq!(value["state"], "alarm"); +//! assert!(value["message"].as_str().unwrap().contains("give way (Rule 15)")); +//! # Ok::<(), serde_json::Error>(()) +//! ``` +//! +//! Requires the standard library: JSON, and a picture as large as the server's. + +mod delta; +mod picture; +mod time; +mod watch; + +/// Runs the `README.md` example as a doctest. +#[cfg(doctest)] +#[doc = include_str!("../README.md")] +pub struct ReadmeExamples; + +pub use delta::{Delta, PathValue, Update}; +/// The clock [`Watch::assess`] takes. +pub use kinavis_kernel::time::{Instant, Utc}; +pub use picture::{Category, Picture, VesselState, SELF_CONTEXT}; +pub use time::parse_timestamp; +pub use watch::{message, Emission, Level, TargetReport, Watch, WatchConfig, NOTIFICATION_PATH}; diff --git a/crates/kinavis-signalk/src/picture.rs b/crates/kinavis-signalk/src/picture.rs new file mode 100644 index 0000000..7f3c902 --- /dev/null +++ b/crates/kinavis-signalk/src/picture.rs @@ -0,0 +1,262 @@ +//! The vessels a Signal K server knows, as the last value of each path this +//! crate uses. + +use std::collections::BTreeMap; + +use kinavis_colregs::VesselCategory; +use kinavis_kernel::angle::TrueCourse; +use kinavis_kernel::position::Position; +use kinavis_kernel::time::{Instant, Utc}; +use kinavis_kernel::units::Speed; +use serde_json::Value; + +use crate::delta::Delta; +use crate::time::parse_timestamp; + +/// Context of own vessel in Signal K. +pub const SELF_CONTEXT: &str = "vessels.self"; + +/// Last known state of one vessel. +#[derive(Debug, Clone, Default, PartialEq)] +pub struct VesselState { + /// Name, from the vessel's root `name`. + pub name: Option, + /// MMSI, from the vessel's root `mmsi`. + pub mmsi: Option, + /// `navigation.position`. + pub position: Option, + /// `navigation.courseOverGroundTrue`. + pub course: Option, + /// `navigation.speedOverGround`. + pub speed: Option, + /// `navigation.headingTrue`. + pub heading: Option, + /// `navigation.state`, as sent. + pub state: Option, + /// `design.aisShipType.id`. + pub ship_type: Option, + /// Latest timestamp of any update. + pub last_seen: Option>, +} + +impl VesselState { + /// Name, else MMSI, else the context it is known by. + #[must_use] + pub fn label<'a>(&'a self, context: &'a str) -> &'a str { + self.name + .as_deref() + .or(self.mmsi.as_deref()) + .unwrap_or(context) + } + + /// COLREGs category from `navigation.state`, else from the AIS ship type; + /// `None` if neither says. A vessel at anchor, moored or aground is not + /// under way and has no category under the steering rules. + #[must_use] + pub fn category(&self) -> Option { + if let Some(state) = self.state.as_deref() { + if let Some(category) = category_of_state(state) { + return Some(category); + } + } + match self.ship_type? { + 30 => Some(Category::UnderWay(VesselCategory::Fishing)), + 36 => Some(Category::UnderWay(VesselCategory::Sailing { tack: None })), + _ => None, + } + } + + fn apply(&mut self, path: &str, value: &Value) { + match path { + "" => { + if let Some(name) = value.get("name").and_then(Value::as_str) { + self.name = Some(name.trim().to_owned()); + } + if let Some(mmsi) = value.get("mmsi").and_then(Value::as_str) { + self.mmsi = Some(mmsi.to_owned()); + } + } + "name" => self.name = value.as_str().map(|name| name.trim().to_owned()), + "mmsi" => self.mmsi = value.as_str().map(str::to_owned), + "navigation.position" => { + self.position = value + .get("latitude") + .and_then(Value::as_f64) + .zip(value.get("longitude").and_then(Value::as_f64)) + .and_then(|(latitude, longitude)| { + Position::from_degrees(latitude, longitude).ok() + }); + } + "navigation.courseOverGroundTrue" => self.course = value.as_f64().and_then(direction), + "navigation.headingTrue" => self.heading = value.as_f64().and_then(direction), + "navigation.speedOverGround" => { + self.speed = value + .as_f64() + .filter(|speed| *speed >= 0.0) + .and_then(|speed| Speed::from_metres_per_second(speed).ok()); + } + "navigation.state" => self.state = value.as_str().map(str::to_owned), + "design.aisShipType" => { + self.ship_type = value + .get("id") + .and_then(Value::as_u64) + .and_then(|id| u32::try_from(id).ok()); + } + _ => {} + } + } +} + +/// What a vessel is, for the steering rules. +/// +/// `#[non_exhaustive]`; match with a wildcard arm. +#[non_exhaustive] +#[derive(Debug, Clone, Copy, PartialEq)] +pub enum Category { + /// Under way, of this category. + UnderWay(VesselCategory), + /// At anchor, moored, aground or otherwise not under way. + NotUnderWay, +} + +/// Category of a Signal K `navigation.state`; `None` for states that do not +/// say (motoring counts as power-driven). +fn category_of_state(state: &str) -> Option { + let category = match state { + "motoring" | "towing < 200m" | "towing > 200m" | "pushing" | "pilotage" => { + VesselCategory::PowerDriven + } + "sailing" => VesselCategory::Sailing { tack: None }, + "fishing" | "fishing-hampered" | "trawling" | "trawling-shooting" | "trawling-hauling" => { + VesselCategory::Fishing + } + "constrained by draft" => VesselCategory::ConstrainedByDraught, + "not under command" => VesselCategory::NotUnderCommand, + "anchored" | "moored" | "aground" | "not-under-way" => return Some(Category::NotUnderWay), + state if state.starts_with("restricted manouverability") || state == "mine clearance" => { + VesselCategory::RestrictedInAbilityToManoeuvre + } + _ => return None, + }; + Some(Category::UnderWay(category)) +} + +/// True direction of an angle in radians, wrapped to `[0, 360)`. +fn direction(radians: f64) -> Option { + radians + .is_finite() + .then(|| TrueCourse::from_degrees_wrapped(radians.to_degrees())) +} + +/// Every vessel seen, by context. +#[derive(Debug, Clone, Default)] +pub struct Picture { + vessels: BTreeMap, + own: Option, +} + +impl Picture { + /// Empty picture; own vessel is known as [`SELF_CONTEXT`] until + /// [`Picture::set_own_context`] names it. + #[must_use] + pub fn new() -> Self { + Self::default() + } + + /// Names own vessel's full context, `vessels.urn:mrn:imo:mmsi:…`, as the + /// server writes it in deltas. Values already held under either name are + /// merged. + pub fn set_own_context(&mut self, context: &str) { + self.own = Some(context.to_owned()); + if let Some(held) = self.vessels.remove(SELF_CONTEXT) { + self.vessels.entry(context.to_owned()).or_insert(held); + } + } + + /// Whether own vessel's full context is known, named or learnt. + #[must_use] + pub const fn own_is_named(&self) -> bool { + self.own.is_some() + } + + /// Context own vessel is held under. + #[must_use] + pub fn own_context(&self) -> &str { + self.own.as_deref().unwrap_or(SELF_CONTEXT) + } + + /// Whether a context is own vessel. + #[must_use] + pub fn is_own(&self, context: &str) -> bool { + context == SELF_CONTEXT || context == self.own_context() + } + + /// Applies a delta. Paths this crate does not use, and values it cannot + /// read, are ignored. + /// + /// Until [`Picture::set_own_context`] names own vessel, the first vessel + /// context with a position from own sensors (see + /// [`Update::is_own_sensor`]) is taken as own vessel. + /// + /// [`Update::is_own_sensor`]: crate::Update::is_own_sensor + pub fn apply(&mut self, delta: &Delta) { + if self.own.is_none() { + if let Some(context) = delta.context.as_deref() { + let own_position = delta.updates.iter().any(|update| { + update.is_own_sensor() + && update + .values + .iter() + .any(|value| value.path == "navigation.position") + }); + if own_position && context.starts_with("vessels.") && context != SELF_CONTEXT { + self.set_own_context(context); + } + } + } + let context = match delta.context.as_deref() { + None => self.own_context().to_owned(), + Some(context) if self.is_own(context) => self.own_context().to_owned(), + Some(context) => context.to_owned(), + }; + if !context.starts_with("vessels.") { + return; + } + let vessel = self.vessels.entry(context).or_default(); + for update in &delta.updates { + let at = update.timestamp.as_deref().and_then(parse_timestamp); + if let Some(at) = at { + vessel.last_seen = Some(vessel.last_seen.map_or(at, |seen| seen.max(at))); + } + for value in &update.values { + vessel.apply(&value.path, &value.value); + } + } + } + + /// Own vessel, if seen. + #[must_use] + pub fn own(&self) -> Option<&VesselState> { + self.vessels.get(self.own_context()) + } + + /// Other vessels, by context. + pub fn targets(&self) -> impl Iterator + '_ { + self.vessels + .iter() + .filter(|(context, _)| !self.is_own(context)) + .map(|(context, vessel)| (context.as_str(), vessel)) + } + + /// Number of vessels, own included. + #[must_use] + pub fn len(&self) -> usize { + self.vessels.len() + } + + /// Whether no vessel has been seen. + #[must_use] + pub fn is_empty(&self) -> bool { + self.vessels.is_empty() + } +} diff --git a/crates/kinavis-signalk/src/time.rs b/crates/kinavis-signalk/src/time.rs new file mode 100644 index 0000000..4b86e7f --- /dev/null +++ b/crates/kinavis-signalk/src/time.rs @@ -0,0 +1,92 @@ +//! Signal K timestamps: RFC 3339 in UTC, `2024-01-15T12:30:00.000Z`. + +use kinavis_kernel::time::{Civil, Instant, Utc}; + +/// Instant of a Signal K timestamp; `None` if it is not one. +/// +/// Accepts a `Z` or `+00:00` suffix and zero to nine decimals of a second. +/// Other offsets are refused: the specification requires UTC. +#[must_use] +pub fn parse_timestamp(text: &str) -> Option> { + let text = text + .strip_suffix('Z') + .or_else(|| text.strip_suffix("+00:00"))?; + let (date, time) = text.split_once('T')?; + let mut date = date.split('-'); + let year = date.next()?.parse().ok()?; + let month = date.next()?.parse().ok()?; + let day = date.next()?.parse().ok()?; + if date.next().is_some() { + return None; + } + let (clock, fraction) = time.split_once('.').unwrap_or((time, "")); + let mut clock = clock.split(':'); + let hour = clock.next()?.parse().ok()?; + let minute = clock.next()?.parse().ok()?; + let second = clock.next()?.parse().ok()?; + if clock.next().is_some() { + return None; + } + let nanos = nanos(fraction)?; + Instant::from_civil(Civil { + year, + month, + day, + hour, + minute, + second, + nanos, + }) + .ok() +} + +/// Nanoseconds of a decimal fraction of a second, at most nine digits. +fn nanos(fraction: &str) -> Option { + if fraction.len() > 9 || !fraction.bytes().all(|byte| byte.is_ascii_digit()) { + return None; + } + fraction + .bytes() + .chain(core::iter::repeat(b'0')) + .take(9) + .try_fold(0_u32, |nanos, digit| { + nanos + .checked_mul(10)? + .checked_add(u32::from(digit.wrapping_sub(b'0'))) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_signal_k_timestamp_is_read_to_the_nanosecond() { + let at = parse_timestamp("2024-01-15T12:30:00.250Z"); + assert_eq!( + at.map(|at| at.to_string()), + Some("2024-01-15T12:30:00.250 UTC".to_owned()) + ); + let whole = parse_timestamp("2024-01-15T12:30:00+00:00"); + assert_eq!( + whole.map(|at| at.to_string()), + Some("2024-01-15T12:30:00.000 UTC".to_owned()) + ); + } + + #[test] + fn anything_else_is_not_a_timestamp() { + for text in [ + "", + "2024-01-15", + "2024-01-15T12:30:00", + "2024-01-15T12:30:00+02:00", + "2024-02-30T12:30:00Z", + "2024-01-15T12:30:00.1234567890Z", + "2024-01-15T12:30:00.x1Z", + "2024-01-15-01T12:30:00Z", + ] { + assert_eq!(parse_timestamp(text), None, "{text}"); + } + } +} diff --git a/crates/kinavis-signalk/src/watch.rs b/crates/kinavis-signalk/src/watch.rs new file mode 100644 index 0000000..61512bf --- /dev/null +++ b/crates/kinavis-signalk/src/watch.rs @@ -0,0 +1,459 @@ +//! Collision watch over a Signal K picture: CPA and TCPA of every target, the +//! COLREGs ruling for the ones that matter, as deltas and notifications. + +use std::collections::BTreeMap; +use std::fmt::Write as _; +use std::time::Duration; + +use kinavis::relative_motion::{Contact, Vessel}; +use kinavis::sailings::rhumb_line; +use kinavis_colregs::{ + rule_of_the_road, ColregsConfig, Encounter, Party, Responsibility, Ruling, Situation, + VesselCategory, Visibility, +}; +use kinavis_kernel::angle::{Side, TrueCourse}; +use kinavis_kernel::position::Position; +use kinavis_kernel::time::{Instant, Utc}; +use kinavis_kernel::units::{Distance, Speed}; +use kinavis_traffic::{assess, CollisionAssessment, CollisionRisk, CpaPolicy}; +use serde::Deserialize; +use serde_json::{json, Value}; + +use crate::delta::Delta; +use crate::picture::{Category, Picture, VesselState}; + +/// Path of the notification for a target, followed by `.` and the target's +/// context without `vessels.`. +pub const NOTIFICATION_PATH: &str = "notifications.navigation.closestApproach"; + +/// Watch settings: vessel decisions, with defaults for a small vessel in +/// open water. Read from the plugin configuration, `camelCase`. +#[derive(Debug, Clone, Copy, PartialEq, Deserialize)] +#[serde(default, rename_all = "camelCase")] +pub struct WatchConfig { + /// CPA inside which an approach is dangerous, nautical miles. + pub cpa_limit_nm: f64, + /// TCPA inside which a dangerous approach is an alarm, not yet a + /// warning, minutes. + pub tcpa_limit_min: f64, + /// Age after which a target's last report is too old to assess, seconds. + pub stale_after_s: f64, + /// Visibility: the steering rules differ in restricted visibility. + pub restricted_visibility: bool, + /// Horizon of warnings, minutes: a dangerous CPA further off than this + /// is not reported. + pub warn_within_min: f64, +} + +impl Default for WatchConfig { + fn default() -> Self { + Self { + cpa_limit_nm: 1.0, + tcpa_limit_min: 20.0, + stale_after_s: 180.0, + restricted_visibility: false, + warn_within_min: 60.0, + } + } +} + +/// Severity of a target's approach, as a Signal K notification state. +/// +/// `#[non_exhaustive]`; match with a wildcard arm. +#[non_exhaustive] +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +pub enum Level { + /// Nothing to report. + Normal, + /// Dangerous CPA, but beyond the TCPA limit: developing. + Warn, + /// Dangerous CPA within the TCPA limit. + Alarm, +} + +impl Level { + /// Signal K notification state. + #[must_use] + pub const fn state(self) -> &'static str { + match self { + Self::Normal => "normal", + Self::Warn => "warn", + Self::Alarm => "alarm", + } + } + + fn of(assessment: &CollisionAssessment, horizon: Duration) -> Self { + match assessment.risk() { + CollisionRisk::Dangerous => Self::Alarm, + CollisionRisk::Developing if assessment.tcpa().is_some_and(|tcpa| tcpa <= horizon) => { + Self::Warn + } + _ => Self::Normal, + } + } +} + +/// One target's assessment. +#[derive(Debug, Clone, PartialEq)] +pub struct TargetReport { + /// Target's context. + pub context: String, + /// Name, MMSI or context. + pub label: String, + /// CPA, TCPA and risk. + pub assessment: CollisionAssessment, + /// COLREGs ruling, for a dangerous or developing approach between two + /// vessels under way. + pub ruling: Option, + /// Level of the approach. + pub level: Level, +} + +/// What to send to the server after an assessment. +/// +/// `#[non_exhaustive]`; match with a wildcard arm. +#[non_exhaustive] +#[derive(Debug, Clone, PartialEq)] +pub enum Emission { + /// A delta, `navigation.closestApproach` in a target's context. + Delta(Delta), + /// A notification on own vessel: its path and value. + Notification { + /// Full path, [`NOTIFICATION_PATH`]`.`. + path: String, + /// `state`, `method` and `message`. + value: Value, + }, +} + +/// The watch: the picture, and the level each target's notification stands at. +#[derive(Debug, Clone)] +pub struct Watch { + config: WatchConfig, + picture: Picture, + standing: BTreeMap, +} + +impl Watch { + /// Empty watch. + #[must_use] + pub fn new(config: WatchConfig) -> Self { + Self { + config, + picture: Picture::new(), + standing: BTreeMap::new(), + } + } + + /// Settings. + #[must_use] + pub const fn config(&self) -> &WatchConfig { + &self.config + } + + /// The picture, to name own vessel's context or read the vessels. + #[must_use] + pub const fn picture(&self) -> &Picture { + &self.picture + } + + /// The picture, to name own vessel's context. + pub fn picture_mut(&mut self) -> &mut Picture { + &mut self.picture + } + + /// Reads a delta into the picture. + /// + /// # Errors + /// + /// The JSON error if the text is not a delta. + pub fn ingest(&mut self, json: &str) -> Result<(), serde_json::Error> { + let delta: Delta = serde_json::from_str(json)?; + self.picture.apply(&delta); + Ok(()) + } + + /// Assesses every target against own vessel at `now`. + /// + /// `now` is the caller's clock, not a timestamp from the deltas: sources + /// stamp their updates by different clocks (a GNSS receiver by its fix, + /// the server by reception). A vessel is assessed if its last update is + /// no older than [`WatchConfig::stale_after_s`] at `now`; one stamped + /// ahead of `now` counts as current. + /// + /// Returns the reports, closest CPA first, and what to send: a + /// `navigation.closestApproach` delta for every closing target, and a + /// notification wherever a target's level changed. Empty while own + /// vessel has no current position and motion. + pub fn assess(&mut self, now: Instant) -> (Vec, Vec) { + let reports = self.reports(now); + let mut emissions = Vec::new(); + let mut levels = BTreeMap::new(); + for report in &reports { + if let (Some(cpa), Some(tcpa)) = + (report.assessment.cpa_distance(), report.assessment.tcpa()) + { + emissions.push(Emission::Delta(Delta::single( + Some(report.context.clone()), + "navigation.closestApproach", + json!({ "distance": cpa.metres(), "timeTo": tcpa.as_secs_f64() }), + ))); + } + levels.insert(report.context.clone(), report.level); + if self + .standing + .get(&report.context) + .copied() + .unwrap_or(Level::Normal) + != report.level + { + emissions.push(notification( + report.context.as_str(), + report.level, + &message(report), + )); + } + } + // Targets that stood above normal and are no longer assessed: stale, + // gone or opening. + for (context, level) in &self.standing { + if *level != Level::Normal && !levels.contains_key(context) { + emissions.push(notification(context, Level::Normal, "no longer assessed")); + } + } + levels.retain(|_, level| *level != Level::Normal); + self.standing = levels; + (reports, emissions) + } + + fn reports(&self, now: Instant) -> Vec { + let stale = Duration::try_from_secs_f64(self.config.stale_after_s).unwrap_or(Duration::MAX); + let Some(own) = self.picture.own().filter(|own| is_fresh(own, now, stale)) else { + return Vec::new(); + }; + let (Some(own_position), Some(own_motion)) = (own.position, motion(own)) else { + return Vec::new(); + }; + let Some(policy) = self.policy() else { + return Vec::new(); + }; + let mut reports: Vec = self + .picture + .targets() + .filter(|(_, target)| is_fresh(target, now, stale)) + .filter_map(|(context, target)| { + let report = + self.report(own, own_position, own_motion, &policy, context, target)?; + Some(report) + }) + .collect(); + reports.sort_by(|a, b| closeness(&a.assessment).total_cmp(&closeness(&b.assessment))); + reports + } + + fn report( + &self, + own: &VesselState, + own_position: Position, + own_motion: Vessel, + policy: &CpaPolicy, + context: &str, + target: &VesselState, + ) -> Option { + let position = target.position?; + let target_motion = motion(target)?; + let line = rhumb_line(own_position, position).ok()?; + let contact = Contact { + bearing: line.initial_course, + range: line.distance, + }; + let assessment = assess(own_motion, contact, target_motion, policy).ok()?; + let horizon = Duration::try_from_secs_f64(self.config.warn_within_min * 60.0) + .unwrap_or(Duration::MAX); + let level = Level::of(&assessment, horizon); + // The steering rules are for vessels under way: none for own vessel + // at rest (see `AT_REST_KNOTS`). + let ruling = if level == Level::Normal || own_motion.speed == Speed::ZERO { + None + } else { + self.ruling(own, own_motion, target, target_motion, contact) + }; + Some(TargetReport { + context: context.to_owned(), + label: target.label(context).to_owned(), + assessment, + ruling, + level, + }) + } + + fn ruling( + &self, + own: &VesselState, + own_motion: Vessel, + target: &VesselState, + target_motion: Vessel, + contact: Contact, + ) -> Option { + // Own vessel under way unless it says otherwise; a target must say + // nothing contrary either, and counts as power-driven if silent. + let own_category = under_way(own)?; + let target_category = under_way(target)?; + let mut own_party = Party::new(own_motion, own_category); + if let Some(heading) = own.heading { + own_party = own_party.with_heading(heading); + } + let mut target_party = Party::new(target_motion, target_category); + if let Some(heading) = target.heading { + target_party = target_party.with_heading(heading); + } + let visibility = if self.config.restricted_visibility { + Visibility::Restricted + } else { + Visibility::InSight + }; + let situation = Situation::new(own_party, target_party, contact, visibility); + rule_of_the_road(&situation, &ColregsConfig::STANDARD).ok() + } + + fn policy(&self) -> Option { + let limit = Distance::from_nautical_miles(self.config.cpa_limit_nm).ok()?; + let tcpa = Duration::try_from_secs_f64(self.config.tcpa_limit_min * 60.0).ok()?; + CpaPolicy::new(limit, tcpa).ok() + } +} + +/// Speed below which a vessel is taken as at rest, knots: a receiver at rest +/// reports a few centimetres a second of noise, and a course that is noise +/// too, or none. +const AT_REST_KNOTS: f64 = 0.5; + +/// Course and speed; a vessel at rest may report no course. +fn motion(vessel: &VesselState) -> Option { + let speed = vessel.speed?; + if speed.knots() < AT_REST_KNOTS { + return Some(Vessel { + course: TrueCourse::NORTH, + speed: Speed::ZERO, + }); + } + Some(Vessel { + course: vessel.course?, + speed, + }) +} + +/// Category under way: the vessel's own, power-driven if it says nothing, +/// `None` if it is not under way. +fn under_way(vessel: &VesselState) -> Option { + match vessel.category() { + Some(Category::UnderWay(category)) => Some(category), + Some(Category::NotUnderWay) => None, + None => Some(VesselCategory::PowerDriven), + } +} + +/// Whether the last update is no older than `stale`; one stamped ahead of +/// `now` is current. +fn is_fresh(vessel: &VesselState, now: Instant, stale: Duration) -> bool { + vessel.last_seen.is_some_and(|seen| { + now.checked_duration_since(seen) + .is_none_or(|age| age <= stale) + }) +} + +/// Closest first; opening targets last. +fn closeness(assessment: &CollisionAssessment) -> f64 { + assessment + .cpa_distance() + .map_or(f64::INFINITY, Distance::nautical_miles) +} + +fn notification(context: &str, level: Level, message: &str) -> Emission { + let id = context.strip_prefix("vessels.").unwrap_or(context); + let method: &[&str] = match level { + Level::Alarm => &["visual", "sound"], + Level::Warn => &["visual"], + Level::Normal => &[], + }; + Emission::Notification { + path: format!("{NOTIFICATION_PATH}.{id}"), + value: json!({ "state": level.state(), "method": method, "message": message }), + } +} + +/// Notification text: the approach, and the ruling if there is one. +#[must_use] +pub fn message(report: &TargetReport) -> String { + let mut out = report.label.clone(); + match (report.assessment.cpa_distance(), report.assessment.tcpa()) { + (Some(cpa), Some(tcpa)) => { + let seconds = tcpa.as_secs(); + let _ = write!( + out, + ": CPA {:.2} NM in {}:{:02}", + cpa.nautical_miles(), + seconds / 60, + seconds % 60 + ); + } + _ => out.push_str(": opening"), + } + if let Some(ruling) = &report.ruling { + let _ = write!( + out, + "; {}, {} ({})", + encounter(ruling.encounter()), + duty(ruling.responsibility()), + ruling.rule() + ); + let may = ruling.manoeuvre(); + let turn = match (may.starboard, may.port) { + (true, true) => Some("alter to starboard or port"), + (true, false) => Some("alter to starboard"), + (false, true) => Some("alter to port"), + (false, false) => None, + }; + let options: Vec<&str> = [ + turn, + may.slow_down.then_some("slow down"), + may.hold.then_some("hold course and speed"), + ] + .into_iter() + .flatten() + .collect(); + if !options.is_empty() { + let _ = write!(out, ": {}", options.join(", or ")); + } + } + out +} + +fn encounter(encounter: Encounter) -> String { + match encounter { + Encounter::Overtaking => "overtaking".to_owned(), + Encounter::BeingOvertaken => "being overtaken".to_owned(), + Encounter::HeadOn => "head-on".to_owned(), + Encounter::Crossing { target_on } => format!("crossing, target on {}", side(target_on)), + other => format!("{other:?}"), + } +} + +const fn side(side: Side) -> &'static str { + match side { + Side::Port => "port", + Side::Starboard => "starboard", + Side::Ahead => "ahead", + Side::Astern => "astern", + _ => "abeam", + } +} + +const fn duty(responsibility: Responsibility) -> &'static str { + match responsibility { + Responsibility::GiveWay => "give way", + Responsibility::StandOn => "stand on", + Responsibility::Both => "both alter", + _ => "no rule decides", + } +} diff --git a/crates/kinavis-signalk/tests/watch.rs b/crates/kinavis-signalk/tests/watch.rs new file mode 100644 index 0000000..f78ed95 --- /dev/null +++ b/crates/kinavis-signalk/tests/watch.rs @@ -0,0 +1,343 @@ +//! The watch over deltas as a Signal K server sends them. + +#![allow( + clippy::unwrap_used, + clippy::panic, + clippy::indexing_slicing, + clippy::float_cmp +)] + +use kinavis_kernel::time::{Instant, Utc}; +use kinavis_signalk::{parse_timestamp, Emission, Level, Watch, WatchConfig}; +use serde_json::{json, Value}; + +const TARGET: &str = "vessels.urn:mrn:imo:mmsi:244000001"; +const KNOTS_6: f64 = 3.086_666; +const WEST: f64 = 3.0 * std::f64::consts::FRAC_PI_2; + +fn at(time: &str) -> Instant { + parse_timestamp(time).unwrap() +} + +fn delta(context: &str, time: &str, values: &Value) -> String { + json!({ "context": context, "updates": [{ "timestamp": time, "values": values }] }).to_string() +} + +/// Own vessel at 53°N 5°E steering north at 6 knots. +fn own(watch: &mut Watch, context: &str, time: &str) { + watch + .ingest(&delta( + context, + time, + &json!([ + { "path": "navigation.position", "value": { "latitude": 53.0, "longitude": 5.0 } }, + { "path": "navigation.courseOverGroundTrue", "value": 0.0 }, + { "path": "navigation.speedOverGround", "value": KNOTS_6 }, + ]), + )) + .unwrap(); +} + +/// A target at a position, steering west at 6 knots, with extra values. +fn target(watch: &mut Watch, time: &str, latitude: f64, longitude: f64, extra: &[Value]) { + let mut values = vec![ + json!({ "path": "navigation.position", "value": { "latitude": latitude, "longitude": longitude } }), + json!({ "path": "navigation.courseOverGroundTrue", "value": WEST }), + json!({ "path": "navigation.speedOverGround", "value": KNOTS_6 }), + ]; + values.extend_from_slice(extra); + watch + .ingest(&delta(TARGET, time, &Value::Array(values))) + .unwrap(); +} + +fn notifications(emissions: &[Emission]) -> Vec<(String, Value)> { + emissions + .iter() + .filter_map(|emission| match emission { + Emission::Notification { path, value } => Some((path.clone(), value.clone())), + _ => None, + }) + .collect() +} + +#[test] +fn a_crossing_target_on_the_starboard_bow_raises_one_alarm_until_it_clears() { + let mut watch = Watch::new(WatchConfig::default()); + own(&mut watch, "vessels.self", "2026-09-25T12:00:00Z"); + target(&mut watch, "2026-09-25T12:00:00Z", 53.025, 5.041_55, &[]); + + let (reports, emissions) = watch.assess(at("2026-09-25T12:00:01Z")); + assert_eq!(reports.len(), 1); + assert_eq!(reports[0].level, Level::Alarm); + let raised = notifications(&emissions); + assert_eq!(raised.len(), 1); + assert_eq!( + raised[0].0, + "notifications.navigation.closestApproach.urn:mrn:imo:mmsi:244000001" + ); + assert_eq!(raised[0].1["state"], "alarm"); + assert_eq!(raised[0].1["method"], json!(["visual", "sound"])); + // A silent target counts as power-driven: Rule 15, own vessel gives way. + let message = raised[0].1["message"].as_str().unwrap(); + assert!( + message.contains("crossing, target on starboard, give way (Rule 15)"), + "{message}" + ); + // CPA and TCPA in the target's context, in metres and seconds. + let Some(Emission::Delta(cpa)) = emissions.first() else { + panic!("no closestApproach delta") + }; + assert_eq!(cpa.context.as_deref(), Some(TARGET)); + let value = &cpa.updates[0].values[0].value; + assert!( + (value["timeTo"].as_f64().unwrap() - 900.0).abs() < 5.0, + "{value}" + ); + + // The same picture again: the alarm stands, nothing new is notified. + let (_, emissions) = watch.assess(at("2026-09-25T12:00:01Z")); + assert!(notifications(&emissions).is_empty()); + + // The target turns away north-east, opening: the alarm is cleared. + watch + .ingest(&delta( + TARGET, + "2026-09-25T12:00:10Z", + &json!([{ "path": "navigation.courseOverGroundTrue", "value": 0.8 }]), + )) + .unwrap(); + let (_, emissions) = watch.assess(at("2026-09-25T12:00:11Z")); + let cleared = notifications(&emissions); + assert_eq!(cleared.len(), 1); + assert_eq!(cleared[0].1["state"], "normal"); +} + +#[test] +fn a_target_gone_silent_is_cleared_once_stale() { + let mut watch = Watch::new(WatchConfig::default()); + own(&mut watch, "vessels.self", "2026-09-25T12:00:00Z"); + target(&mut watch, "2026-09-25T12:00:00Z", 53.025, 5.041_55, &[]); + assert_eq!( + notifications(&watch.assess(at("2026-09-25T12:00:01Z")).1)[0].1["state"], + "alarm" + ); + + // Own vessel reports for four minutes; the target does not. + own(&mut watch, "vessels.self", "2026-09-25T12:04:00Z"); + let (reports, emissions) = watch.assess(at("2026-09-25T12:04:00Z")); + assert!(reports.is_empty()); + let cleared = notifications(&emissions); + assert_eq!(cleared[0].1["state"], "normal"); + assert_eq!(cleared[0].1["message"], "no longer assessed"); +} + +#[test] +fn a_sailing_vessel_by_its_ais_ship_type_is_given_way_to_under_rule_18() { + let mut watch = Watch::new(WatchConfig::default()); + own(&mut watch, "vessels.self", "2026-09-25T12:00:00Z"); + // On the port bow: by Rule 15 alone own vessel would stand on. + target( + &mut watch, + "2026-09-25T12:00:00Z", + 53.025, + 4.958_45, + &[json!({ "path": "design.aisShipType", "value": { "id": 36, "name": "Sailing" } })], + ); + watch + .ingest(&delta( + TARGET, + "2026-09-25T12:00:00Z", + &json!([{ "path": "navigation.courseOverGroundTrue", "value": std::f64::consts::FRAC_PI_2 }]), + )) + .unwrap(); + let (reports, _) = watch.assess(at("2026-09-25T12:00:01Z")); + let message = kinavis_signalk::message(&reports[0]); + assert!(message.contains("give way (Rule 18)"), "{message}"); +} + +#[test] +fn a_vessel_at_anchor_gets_a_cpa_but_no_ruling() { + let mut watch = Watch::new(WatchConfig::default()); + own(&mut watch, "vessels.self", "2026-09-25T12:00:00Z"); + // Dead ahead, a mile off, at anchor. + watch + .ingest(&delta( + TARGET, + "2026-09-25T12:00:00Z", + &json!([ + { "path": "navigation.position", "value": { "latitude": 53.016_67, "longitude": 5.0 } }, + { "path": "navigation.speedOverGround", "value": 0.0 }, + { "path": "navigation.state", "value": "anchored" }, + ]), + )) + .unwrap(); + let (reports, _) = watch.assess(at("2026-09-25T12:00:01Z")); + assert_eq!(reports[0].level, Level::Alarm); + assert_eq!(reports[0].ruling, None); +} + +#[test] +fn own_vessel_is_known_by_its_full_context_once_named() { + let own_context = "vessels.urn:mrn:imo:mmsi:230000001"; + let mut watch = Watch::new(WatchConfig::default()); + own(&mut watch, "vessels.self", "2026-09-25T12:00:00Z"); + watch.picture_mut().set_own_context(own_context); + // The server writes own vessel's deltas under its full context. + own(&mut watch, own_context, "2026-09-25T12:00:01Z"); + target(&mut watch, "2026-09-25T12:00:01Z", 53.025, 5.041_55, &[]); + let (reports, _) = watch.assess(at("2026-09-25T12:00:02Z")); + assert_eq!(reports.len(), 1, "own vessel is not its own target"); + assert_eq!(watch.picture().len(), 2); +} + +#[test] +fn nothing_is_assessed_without_own_motion_and_bad_input_is_an_error() { + let mut watch = Watch::new(WatchConfig::default()); + target(&mut watch, "2026-09-25T12:00:00Z", 53.025, 5.041_55, &[]); + assert_eq!( + watch.assess(at("2026-09-25T12:00:01Z")), + (Vec::new(), Vec::new()) + ); + assert!(watch.ingest("not json").is_err()); + // Paths the watch does not use, and unreadable values, are ignored. + watch + .ingest(&delta( + TARGET, + "2026-09-25T12:00:01Z", + &json!([ + { "path": "environment.wind.speedApparent", "value": 5.0 }, + { "path": "navigation.speedOverGround", "value": "fast" }, + ]), + )) + .unwrap(); +} + +#[test] +fn the_configuration_reads_from_the_plugin_settings() { + let config: WatchConfig = + serde_json::from_str(r#"{"cpaLimitNm": 0.5, "restrictedVisibility": true}"#).unwrap(); + assert_eq!(config.cpa_limit_nm, 0.5); + assert!(config.restricted_visibility); + assert_eq!(config.tcpa_limit_min, WatchConfig::default().tcpa_limit_min); +} + +#[test] +fn own_vessel_alongside_with_noise_and_no_course_is_at_rest() { + let mut watch = Watch::new(WatchConfig::default()); + // A receiver at rest: 5 mm/s of noise, course null. + watch + .ingest(&delta( + "vessels.self", + "2026-09-25T12:00:00Z", + &json!([ + { "path": "navigation.position", "value": { "latitude": 53.0, "longitude": 5.0 } }, + { "path": "navigation.courseOverGroundTrue", "value": null }, + { "path": "navigation.speedOverGround", "value": 0.0046 }, + ]), + )) + .unwrap(); + target(&mut watch, "2026-09-25T12:00:00Z", 53.0, 5.02, &[]); + let (reports, _) = watch.assess(at("2026-09-25T12:00:01Z")); + assert_eq!(reports.len(), 1); + assert!(reports[0].assessment.cpa_distance().is_some()); +} + +#[test] +fn own_vessel_is_learnt_from_its_own_sensors() { + let own_context = "vessels.urn:mrn:signalk:uuid:cdf30fb9-28bf-4d10-8817-04a6b3d96e33"; + let mut watch = Watch::new(WatchConfig::default()); + // A target first, from AIS: not own vessel. + watch + .ingest( + &json!({ "context": TARGET, "updates": [{ + "timestamp": "2026-09-25T12:00:00Z", + "source": { "type": "NMEA0183", "sentence": "VDM", "talker": "AI" }, + "values": [ + { "path": "navigation.position", "value": { "latitude": 53.025, "longitude": 5.041_55 } }, + { "path": "navigation.courseOverGroundTrue", "value": WEST }, + { "path": "navigation.speedOverGround", "value": KNOTS_6 }, + ]}]}) + .to_string(), + ) + .unwrap(); + assert!(!watch.picture().own_is_named()); + // Then the GNSS receiver: its RMC is own vessel, under the server's + // identity for it. + watch + .ingest( + &json!({ "context": own_context, "updates": [{ + "timestamp": "2026-09-25T12:00:00Z", + "source": { "type": "NMEA0183", "sentence": "RMC", "talker": "GP" }, + "values": [ + { "path": "navigation.position", "value": { "latitude": 53.0, "longitude": 5.0 } }, + { "path": "navigation.courseOverGroundTrue", "value": 0.0 }, + { "path": "navigation.speedOverGround", "value": KNOTS_6 }, + ]}]}) + .to_string(), + ) + .unwrap(); + assert_eq!(watch.picture().own_context(), own_context); + let (reports, _) = watch.assess(at("2026-09-25T12:00:01Z")); + assert_eq!(reports.len(), 1); + assert_eq!(reports[0].context, TARGET); +} + +#[test] +fn a_receiver_clock_ahead_of_the_watch_is_current_and_own_vessel_can_go_stale() { + let mut watch = Watch::new(WatchConfig::default()); + // The GNSS receiver stamps by its fix, eleven hours ahead of the server. + own(&mut watch, "vessels.self", "2026-09-25T23:00:00Z"); + target(&mut watch, "2026-09-25T12:00:00Z", 53.025, 5.041_55, &[]); + assert_eq!(watch.assess(at("2026-09-25T12:00:01Z")).0.len(), 1); + + // A receiver silent for four minutes: nothing is assessed against it. + let mut watch = Watch::new(WatchConfig::default()); + own(&mut watch, "vessels.self", "2026-09-25T12:00:00Z"); + target(&mut watch, "2026-09-25T12:04:00Z", 53.025, 5.041_55, &[]); + assert!(watch.assess(at("2026-09-25T12:04:01Z")).0.is_empty()); +} + +#[test] +fn own_vessel_at_rest_gets_cpas_but_no_ruling_and_far_warnings_are_not_raised() { + let mut watch = Watch::new(WatchConfig::default()); + watch + .ingest(&delta( + "vessels.self", + "2026-09-25T12:00:00Z", + &json!([ + { "path": "navigation.position", "value": { "latitude": 53.0, "longitude": 5.0 } }, + { "path": "navigation.courseOverGroundTrue", "value": 2.1 }, + { "path": "navigation.speedOverGround", "value": 0.05 }, + ]), + )) + .unwrap(); + // Two miles east, steering west at 6 knots: straight at own vessel. + target(&mut watch, "2026-09-25T12:00:00Z", 53.0, 5.055_4, &[]); + let (reports, _) = watch.assess(at("2026-09-25T12:00:01Z")); + assert_eq!( + reports[0].level, + Level::Warn, + "TCPA 20 min: beyond the alarm limit" + ); + assert_eq!(reports[0].ruling, None); + + // The same approach from forty miles: CPA is dangerous, but not within + // the hour. + let mut watch = Watch::new(WatchConfig::default()); + own(&mut watch, "vessels.self", "2026-09-25T12:00:00Z"); + watch + .ingest(&delta( + TARGET, + "2026-09-25T12:00:00Z", + &json!([ + { "path": "navigation.position", "value": { "latitude": 53.666_7, "longitude": 5.0 } }, + { "path": "navigation.courseOverGroundTrue", "value": std::f64::consts::PI }, + { "path": "navigation.speedOverGround", "value": KNOTS_6 }, + ]), + )) + .unwrap(); + let (reports, emissions) = watch.assess(at("2026-09-25T12:00:01Z")); + assert_eq!(reports[0].level, Level::Normal); + assert!(notifications(&emissions).is_empty()); +} diff --git a/crates/kinavis/README.md b/crates/kinavis/README.md index 93bc23c..cfd66d1 100644 --- a/crates/kinavis/README.md +++ b/crates/kinavis/README.md @@ -67,6 +67,7 @@ build on top of it, each `no_std` and without an allocator: | [`kinavis-traffic`](https://crates.io/crates/kinavis-traffic) | target tracking from radar and AIS, CPA and TCPA, the avoiding manoeuvre | | [`kinavis-colregs`](https://crates.io/crates/kinavis-colregs) | the steering and sailing rules of the COLREGs as data | | [`kinavis-alerts`](https://crates.io/crates/kinavis-alerts) | bridge alert management: alarms, warnings and cautions, with acknowledgement | +| [`kinavis-signalk`](https://crates.io/crates/kinavis-signalk) | Signal K deltas in, CPA, TCPA and the COLREGs ruling out, for a Signal K server plugin | ## Features