Runtime Classes

Business logic and data processing for cruise planning.

cruiseplan.runtime package.

This package contains the runtime classes and business logic for cruise planning:

Cruise: Main container for cruise configuration, validation, and data management
Leg: Represents discrete working areas or time periods in a cruise
BaseOperation: Base class for individual cruise operations
Cluster: Container for operation boundary management and reordering
validation: Pydantic models and validation schemas for cruise data

The runtime package provides the fundamental building blocks for defining and managing oceanographic cruise plans, including station definitions, transit routes, and operational sequences.

Submodules

cruiseplan.runtime.cruise module

Core cruise management and organizational runtime classes.

This module provides the main CruiseInstance class for loading, validating, and managing cruise configurations from YAML files, along with the organizational runtime classes (Leg, Cluster) that form the hierarchical structure for cruise execution. The BaseOrganizationUnit abstract base class provides the common interface for all organizational units in the cruise planning hierarchy.

class cruiseplan.runtime.cruise.CruiseInstance(config_path: str | Path)[source]

Bases: object

The main container object for cruise planning.

Responsible for parsing YAML configuration files, validating the schema using Pydantic models, and resolving string references to full objects from the catalog registries.

config_path

Absolute path to the configuration file.

Type:: Path

raw_data

Raw dictionary data loaded from the YAML file.

Type:: Dict[str, Any]

config

Validated Pydantic configuration object.

Type:: CruiseConfig

point_registry

Dictionary mapping point names to PointDefinition objects.

Type:: Dict[str, PointDefinition]

line_registry

Dictionary mapping line names to LineDefinition objects.

Type:: Dict[str, LineDefinition]

port_registry

Dictionary mapping port names to PointDefinition objects.

Type:: Dict[str, PointDefinition]

area_registry

Dictionary mapping area names to AreaDefinition objects.

Type:: Dict[str, AreaDefinition]

runtime_legs

List of runtime Leg objects converted from LegDefinition objects.

Type:: List[Leg]

__init__(config_path: str | Path)[source]

Initialize a CruiseInstance object from a YAML configuration file.

Performs three main operations: 1. Loads and validates the YAML configuration using Pydantic 2. Builds registries for points and lines 3. Resolves string references to full objects

Parameters:

config_path (Union[str, Path]) – Path to the YAML configuration file containing cruise definition.

Raises:

FileNotFoundError – If the configuration file does not exist.
YAMLIOError – If the YAML file cannot be parsed.
ValidationError – If the configuration does not match the expected schema.
ReferenceError – If referenced points or lines are not found in the catalog.

add_coordinate_displays(coord_format: str = 'ddm') → int[source]

Add human-readable coordinate display fields for final YAML output.

This method adds formatted coordinate annotations that will appear in the YAML output but don’t affect the core cruise data. This is for display enhancement only.

Parameters:: coord_format (str, optional) – Coordinate format to use for display. Default is “ddm”.
Returns:: Number of coordinate display fields added.
Return type:: int

add_station_defaults() → int[source]

Add missing defaults to station definitions.

This method adds default duration to mooring operations and other stations that lack required default values.

Returns:: Number of station defaults added.
Return type:: int

enrich_depths(bathymetry_source: str = 'etopo2022', bathymetry_dir: str = 'data') → set[str][source]

Add bathymetry depths to stations that are missing water_depth values.

This method modifies the point_registry directly by adding water depth information from bathymetry datasets to stations that don’t have depth values or have placeholder values.

Parameters:

bathymetry_source (str, optional) – Bathymetry dataset to use. Default is “etopo2022”.
bathymetry_dir (str, optional) – Directory containing bathymetry data. Default is “data”.

Returns:

Set of station names that had depths added.

Return type:

set[str]

expand_ports() → dict[str, int][source]

Expand global port references into full PortDefinition objects.

This method finds string port references and expands them into full PortDefinition objects with coordinates and other metadata from the global ports database.

Returns:: Dictionary with expansion summary: - ports_expanded: Number of global ports expanded - leg_ports_expanded: Number of leg ports expanded
Return type:: dict[str, int]

expand_sections(default_depth: float = -9999.0) → dict[str, int][source]

Expand CTD sections into individual station definitions.

This method finds CTD sections in lines catalog and expands them into individual stations, adding them to the point_registry. This is structural enrichment that modifies the cruise configuration.

Parameters:: default_depth (float, optional) – Default depth value for expanded stations. Default is -9999.0.
Returns:: Dictionary with expansion summary: - sections_expanded: Number of sections expanded - stations_from_expansion: Number of stations created
Return type:: dict[str, int]

classmethod from_dict(config_dict: dict[str, Any]) → CruiseInstance[source]

Create a CruiseInstance from a dictionary without file I/O.

This class method provides single source of truth functionality by creating a CruiseInstance object directly from a configuration dictionary, eliminating the need for temporary file creation during enrichment operations.

Parameters:

config_dict (Dict[str, Any]) – Dictionary containing cruise configuration data (e.g., from YAML parsing).

Returns:

New CruiseInstance with all registries built and references resolved.

Return type:

CruiseInstance

Raises:

ValidationError – If the configuration does not match the expected schema.
ReferenceError – If referenced points or lines are not found in the catalog.

Examples

>>> config = {
...     "cruise_name": "Test Cruise",
...     "default_vessel_speed": 10.0,
...     "points": [{"name": "P1", "latitude": 60.0, "longitude": 5.0}],
...     "legs": [{"name": "Leg1", "departure_port": "Bergen", "arrival_port": "Tromsø"}]
... }
>>> cruise = CruiseInstance.from_dict(config)
>>> cruise.config.cruise_name
'Test Cruise'

to_commented_dict() → dict[str, Any][source]

Export CruiseInstance configuration to a structured dictionary with comment preservation.

This method provides the foundation for YAML output with canonical field ordering and comment preservation capabilities. Returns a dictionary that can be processed by ruamel.yaml for structured output with comments.

Returns:: Dictionary with canonical field ordering suitable for YAML export with comment preservation.
Return type:: Dict[str, Any]

Notes

The output dictionary follows canonical ordering: 1. Cruise Metadata (cruise_name, description, start_date, start_time) 2. Vessel Parameters (default_vessel_speed, turnaround_time, etc.) 3. Calculation Settings (calculate_*, day_start_hour, etc.) 4. Catalog Definitions (points, lines, areas, ports) 5. Schedule Organization (legs)

Comment preservation is handled at the YAML layer using ruamel.yaml with end-of-line and section header comment support.

to_yaml(output_path: str | Path, enrichment_command: str | None = None) → None[source]

Export CruiseInstance configuration to YAML file with canonical ordering.

This method provides direct YAML export capability with standardized field ordering and basic comment preservation. Uses ruamel.yaml for structured output that maintains readability.

Parameters:

output_path (Union[str, Path]) – Path where the YAML file should be written.
enrichment_command (Optional[str]) – The enrichment command that was used to create this file, for documentation.

Raises:

IOError – If the output file cannot be written.

Examples

>>> cruise = CruiseInstance.from_dict(config_dict)
>>> cruise.to_yaml("enhanced_cruise.yaml")

Notes

The exported YAML follows canonical field ordering and includes section comments for improved readability. This replaces the need for dual state management during enrichment operations.

cruiseplan.runtime.organizational module

Organizational unit classes for cruise planning.

This module contains the hierarchical organizational classes (BaseOrganizationUnit, Cluster, Leg) that form the structural framework for cruise execution and scheduling.

class cruiseplan.runtime.organizational.BaseOrganizationUnit[source]

Bases: ABC

Abstract base class for organizational units in cruise planning.

Provides common interface for hierarchical organization units (Leg, Cluster) that can contain operations and support parameter inheritance, boundary management, and geographic routing.

All organizational units share common capabilities: - Entry/exit point management for routing - Operation counting and duration calculation - Parameter inheritance from parent units - Reordering policies for scheduling flexibility

abstractmethod allows_reordering() → bool[source]: Check if this unit allows operation reordering.

abstractmethod get_all_operations() → list[BaseOperation][source]: Get all operations within this organizational unit.

abstractmethod get_entry_point() → tuple[float, float] | None[source]: Get geographic entry point for this unit.

abstractmethod get_exit_point() → tuple[float, float] | None[source]: Get geographic exit point for this unit.

abstractmethod get_operation_count() → int[source]: Get total number of operations in this unit.

class cruiseplan.runtime.organizational.Cluster(name: str, description: str | None = None, strategy: StrategyEnum = StrategyEnum.SEQUENTIAL, ordered: bool = True)[source]

Bases: BaseOrganizationUnit

Runtime container for operation boundary management during scheduling.

Clusters define boundaries for operation shuffling/reordering. Operations within a cluster can be reordered according to the cluster’s strategy, but cannot be mixed with operations from other clusters or the parent leg.

This provides scientific flexibility (weather-dependent reordering) while maintaining operational safety (critical sequences protected).

name

Unique identifier for this cluster.

Type:: str

description

Human-readable description of the cluster’s purpose.

Type:: Optional[str]

strategy

Scheduling strategy for operations within this cluster.

Type:: StrategyEnum

ordered

Whether operations should maintain their defined order.

Type:: bool

operations

List of operations contained within this cluster boundary.

Type:: List[BaseOperation]

Examples

>>> # Weather-flexible CTD cluster
>>> ctd_cluster = Cluster(
...     name="CTD_Survey",
...     description="CTD operations that can be reordered for weather",
...     strategy=StrategyEnum.SEQUENTIAL,
...     ordered=False  # Allow weather-based reordering
... )

>>> # Critical mooring sequence cluster
>>> mooring_cluster = Cluster(
...     name="Mooring_Deployment",
...     description="Critical mooring sequence - strict order required",
...     strategy=StrategyEnum.SEQUENTIAL,
...     ordered=True  # Maintain deployment order for safety
... )

__init__(name: str, description: str | None = None, strategy: StrategyEnum = StrategyEnum.SEQUENTIAL, ordered: bool = True)[source]

Initialize a Cluster with the specified parameters.

Parameters:

name (str) – Unique identifier for this cluster.
description (Optional[str], optional) – Human-readable description of the cluster’s purpose.
strategy (StrategyEnum, optional) – Scheduling strategy for operations within this cluster. Default is StrategyEnum.SEQUENTIAL.
ordered (bool, optional) – Whether operations should maintain their defined order. Default is True (strict ordering).

__repr__() → str[source]: String representation for debugging.

__str__() → str[source]: Human-readable string representation.

add_operation(operation: BaseOperation) → None[source]

Add an operation to this cluster.

The operation is added to the end of the cluster’s operation list. For ordered clusters, this determines the execution sequence.

Parameters:: operation (BaseOperation) – The operation to add to this cluster.
Raises:: ValueError – If operation with the same name already exists in cluster.

allows_reordering() → bool[source]

Check if this cluster allows operation reordering.

Returns:: True if operations can be reordered, False if strict order required.
Return type:: bool

calculate_total_duration(rules: Any) → float[source]

Calculate total duration for all operations in this cluster.

Parameters:: rules (Any) – Duration calculation rules and parameters.
Returns:: Total duration in hours for all operations in cluster.
Return type:: float

classmethod from_definition(cluster_def: ClusterDefinition) → Cluster[source]

Create a Cluster instance from a ClusterDefinition.

Parameters:: cluster_def (ClusterDefinition) – The cluster definition to convert.
Returns:: New Cluster instance with parameters from definition.
Return type:: Cluster

get_all_operations() → list[BaseOperation][source]

Get all operations within this cluster.

Returns operations in their current order. For ordered clusters, this represents the required execution sequence.

Returns:: List of all operations in this cluster.
Return type:: List[BaseOperation]

get_entry_point() → tuple[float, float] | None[source]

Get geographic entry point for this cluster.

Returns the coordinates of the first operation in the cluster.

Returns:: (latitude, longitude) of entry point, or None if cluster is empty.
Return type:: Optional[Tuple[float, float]]

get_exit_point() → tuple[float, float] | None[source]

Get geographic exit point for this cluster.

Returns the coordinates of the last operation in the cluster.

Returns:: (latitude, longitude) of exit point, or None if cluster is empty.
Return type:: Optional[Tuple[float, float]]

get_operation(operation_name: str) → BaseOperation | None[source]

Get an operation from this cluster by name.

Parameters:: operation_name (str) – Name of the operation to find.
Returns:: The operation if found, None otherwise.
Return type:: Optional[BaseOperation]

get_operation_count() → int[source]

Get total number of operations in this cluster.

Returns:: Number of operations in this cluster.
Return type:: int

get_operation_names() → list[str][source]

Get names of all operations in this cluster.

Returns:: List of operation names in cluster order.
Return type:: List[str]

is_empty() → bool[source]

Check if this cluster contains any operations.

Returns:: True if cluster has no operations, False otherwise.
Return type:: bool

remove_operation(operation_name: str) → bool[source]

Remove an operation from this cluster by name.

Parameters:: operation_name (str) – Name of the operation to remove.
Returns:: True if operation was found and removed, False otherwise.
Return type:: bool

Bases: BaseOrganizationUnit

Port-to-port maritime leg container following nautical terminology.

A Leg represents a discrete maritime journey between two ports, containing all scientific operations and clusters executed during that voyage segment. This follows maritime tradition where a ‘leg’ always has departure and arrival ports, providing clear operational boundaries.

The Leg manages parameter inheritance (from parent Cruise), cluster boundaries, and port-to-port routing for realistic maritime scheduling.

name

Unique identifier for this leg.

Type:: str

description

Optional human-readable description of the leg’s purpose.

Type:: Optional[str]

departure_port

Required departure port for this maritime leg.

Type:: PointDefinition

arrival_port

Required arrival port for this maritime leg.

Type:: PointDefinition

strategy

Execution strategy for operations (default: SEQUENTIAL).

Type:: StrategyEnum

ordered

Whether operations should maintain their specified order (default: True).

Type:: bool

operations

List of standalone operations (e.g., single CTD, single Transit).

Type:: List[BaseOperation]

clusters

List of cluster boundaries for operation shuffling control.

Type:: List[Cluster]

first_activity

First waypoint/navigation marker for routing (not executed).

Type:: Optional[str]

last_activity

Last waypoint/navigation marker for routing (not executed).

Type:: Optional[str]

vessel_speed

Leg-specific vessel speed override (None uses cruise default).

Type:: Optional[float]

turnaround_time

Leg-specific turnaround time override in minutes (None uses cruise default).

Type:: Optional[float]

distance_between_stations

Leg-specific station spacing override (None uses cruise default).

Type:: Optional[float]

Examples

>>> # Arctic research leg with weather-flexible clusters
>>> leg = Leg(
...     name="Arctic_Survey",
...     departure_port=resolve_port_reference("port_tromsoe"),
...     arrival_port=resolve_port_reference("port_longyearbyen"),
...     vessel_speed=12.0,  # Faster speed for ice conditions
...     turnaround_time=45.0  # Extra time for Arctic operations
... )

Initialize a maritime Leg with port-to-port structure.

Parameters:

name (str) – Unique identifier for this leg.
departure_port (Union[str, PointDefinition, dict]) – Required departure port (can be global reference, PointDefinition, or dict).
arrival_port (Union[str, PointDefinition, dict]) – Required arrival port (can be global reference, PointDefinition, or dict).
description (Optional[str], optional) – Human-readable description of the leg’s purpose.
strategy (StrategyEnum, optional) – Execution strategy for operations (default: SEQUENTIAL).
ordered (bool, optional) – Whether operations should maintain their specified order (default: True).
first_activity (Optional[str], optional) – First waypoint for navigation (not executed).
last_activity (Optional[str], optional) – Last waypoint for navigation (not executed).

__repr__() → str[source]

String representation of the leg.

Returns:: String representation showing leg name, ports, and operation count.
Return type:: str

__str__() → str[source]

Human-readable string representation.

Returns:: Human-readable description of the leg with port-to-port information.
Return type:: str

add_cluster(cluster: Cluster) → None[source]

Add a cluster boundary to this leg for operation shuffling control.

Parameters:: cluster (Cluster) – The cluster boundary to add for operation reordering management.

add_operation(operation: BaseOperation) → None[source]

Add a single, standalone operation to this leg.

Parameters:: operation (BaseOperation) – The operation to add (e.g., a single CTD cast or section).

allows_reordering() → bool[source]

Check if this leg allows operation reordering.

A leg allows reordering if it’s not strictly ordered or if any of its clusters allow reordering.

Returns:: True if operations can be reordered within this leg, False if strict order required.
Return type:: bool

calculate_total_duration_legacy(rules: Any) → float[source]

Calculate total duration for all operations in this leg.

Includes port transit time, standalone operations, and cluster operations with proper boundary management.

Parameters:: rules (Any) – Duration calculation rules/parameters containing config.
Returns:: Total duration in minutes including all operations and port transits.
Return type:: float

classmethod from_definition(leg_def: LegDefinition) → Leg[source]

Create a Leg runtime instance from a LegDefinition.

This factory method converts a validated LegDefinition into a runtime Leg with proper port-to-port structure and default cluster creation.

Parameters:: leg_def (LegDefinition) – Validated leg definition from YAML configuration.
Returns:: New Leg runtime instance with resolved ports and clusters.
Return type:: Leg

get_all_clusters() → list[Cluster][source]

Get all clusters within this leg for boundary management.

Returns:: List of all cluster boundaries within this leg.
Return type:: List[Cluster]

get_all_operations() → list[BaseOperation][source]

Flatten all operations including those within cluster boundaries.

This provides a unified list of atomic operations for route optimization that respects the Leg’s port-to-port boundaries.

Returns:: Unified list containing both standalone operations and operations from within cluster boundaries.
Return type:: List[BaseOperation]

get_boundary_waypoints() → tuple[str | None, str | None][source]

Get the first and last waypoint boundaries for this leg.

Returns:: Tuple of (first_activity, last_activity) for boundary management.
Return type:: tuple[Optional[str], Optional[str]]

get_entry_point() → tuple[float, float][source]

Get the geographic entry point for this leg (departure port).

This provides a standardized interface regardless of internal field names.

Returns:: (latitude, longitude) of the leg’s entry point.
Return type:: tuple[float, float]

get_exit_point() → tuple[float, float][source]

Get the geographic exit point for this leg (arrival port).

This provides a standardized interface regardless of internal field names.

Returns:: (latitude, longitude) of the leg’s exit point.
Return type:: tuple[float, float]

get_operation_count() → int[source]

Get the total number of operations in this leg.

Returns:: Total count of operations including those within clusters.
Return type:: int

get_operational_entry_point(resolver=None) → tuple[float, float] | None[source]

Get the geographic entry point for operations within this leg.

Uses first_activity if available, otherwise first activity.

Parameters:: resolver (object, optional) – Operation resolver to look up waypoint coordinates.
Returns:: (latitude, longitude) of the operational entry point, or None if not resolvable.
Return type:: tuple[float, float] or None

get_operational_exit_point(resolver=None) → tuple[float, float] | None[source]

Get the geographic exit point for operations within this leg.

Uses last_activity if available, otherwise last activity.

Parameters:: resolver (object, optional) – Operation resolver to look up waypoint coordinates.
Returns:: (latitude, longitude) of the operational exit point, or None if not resolvable.
Return type:: tuple[float, float] or None

get_port_positions() → tuple[tuple[float, float], tuple[float, float]][source]

Get the geographic positions of departure and arrival ports.

Returns:: Tuple of ((dep_lat, dep_lon), (arr_lat, arr_lon)) port positions.
Return type:: tuple[tuple[float, float], tuple[float, float]]

get_station_spacing(default_spacing: float) → float[source]

Get station spacing for this leg (leg-specific override or cruise default).

Parameters:: default_spacing (float) – The default spacing from the parent cruise configuration.
Returns:: The effective station spacing for this leg.
Return type:: float

get_turnaround_time(default_turnaround: float) → float[source]

Get turnaround time for this leg (leg-specific override or cruise default).

Parameters:: default_turnaround (float) – The default turnaround time from the parent cruise configuration.
Returns:: The effective turnaround time for this leg in minutes.
Return type:: float

get_vessel_speed(default_speed: float) → float[source]

Get vessel speed for this leg (leg-specific override or cruise default).

Parameters:: default_speed (float) – The default speed from the parent cruise configuration.
Returns:: The effective vessel speed for this leg.
Return type:: float

is_same_port_leg() → bool[source]

Check if this leg departs and arrives at the same port.

Returns:: True if departure and arrival ports are the same, False otherwise.
Return type:: bool

exception cruiseplan.runtime.organizational.ReferenceError[source]

Bases: Exception

Exception raised when a referenced item is not found in the catalog.

This exception is raised during the reference resolution phase when string identifiers in the cruise configuration cannot be matched to their corresponding definitions in the station or transit registries.

cruiseplan.runtime.operations module

Runtime operation classes for cruise planning activities.

Provides the business logic classes that represent active cruise operations. These are the “core layer” that actually perform cruise planning work.

Relationship to schema/definitions.py: - schema/definitions.py: Pydantic models that validate YAML → Python data - This module: Runtime business objects that do the actual work

Example flow: YAML → PointDefinition (schema) → PointOperation (this module)

This module provides the base operation classes and specialized implementations for stations, moorings, areas, and transects.

class cruiseplan.runtime.operations.AreaOperation(name: str, boundary_polygon: list[GeoPoint], area_km2: float, duration: float | None = None, start_point: tuple[float, float] | None = None, end_point: tuple[float, float] | None = None, sampling_density: float = 1.0, comment: str | None = None, display_name: str | None = None, op_type: str = 'area', action: str | None = None)[source]

Bases: BaseOperation

Activities within defined polygonal regions.

Examples: grid surveys, area monitoring, search patterns. Operations that cover a defined geographic area rather than specific points or lines.

boundary_polygon

List of (latitude, longitude) tuples defining the area boundary.

Type:: List[Tuple[float, float]]

area_km2

Area of the polygon in square kilometers.

Type:: float

sampling_density

Sampling density factor for duration calculations.

Type:: float

duration

User-specified duration in minutes (required like moorings).

Type:: Optional[float]

start_point

Starting coordinates for area operation (latitude, longitude).

Type:: Tuple[float, float]

end_point

Ending coordinates for area operation (latitude, longitude).

Type:: Tuple[float, float]

__init__(name: str, boundary_polygon: list[GeoPoint], area_km2: float, duration: float | None = None, start_point: tuple[float, float] | None = None, end_point: tuple[float, float] | None = None, sampling_density: float = 1.0, comment: str | None = None, display_name: str | None = None, op_type: str = 'area', action: str | None = None)[source]

Initialize an area operation.

Parameters:

name (str) – Unique identifier for this operation.
boundary_polygon (List[Tuple[float, float]]) – List of (latitude, longitude) tuples defining the area boundary.
area_km2 (float) – Area of the polygon in square kilometers.
duration (Optional[float], optional) – User-specified duration in minutes (required for scheduling).
start_point (Optional[Tuple[float, float]], optional) – Starting coordinates (latitude, longitude). Defaults to first corner.
end_point (Optional[Tuple[float, float]], optional) – Ending coordinates (latitude, longitude). Defaults to last corner.
sampling_density (float, optional) – Sampling density factor for duration calculations (default: 1.0).
comment (str, optional) – Human-readable comment or description.

calculate_duration(rules: Any) → float[source]

Calculate duration using user-specified duration or fallback formula.

For area operations, duration must be specified by the user (like moorings) since area coverage patterns are highly variable.

Parameters:: rules (Any) – Duration calculation rules and parameters (unused for area operations).
Returns:: Duration in minutes.
Return type:: float
Raises:: ValueError – If duration is not specified by user.

classmethod from_pydantic(obj: AreaDefinition) → AreaOperation[source]

Factory to create an area operation from a validated Pydantic model.

Parameters:: obj (AreaDefinition) – Validated Pydantic area definition model.
Returns:: New AreaOperation instance.
Return type:: AreaOperation
Raises:: ValueError – If duration is not specified in the area definition.

get_entry_point() → tuple[float, float][source]

Get the geographic entry point for this area operation.

For area operations, this is the start point of the survey pattern.

Returns:: (latitude, longitude) of the area entry point.
Return type:: tuple[float, float]

get_exit_point() → tuple[float, float][source]

Get the geographic exit point for this area operation.

For area operations, this is the end point of the survey pattern.

Returns:: (latitude, longitude) of the area exit point.
Return type:: tuple[float, float]

class cruiseplan.runtime.operations.BaseOperation(name: str, comment: str | None = None, display_name: str | None = None)[source]

Bases: ABC

Abstract base class for all cruise operations.

This class defines the common interface that all cruise operations must implement, providing a foundation for different types of oceanographic activities.

name

Unique identifier for this operation.

Type:: str

comment

Optional human-readable comment or description.

Type:: Optional[str]

__init__(name: str, comment: str | None = None, display_name: str | None = None)[source]

Initialize a base operation.

Parameters:

name (str) – Unique identifier for this operation.
comment (Optional[str], optional) – Human-readable comment or description.
display_name (Optional[str], optional) – Human-readable display name for maps, CSV, and HTML output. Defaults to name if not provided.

abstractmethod calculate_duration(rules: Any) → float[source]

Calculate duration in minutes based on provided rules.

Parameters:: rules (Any) – Duration calculation rules and parameters.
Returns:: Duration in minutes.
Return type:: float

get_coordinates() → tuple[GeoPoint, GeoPoint][source]

Get entry and exit coordinates as GeoPoint objects.

Returns:: (entry_point, exit_point) as GeoPoint objects.
Return type:: Tuple[GeoPoint, GeoPoint]

abstractmethod get_entry_point() → tuple[float, float][source]

Get the geographic entry point for this operation.

For point operations (stations, moorings): same as operation location. For line operations (transits): start of the route.

Returns:: (latitude, longitude) of the operation’s entry point.
Return type:: tuple[float, float]

abstractmethod get_exit_point() → tuple[float, float][source]

Get the geographic exit point for this operation.

For point operations (stations, moorings): same as operation location. For line operations (transits): end of the route.

Returns:: (latitude, longitude) of the operation’s exit point.
Return type:: tuple[float, float]

get_label() → str[source]

Get human-readable label for this operation.

Returns:: Human-readable label, defaults to operation name.
Return type:: str

get_operation_type() → str[source]

Get operation type for timeline display.

Returns:: Operation type identifier (e.g., “Station”, “Transit”, “Area”).
Return type:: str

class cruiseplan.runtime.operations.LineOperation(name: str, route: list[GeoPoint], speed: float = 10.0, comment: str | None = None, display_name: str | None = None, op_type: str = 'line', action: str | None = None)[source]

Bases: BaseOperation

Continuous activity involving movement (Transit, Towyo).

Represents operations that involve traveling between points, such as vessel transits or towed instrument deployments.

route

List of geographic waypoints as (latitude, longitude) tuples.

Type:: List[tuple]

speed

Vessel speed in knots.

Type:: float

__init__(name: str, route: list[GeoPoint], speed: float = 10.0, comment: str | None = None, display_name: str | None = None, op_type: str = 'line', action: str | None = None)[source]

Initialize a line operation.

Parameters:

name (str) – Unique identifier for this operation.
route (List[tuple]) – List of geographic waypoints as (latitude, longitude) tuples.
speed (float, optional) – Vessel speed in knots (default: 10.0).
comment (str, optional) – Human-readable comment or description.

calculate_duration(rules: Any) → float[source]

Calculate duration for the line operation based on route distance and vessel speed.

Parameters:: rules (Any) – Duration calculation rules containing config with default_vessel_speed.
Returns:: Duration in minutes.
Return type:: float

classmethod from_pydantic(obj: LineDefinition, default_speed: float) → LineOperation[source]

Factory to create a line operation from a validated Pydantic model.

Parameters:

obj (LineDefinition) – Validated Pydantic transit definition model.
default_speed (float) – Default vessel speed to use if not specified in the model.

Returns:

New LineOperation instance.

Return type:

LineOperation

get_entry_point() → tuple[float, float][source]

Get the geographic entry point for this line operation.

For line operations, this is the start of the route.

Returns:: (latitude, longitude) of the route start point.
Return type:: tuple[float, float]

get_exit_point() → tuple[float, float][source]

Get the geographic exit point for this line operation.

For line operations, this is the end of the route.

Returns:: (latitude, longitude) of the route end point.
Return type:: tuple[float, float]

get_operation_distance_nm() → float[source]

Calculate the total route distance for this line operation.

Returns:: Total route distance in nautical miles.
Return type:: float

class cruiseplan.runtime.operations.PointOperation(name: str, position: GeoPoint, operation_depth: float | None = None, water_depth: float = 0.0, duration: float = 0.0, comment: str | None = None, op_type: str = 'station', action: str | None = None, display_name: str | None = None)[source]

Bases: BaseOperation

Atomic activity at a fixed location.

Handles both Stations (CTD casts) and Moorings (deploy/recover operations). Represents the most basic unit of work in a cruise plan.

position

Geographic position as (latitude, longitude).

Type:: tuple

depth

Operation depth in meters.

Type:: float

manual_duration

User-specified duration override in minutes.

Type:: float

op_type

Type of operation (‘station’ or ‘mooring’).

Type:: str

action

Specific action for moorings (deploy/recover).

Type:: str

__init__(name: str, position: GeoPoint, operation_depth: float | None = None, water_depth: float = 0.0, duration: float = 0.0, comment: str | None = None, op_type: str = 'station', action: str | None = None, display_name: str | None = None)[source]

Initialize a point operation.

Parameters:

name (str) – Unique identifier for this operation.
position (tuple) – Geographic position as (latitude, longitude).
depth (float, optional) – Operation depth in meters (default: 0.0).
duration (float, optional) – Manual duration override in minutes (default: 0.0).
comment (str, optional) – Human-readable comment or description.
op_type (str, optional) – Type of operation (‘station’ or ‘mooring’, default: ‘station’).
action (str, optional) – Specific action for moorings (deploy/recover).

calculate_duration(rules: Any) → float[source]

Calculate duration based on operation type and rules.

Uses manual duration if specified, otherwise calculates based on operation type (CTD time for stations, default duration for moorings).

Parameters:: rules (Any) – Duration calculation rules containing config.
Returns:: Duration in minutes.
Return type:: float

classmethod from_port(obj: PointDefinition) → PointOperation[source]

Factory to create a PointOperation from a PointDefinition.

Parameters:: obj (PointDefinition) – Port definition model.
Returns:: New PointOperation instance representing a port.
Return type:: PointOperation

classmethod from_pydantic(obj: PointDefinition) → PointOperation[source]

Factory to create a logical operation from a validated Pydantic model.

Handles the internal ‘position’ normalization done by FlexibleLocationModel.

Parameters:: obj (PointDefinition) – Validated Pydantic station definition model.
Returns:: New PointOperation instance.
Return type:: PointOperation

get_depth() → float[source]

Get the appropriate depth for this operation.

Returns operation_depth if available, otherwise water_depth, otherwise 0.0.

Returns:: Depth in meters.
Return type:: float

get_entry_point() → tuple[float, float][source]

Get the geographic entry point for this point operation.

For point operations, entry and exit are the same location.

Returns:: (latitude, longitude) of the operation location.
Return type:: tuple[float, float]

get_exit_point() → tuple[float, float][source]

Get the geographic exit point for this point operation.

For point operations, entry and exit are the same location.

Returns:: (latitude, longitude) of the operation location.
Return type:: tuple[float, float]

get_operation_type() → str[source]

Get operation type for timeline display.

Returns appropriate display type based on the op_type and action attributes. For ports, returns “Port_Departure” or “Port_Arrival” based on action.

Returns:: Operation type identifier (“Station”, “Mooring”, “Port_Departure”, “Port_Arrival”, etc.).
Return type:: str

cruiseplan.runtime.enrichment module

Cruise configuration enrichment functions.

This module contains functions for enriching cruise configurations by adding missing data like depths, coordinate displays, and expanding sections into individual stations. These functions operate on CruiseInstance objects.

cruiseplan.runtime.enrichment.add_coordinate_displays(cruise_instance: CruiseInstance, coord_format: str = 'ddm') → int[source]

Add human-readable coordinate display fields for final YAML output.

This function adds formatted coordinate fields (latitude_display, longitude_display) to all geographic entities (stations, ports, etc.) for better human readability in generated YAML files.

Parameters:

cruise_instance (CruiseInstance) – The cruise instance to modify
coord_format (str, optional) – Coordinate format (“ddm” for degrees/decimal minutes, “dms” for degrees/minutes/seconds) Default is “ddm”.

Returns:

Number of entities that had coordinate displays added

Return type:

int

cruiseplan.runtime.enrichment.add_station_defaults(cruise_instance: CruiseInstance) → int[source]

Add missing defaults to station definitions.

This function adds default values for station fields based on operation type and scientific best practices. Currently focuses on mooring duration defaults.

Parameters:: cruise_instance (CruiseInstance) – The cruise instance to modify
Returns:: Number of stations that had defaults added
Return type:: int

cruiseplan.runtime.enrichment.enrich_depths(cruise_instance: CruiseInstance, bathymetry_source: str = 'etopo2022', bathymetry_dir: str = 'data') → set[str][source]

Add missing depth values to stations using bathymetry data.

This function queries bathymetry data to fill in missing water depth values for stations that don’t have them specified.

Parameters:

cruise_instance (CruiseInstance) – The cruise instance to modify
bathymetry_source (str, optional) – Bathymetry dataset to use (default: “etopo2022”)
bathymetry_dir (str, optional) – Directory containing bathymetry data (default: “data”)

Returns:

Set of station names that had depths added

Return type:

set[str]

cruiseplan.runtime.enrichment.expand_ports(cruise_instance: CruiseInstance) → dict[str, int][source]

Expand global port references into full PortDefinition objects.

This function processes port references in the cruise configuration and expands them into full port definitions with coordinates and metadata.

Note: In v0.3.4+ architecture, ports are automatically resolved during CruiseInstance creation, so this function may be largely redundant.

Parameters:: cruise_instance (CruiseInstance) – The cruise instance to modify
Returns:: Dictionary with expansion summary: - ports_expanded: Number of port references expanded - ports_added: Number of port definitions added
Return type:: dict[str, int]

cruiseplan.runtime.enrichment.expand_sections(cruise_instance: CruiseInstance, default_depth: float = -9999.0) → dict[str, int][source]

Expand CTD sections into individual station definitions.

This method finds CTD sections in lines catalog and expands them into individual stations, adding them to the point_registry. This is structural enrichment that modifies the cruise configuration.

Parameters:

cruise_instance (CruiseInstance) – The cruise instance to modify
default_depth (float, optional) – Default depth value for expanded stations. Default is -9999.0.

Returns:

Dictionary with expansion summary: - sections_expanded: Number of sections expanded - stations_from_expansion: Number of stations created

Return type:

dict[str, int]

cruiseplan.runtime.serialization module

Cruise configuration serialization functions.

This module contains high-level business logic for serializing CruiseInstance objects back to dictionary and YAML formats with proper field ordering and comment preservation.

I/O Module Architecture: - cruiseplan.utils.io: File system validation, path handling, directory creation - cruiseplan.schema.yaml_io: YAML file format reading/writing with comment preservation (used by this module) - cruiseplan.core.serialization (this module): High-level CruiseInstance object serialization to YAML - cruiseplan.output.*_generator: Specialized output format generators (HTML, LaTeX, CSV, etc.)

Dependencies: Uses cruiseplan.schema.yaml_io for YAML operations and cruiseplan.utils.io for file handling.

See Also: - For file system operations: cruiseplan.utils.io - For YAML file operations: cruiseplan.schema.yaml_io - For generating specific output formats: cruiseplan.output.html_generator, cruiseplan.output.latex_generator, etc. - For the enrichment process that uses these functions: cruiseplan.core.enrichment

cruiseplan.runtime.serialization.deserialize_inline_definition(definition_dict: dict) → PointDefinition | LineDefinition | AreaDefinition[source]

Convert an inline dictionary definition to the appropriate definition object.

Determines the type of definition based on the presence of key fields and creates the corresponding Pydantic object.

Parameters:: definition_dict (dict) – Dictionary containing the inline definition fields.
Returns:: The appropriate definition object created from the dictionary.
Return type:: Union[PointDefinition, LineDefinition, AreaDefinition]
Raises:: ValueError – If the definition type cannot be determined or validation fails.

cruiseplan.runtime.serialization.serialize_area_definition(area: AreaDefinition) → dict[str, Any][source]

Serialize an AreaDefinition to dictionary format.

Parameters:: area (AreaDefinition) – The area definition to serialize
Returns:: Serialized area definition dictionary
Return type:: dict[str, Any]

cruiseplan.runtime.serialization.serialize_cluster_definition(cluster: ClusterDefinition) → dict[str, Any][source]

Serialize a ClusterDefinition to dictionary format.

Parameters:: cluster (ClusterDefinition) – The cluster definition to serialize
Returns:: Serialized cluster definition dictionary
Return type:: dict[str, Any]

Convert a Pydantic definition object to a dictionary with field filtering.

This function extracts only the allowed fields from the object, filtering out internal fields and maintaining canonical field ordering from YAML_FIELD_ORDER.

Parameters:

obj (Union[PointDefinition, LineDefinition, AreaDefinition, ClusterDefinition, LegDefinition]) – The Pydantic object to serialize
allowed_fields (Union[list[str], set[str]]) – Collection of field names that should be included in the output. Field ordering is determined by YAML_FIELD_ORDER, not by this parameter.

Returns:

Dictionary containing only the allowed fields with their values in canonical order

Return type:

dict[str, Any]

cruiseplan.runtime.serialization.serialize_leg_definition(leg: LegDefinition) → dict[str, Any][source]

Serialize a LegDefinition to dictionary format.

Parameters:: leg (LegDefinition) – The leg definition to serialize
Returns:: Serialized leg definition dictionary
Return type:: dict[str, Any]

cruiseplan.runtime.serialization.serialize_line_definition(line: LineDefinition) → dict[str, Any][source]

Serialize a LineDefinition to dictionary format.

Parameters:: line (LineDefinition) – The line definition to serialize
Returns:: Serialized line definition dictionary
Return type:: dict[str, Any]

cruiseplan.runtime.serialization.serialize_point_definition(point: PointDefinition) → dict[str, Any][source]

Serialize a PointDefinition to dictionary format.

Parameters:: point (PointDefinition) – The point definition to serialize
Returns:: Serialized point definition dictionary
Return type:: dict[str, Any]

cruiseplan.runtime.serialization.to_commented_dict(cruise_instance: CruiseInstance) → dict[str, Any][source]

Export CruiseInstance configuration to a structured dictionary with comment preservation.

This method provides the foundation for YAML output with canonical field ordering and comment preservation capabilities. Returns a dictionary that can be processed by ruamel.yaml for structured output with comments.

Parameters:: cruise_instance (CruiseInstance) – The cruise instance to serialize
Returns:: Dictionary with canonical field ordering suitable for YAML export with comment preservation.
Return type:: Dict[str, Any]

Notes

The output dictionary follows canonical ordering: 1. Cruise Metadata (cruise_name, description, start_date, start_time) 2. Vessel Parameters (default_vessel_speed, turnaround_time, etc.) 3. Calculation Settings (calculate_*, day_start_hour, etc.) 4. Catalog Definitions (points, lines, areas, ports) 5. Schedule Organization (legs)

Comment preservation is handled at the YAML layer using ruamel.yaml with end-of-line and section header comment support.

cruiseplan.runtime.serialization.to_yaml(cruise_instance: CruiseInstance, output_file: str | Path | None = None, backup: bool = True, add_comments: bool = True) → str | None[source]

Export CruiseInstance configuration to YAML format with comment preservation.

Parameters:

cruise_instance (CruiseInstance) – The cruise instance to serialize
output_file (Optional[Union[str, Path]], optional) – Path to write YAML file. If None, returns YAML string.
backup (bool, optional) – Whether to create backup of existing file (default: True)
add_comments (bool, optional) – Whether to add descriptive comments to YAML (default: True)

Returns:

YAML string if output_file is None, otherwise None

Return type:

Optional[str]

Examples

>>> # Save to file
>>> cruise.to_yaml("enhanced_cruise.yaml")
>>> # Get YAML string
>>> yaml_str = cruise.to_yaml()

cruiseplan.runtime.validation module

Cruise Configuration Validation Functions.

This module provides comprehensive validation for cruise configurations, organized into distinct validation categories:

Duplicate Detection: Find naming conflicts and identical entries with proper scoping - check_duplicate_names() - Enforce uniqueness scopes: global catalog (points/lines/areas), legs, and per-leg clusters - check_complete_duplicates() - Find true duplicates (same name + same attributes) - likely copy-paste errors
Scientific Data Validation: Verify oceanographic accuracy - validate_depth_accuracy() - Compare stated depths with bathymetry data
Configuration Completeness: Check for missing/incomplete configuration - check_unexpanded_ctd_sections() - Find CTD sections needing expansion - check_cruise_metadata() - Verify cruise metadata completeness
Pydantic Warning Processing: Convert technical validation errors to user-friendly messages - format_validation_warnings() - Main entry point for processing Pydantic warnings - Helper functions for text matching and message cleanup

All validation functions operate on CruiseInstance objects and return structured error/warning information suitable for display to users.

cruiseplan.runtime.validation.check_complete_duplicates(cruise_instance: CruiseInstance) → tuple[list[str], list[str]][source]

Check for completely identical entries across all entity types.

This catches true duplicates where everything is identical - likely copy-paste errors or accidental duplicates. This is a subset of entities that would also be caught by check_duplicate_names(), but indicates a more serious duplication issue.

Parameters:: cruise_instance (CruiseInstance) – Loaded cruise configuration object.
Returns:: Tuple of (errors, warnings) for complete duplicate detection.
Return type:: Tuple[List[str], List[str]]

cruiseplan.runtime.validation.check_cruise_metadata(cruise_instance: CruiseInstance) → list[str][source]

Check cruise metadata for placeholder values and default coordinates.

Uses specific placeholder patterns defined in cruiseplan.config.values: - UPDATE- prefix for placeholder values - port_update_ prefix for port placeholders - Default cruise name placeholder

Parameters:: cruise_instance (CruiseInstance) – Cruise instance to check.
Returns:: List of warnings about metadata issues.
Return type:: List[str]

cruiseplan.runtime.validation.check_duplicate_names(cruise_instance: CruiseInstance) → tuple[list[str], list[str]][source]

Check for duplicate names across different configuration scopes.

Enforces three uniqueness scopes: 1. Global catalog: points, lines, areas must all have unique names (cross-type) 2. Leg scope: legs must have unique names within legs 3. Cluster scope: clusters must have unique names within clusters

Parameters:: cruise_instance (CruiseInstance) – Loaded cruise configuration object.
Returns:: Tuple of (errors, warnings) for duplicate detection.
Return type:: Tuple[List[str], List[str]]

cruiseplan.runtime.validation.check_unexpanded_ctd_sections(cruise_instance: CruiseInstance) → list[str][source]

Check for CTD sections that haven’t been expanded yet.

Parameters:: cruise_instance (CruiseInstance) – Cruise instance to check.
Returns:: List of warnings about unexpanded CTD sections.
Return type:: List[str]

cruiseplan.runtime.validation.clean_warning_message(warning_msg: str) → str[source]

Clean up Pydantic validation warnings for user display.

Removes technical Pydantic-specific text and formats warnings in a more user-friendly way.

Parameters:: warning_msg (str) – Raw Pydantic validation warning message
Returns:: Cleaned warning message suitable for display to users
Return type:: str

cruiseplan.runtime.validation.format_validation_warnings(captured_warnings: list[str], cruise_instance: CruiseInstance) → list[str][source]

Format captured Pydantic warnings into user-friendly grouped messages.

This is the main entry point for converting technical Pydantic validation errors into readable warnings grouped by entity type (Points/Lines/Areas).

Parameters:

captured_warnings (List[str]) – List of captured warning messages from Pydantic validators.
cruise_instance (CruiseInstance) – Cruise instance to map warnings to specific entities.

Returns:

Formatted warning messages grouped by type and sorted alphabetically.

Return type:

List[str]

cruiseplan.runtime.validation.validate_depth_accuracy(cruise_instance: CruiseInstance, bathymetry_manager: BathymetryManager, tolerance: float) → tuple[int, list[str]][source]

Compare station water depths with bathymetry data.

Validates that stated water depths are reasonably close to bathymetric depths.

Parameters:

cruise_instance (CruiseInstance) – Loaded cruise configuration object.
bathymetry_manager (BathymetryManager) – Bathymetry data manager instance.
tolerance (float) – Tolerance percentage for depth differences.

Returns:

Tuple of (stations_checked, warning_messages) where: - stations_checked: Number of stations with depth data - warning_messages: List of depth discrepancy warnings

Return type:

Tuple[int, List[str]]

cruiseplan.runtime.validation.warning_relates_to_entity(warning_msg: str, entity: PointDefinition | LineDefinition | AreaDefinition) → bool[source]

Check if a Pydantic warning message relates to a specific entity.

Uses text pattern matching to determine which station/line/area a validation error belongs to by checking if entity field values appear in the warning text.

Parameters:

warning_msg (str) – Raw Pydantic validation warning message
entity (Union[PointDefinition, LineDefinition, AreaDefinition]) – Entity to check against the warning message

Returns:

True if the warning appears to relate to this entity

Return type:

bool