From 775edadd7d70f3693d42113c937fc8e6326bdff9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Grunde=20L=C3=B8voll?= Date: Mon, 10 Jun 2024 10:19:30 +0200 Subject: [PATCH] started refactoring --- .vscode/settings.json | 4 +- src/maritime_schema/types/__init__.py | 4 +- src/maritime_schema/types/caga.py | 278 ++--------------------- src/maritime_schema/types/environment.py | 10 +- src/maritime_schema/types/meta.py | 13 ++ src/maritime_schema/types/physics.py | 22 ++ src/maritime_schema/types/route.py | 140 ++++++++++++ src/maritime_schema/types/ship.py | 71 ++++++ src/maritime_schema/types/state.py | 24 ++ 9 files changed, 305 insertions(+), 261 deletions(-) diff --git a/.vscode/settings.json b/.vscode/settings.json index 77f7a94..d208e36 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -1,6 +1,6 @@ { "python.languageServer": "Pylance", - "editor.formatOnSave": true, + "editor.formatOnSave": false, "notebook.formatOnSave.enabled": true, "notebook.codeActionsOnSave": { "notebook.source.fixAll": true, @@ -29,4 +29,4 @@ "python.terminal.executeInFileDir": true, "python.terminal.activateEnvironment": true, "python.terminal.activateEnvInCurrentTerminal": false, -} +} \ No newline at end of file diff --git a/src/maritime_schema/types/__init__.py b/src/maritime_schema/types/__init__.py index 4c73b6f..b256c8b 100644 --- a/src/maritime_schema/types/__init__.py +++ b/src/maritime_schema/types/__init__.py @@ -3,6 +3,6 @@ """ # from .state import Position, StateVector +from .meta import MTSBaseModel - -__all__ = [] \ No newline at end of file +__all__ = [MTSBaseModel, ] diff --git a/src/maritime_schema/types/caga.py b/src/maritime_schema/types/caga.py index 77fd9b1..2694e6c 100644 --- a/src/maritime_schema/types/caga.py +++ b/src/maritime_schema/types/caga.py @@ -14,259 +14,25 @@ # change from maritime_schema.types.caga_examples import ( - create_caga_config_example, - create_caga_data_example, - create_caga_event_example, - create_caga_time_frame_example, - create_detected_ship_example, - create_environment_example, - create_initial_example, - create_position_example, - create_predicted_point_example, - create_ship_example, - create_ship_static_example, - create_simulation_data_example, - create_simulation_timeframe_example, - create_software_config_example, - create_waypoint_example, -) + create_caga_config_example, create_caga_data_example, + create_caga_event_example, create_caga_time_frame_example, + create_detected_ship_example, create_environment_example, + create_initial_example, create_position_example, + create_predicted_point_example, create_ship_example, + create_ship_static_example, create_simulation_data_example, + create_simulation_timeframe_example, create_software_config_example, + create_waypoint_example) from maritime_schema.utils.strings import to_camel +from . import MTSBaseModel + __ALL__ = ["TrafficSituation", "OutputSchema", "publish_schema"] logger = logging.getLogger(__name__) -class BaseModelConfig(BaseModel): - """Enables the alias_generator for all cases.""" - - model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True) - - - - - - - - - -class Initial(BaseModelConfig): - position: Position = Field( - None, - description="Initial longitude and latitude of the ship.", - examples=[create_position_example()], - ) - sog: float = Field( - None, - ge=0, - description="Initial ship speed over ground in knots", - examples=[10.0], - ) - cog: float = Field( - None, - ge=0, - le=360, - description="Initial ship course over ground in degrees", - examples=[45.0], - ) - heading: float = Field( - None, - ge=0, - le=360, - description="Initial ship heading in degrees", - examples=[45.2], - ) - nav_status: Optional[AISNavStatus] = Field(None, description="AIS Navigational Status") - - -class DataPoint(BaseModelConfig): - value: Optional[float] = Field( - None, - description="the value of the data at the current timestep", - examples=[12.3], - ) - m_before_leg_change: Optional[float] = Field( - None, - description="meters before the waypoint to start interpolating to the new value", - examples=[10], - ) - m_after_leg_change: Optional[float] = Field( - None, - description="meters after the waypoint to finish interpolating to the new value", - examples=[10], - ) - interp_method: Optional[Union[InterpolationMethod, str]] = Field(None, description="Method used for interpolation") - - -class Data(BaseModelConfig): - model_config = ConfigDict( - extra="allow", - json_schema_extra={ - "additionalProperties": { - "type": "object", - "properties": { - "value": {"type": "number"}, - "mBeforeLegChange": {"type": "number"}, - "mAfterLegChange": {"type": "number"}, - "interpMethod": {"type": "string"}, - }, - "required": [], - "description": "The 'data' field can include additional properties. All additional properties should be DataPoint objects.", - }, - "sog": { - "type": "object", - "properties": { - "value": {"type": "number"}, - "mBeforeLegChange": {"type": "number"}, - "mAfterLegChange": {"type": "number"}, - "interpMethod": {"type": "string"}, - }, - "required": [], - "description": "Speed data point", - "examples": [ - { - "value": 12.3, - "mBeforeLegChange": 100, - "mAfterLegChange": 100, - "interpMethod": "linear", - } - ], - }, - "heading": { - "type": "object", - "properties": { - "value": {"type": "number"}, - "mBeforeLegChange": {"type": "number"}, - "mAfterLegChange": {"type": "number"}, - "interpMethod": {"type": "string"}, - }, - "required": [], - "description": "Heading data point", - "examples": [ - { - "value": 180, - "mBeforeLegChange": 100, - "mAfterLegChange": 100, - "interpMethod": "linear", - } - ], - }, - }, - ) - - -class Waypoint(BaseModelConfig): - position: Position = Field( - description="A geographical coordinate", - examples=[Position(latitude=51.2123, longitude=11.2313)], - ) - turn_radius: Optional[float] = Field( - None, - description="Orthodrome turn radius as defined in RTZ format", - examples=[200], - ) - data: Optional[Data] = Field( - None, - description="A `Data` object that includes `speed`, `course`, and `heading` data points", - ) - - -class Ship(BaseModelConfig): - static: ShipStatic = Field( - None, - description="Static ship information which does not change during a scenario.", - examples=[create_ship_static_example()], - ) - initial: Optional[Initial] = Field(None, examples=[create_initial_example()]) - waypoints: Optional[List[Waypoint]] = Field( - None, - description="An array of `Waypoint` objects. Each waypoint object must have a `position` property.
If no turn radius is provided, it will be assumed to be `0`.
Additional data can be added to each waypoint leg. This allows varying parameters on a per-leg basis, such as speed and heading, or navigational status ", - examples=[create_waypoint_example()], - ) - - def __init__(self, **data: Any): - super().__init__(**data) - if not self.waypoints: - self.waypoints = self.generate_waypoints() - - def generate_waypoints(self) -> Union[List[Waypoint], None]: - """Generate waypoints if they don't exist.""" - waypoints = [] - - if self.initial: - # Create waypoints from initial position - geod = Geod(ellps="WGS84") - lon, lat, _ = geod.fwd( - self.initial.position.longitude, - self.initial.position.latitude, - self.initial.cog, - 10000, # distance in meters - ) - position1 = Position(latitude=lat, longitude=lon) - waypoint1 = Waypoint(position=position1) - - position0 = Position( - latitude=self.initial.position.latitude, - longitude=self.initial.position.longitude, - ) - waypoint0 = Waypoint(position=position0, data=None) - - waypoints = [waypoint0, waypoint1] - return waypoints - - else: - return None - - -class OwnShip(Ship): - pass - - -class TargetShip(Ship): - pass - - -class TrafficSituation(BaseModelConfig): - title: str = Field( - None, - description="The title of the traffic situation", - examples=["overtaking_18"], - ) - description: Optional[str] = Field( - None, - description="A description of the traffic situation", - examples=["Crossing situation with 3 target vessels in the Oslofjord"], - ) - start_time: Optional[datetime] = Field( - None, - description="Starting time of the situation in `ISO 8601` format `YYYY-MM-DDThh:mm:ssZ`", - examples=[datetime.now()], - ) - own_ship: Union[OwnShip, Ship] = Field( - title="Own Ship data", - description="Own Ship data", - examples=[create_ship_example()], - ) - target_ships: List[Union[TargetShip, Ship]] = Field( - None, - title="Target Ship data", - description="Target Ship data", - examples=[[create_ship_example()]], - ) - environment: Optional[Environment] = Field( - None, - description="environmental parameters", - examples=[create_environment_example()], - ) - - model_config = ConfigDict( - extra="allow", - title="Test Input Schema", - json_schema_extra={"additionalProperties": True, "omit_default": True}, - ) - -class SoftwareConfig(BaseModelConfig): +class SoftwareConfig(MTSBaseModel): name: str = Field(..., description="The name of the system", examples=["AutoNavigation-System 1"]) vendor: str = Field(..., description="The name of the system vendor", examples=["CompanyABC"]) version: str = Field(..., description="The software version", examples=["1.2.3"]) @@ -274,7 +40,7 @@ class SoftwareConfig(BaseModelConfig): model_config = ConfigDict(json_schema_extra={"additionalProperties": True}) -class CagaConfiguration(BaseModelConfig): +class CagaConfiguration(MTSBaseModel): name: str = Field(..., description="The name of the system", examples=["AutoNavigation-System 1"]) vendor: str = Field(..., description="The name of the system vendor", examples=["CompanyABC"]) version: str = Field(..., description="The software version", examples=["1.2.3"]) @@ -317,7 +83,7 @@ class EncounterType(str, Enum): NO_RISK = "No Risk" -class PredictedPoint(BaseModelConfig): +class PredictedPoint(MTSBaseModel): time: Union[datetime, int] = Field( ..., description="Date and Time of the predicted value `ISO 8601` format `YYYY-MM-DDThh:mm:ssZ`", @@ -330,7 +96,7 @@ class PredictedPoint(BaseModelConfig): ) -class DetectedShip(BaseModelConfig): +class DetectedShip(MTSBaseModel): id: UUID = Field(..., description="Unique Identifier", examples=[uuid4()]) position: Position = Field( @@ -389,7 +155,7 @@ class DetectedShip(BaseModelConfig): model_config = ConfigDict(json_schema_extra={"additionalProperties": True}) -class SimulatedShip(BaseModelConfig): +class SimulatedShip(MTSBaseModel): id: UUID = Field(..., description="Unique Identifier", examples=[uuid4()]) position: Position = Field( @@ -426,7 +192,7 @@ class SimulatedShip(BaseModelConfig): model_config = ConfigDict(json_schema_extra={"additionalProperties": True}) -class CagaTimeStep(BaseModelConfig): +class CagaTimeStep(MTSBaseModel): time: Union[datetime, int] = Field( ..., description="Date and Time of the predicted value `ISO 8601` format `YYYY-MM-DDThh:mm:ssZ`", @@ -443,7 +209,7 @@ class CagaTimeStep(BaseModelConfig): ) -class CagaEvent(BaseModelConfig): +class CagaEvent(MTSBaseModel): time: Union[datetime, int] = Field(..., description="Date and Time of the event", examples=[datetime.now()]) route: List[Waypoint] = Field( None, @@ -462,13 +228,13 @@ class CagaEvent(BaseModelConfig): model_config = ConfigDict(json_schema_extra={"additionalProperties": True}) -class SimulatorEvent(BaseModelConfig): +class SimulatorEvent(MTSBaseModel): time: Union[datetime, int] = Field(..., description="Date and Time of the event", examples=[datetime.now()]) model_config = ConfigDict(json_schema_extra={"additionalProperties": True}) -class CagaData(BaseModelConfig): +class CagaData(MTSBaseModel): configuration: Optional[CagaConfiguration] = Field( None, description="System Configuration", examples=[create_caga_config_example()] ) @@ -484,7 +250,7 @@ class CagaData(BaseModelConfig): ) -class SimulationTimeFrame(BaseModelConfig): +class SimulationTimeFrame(MTSBaseModel): time: Union[datetime, int] = Field( ..., description="Date and Time of the predicted value `ISO 8601` format `YYYY-MM-DDThh:mm:ssZ`", @@ -494,7 +260,7 @@ class SimulationTimeFrame(BaseModelConfig): target_ships: List[SimulatedShip] -class SimulationData(BaseModelConfig): +class SimulationData(MTSBaseModel): configuration: Optional[SoftwareConfig] = Field( None, description="Simulator software configuration", @@ -513,7 +279,7 @@ class SimulationData(BaseModelConfig): event_data: Optional[List[SimulatorEvent]] = Field(None, description="Event data from the simulator") -class OutputSchema(BaseModelConfig): +class OutputSchema(MTSBaseModel): creation_time: datetime = Field( ..., description="Date and Time that this file was created, in `ISO 8601` format `YYYY-MM-DDThh:mm:ssZ`. This should be the simulation end time.", diff --git a/src/maritime_schema/types/environment.py b/src/maritime_schema/types/environment.py index 2985b09..62f649d 100644 --- a/src/maritime_schema/types/environment.py +++ b/src/maritime_schema/types/environment.py @@ -1,3 +1,11 @@ +import logging +from enum import Enum + +from .meta import MTSBaseModel + +logger = logging.getLogger(__name__) + + class WaveSpectra(Enum): JONSWAP = "JONSWAP" PiersonMoskowitz = "Pierson-Moskowitz" @@ -20,7 +28,7 @@ class PrecipitationType(Enum): Hail = "Hail" -class Environment(BaseModelConfig): +class Environment(MTSBaseModel): air_temperature: float = Field(None, description="The air temperature in degrees Celsius", examples=[20.0]) water_remperature: float = Field(None, description="The water temperature in degrees Celsius", examples=[15.0]) precipitation: PrecipitationType = Field( diff --git a/src/maritime_schema/types/meta.py b/src/maritime_schema/types/meta.py index e69de29..de8a430 100644 --- a/src/maritime_schema/types/meta.py +++ b/src/maritime_schema/types/meta.py @@ -0,0 +1,13 @@ +from typing import Any, Dict, List, Optional, Union +from uuid import UUID, uuid4 + +from pydantic import BaseModel, ConfigDict, Field + +from maritime_schema.utils import to_camel + + +class MTSBaseModel(BaseModel): + """Enables the alias_generator for all cases.""" + id: Optional[UUID] = Field(description="Unique identifier", + default=None) + model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True) diff --git a/src/maritime_schema/types/physics.py b/src/maritime_schema/types/physics.py index e69de29..2e26bdd 100644 --- a/src/maritime_schema/types/physics.py +++ b/src/maritime_schema/types/physics.py @@ -0,0 +1,22 @@ +import logging +from venv import logger +from pydantic import Field +from . import MTSBaseModel + + +logger = logging.getLogger(__name__) + +class Mass(MTSBaseModel): + """Mass of an object.""" + value: float + unit: str = "kg" + description: str = "Mass of an object in kilograms." + + +class Shape(MTSBaseModel): + """Shape of an object.""" + length: float = Field(None, description="Length of the object.", gt=0) + width: float = Field(None, description="Width of the object.", gt=0) + height: float = Field(None, description="Height of the object.", gt=0) + draft: Optional[float] = Field(None, description="Draft of the object.", gt=0) + description: str = "Shape of an object." \ No newline at end of file diff --git a/src/maritime_schema/types/route.py b/src/maritime_schema/types/route.py index e69de29..fb75a6b 100644 --- a/src/maritime_schema/types/route.py +++ b/src/maritime_schema/types/route.py @@ -0,0 +1,140 @@ +from pydantic import BaseModel, ConfigDict +from pydantic.fields import Field # , FieldInfo + +from .meta import MTSBaseModel + +class DataPoint(MTSBaseModel): + """ + Data point to associate with a Waypoint + """ + value: Optional[float] = Field( + None, + description="the value of the data at the current waypoint", + examples=[12.3], + ) + m_before_leg_change: Optional[float] = Field( + None, + description="meters before the waypoint to start interpolating to the new value", + examples=[10], + ) + m_after_leg_change: Optional[float] = Field( + None, + description="meters after the waypoint to finish interpolating to the new value", + examples=[10], + ) + interp_method: Optional[Union[InterpolationMethod, str]] = Field(None, description="Method used for interpolation") + + + +class Data(MTSBaseModel): + # FIXME: I don't get the purpose of this class... must ask @Minos + model_config = ConfigDict( + extra="allow", + json_schema_extra={ + "additionalProperties": { + "type": "object", + "properties": { + "value": {"type": "number"}, + "mBeforeLegChange": {"type": "number"}, + "mAfterLegChange": {"type": "number"}, + "interpMethod": {"type": "string"}, + }, + "required": [], + "description": "The 'data' field can include additional properties. All additional properties should be DataPoint objects.", + }, + "sog": { + "type": "object", + "properties": { + "value": {"type": "number"}, + "mBeforeLegChange": {"type": "number"}, + "mAfterLegChange": {"type": "number"}, + "interpMethod": {"type": "string"}, + }, + "required": [], + "description": "Speed data point", + "examples": [ + { + "value": 12.3, + "mBeforeLegChange": 100, + "mAfterLegChange": 100, + "interpMethod": "linear", + } + ], + }, + "heading": { + "type": "object", + "properties": { + "value": {"type": "number"}, + "mBeforeLegChange": {"type": "number"}, + "mAfterLegChange": {"type": "number"}, + "interpMethod": {"type": "string"}, + }, + "required": [], + "description": "Heading data point", + "examples": [ + { + "value": 180, + "mBeforeLegChange": 100, + "mAfterLegChange": 100, + "interpMethod": "linear", + } + ], + }, + }, + ) + +class Waypoint(BaseModelConfig): + position: Position = Field( + description="A geographical coordinate", + examples=[Position(latitude=51.2123, longitude=11.2313)], + ) + turn_radius: Optional[float] = Field( + None, + description="Orthodrome turn radius as defined in RTZ format", + examples=[200], + ) + data: Optional[Data] = Field( + None, + description="A `Data` object that includes `speed`, `course`, and `heading` data points", + ) + + +class TrafficSituation(MTSBaseModel): + title: str = Field( + None, + description="The title of the traffic situation", + examples=["overtaking_18"], + ) + description: Optional[str] = Field( + None, + description="A description of the traffic situation", + examples=["Crossing situation with 3 target vessels in the Oslofjord"], + ) + start_time: Optional[datetime] = Field( + None, + description="Starting time of the situation in `ISO 8601` format `YYYY-MM-DDThh:mm:ssZ`", + examples=[datetime.now()], + ) + own_ship: Union[OwnShip, Ship] = Field( + title="Own Ship data", + description="Own Ship data", + examples=[create_ship_example()], + ) + target_ships: List[Union[TargetShip, Ship]] = Field( + None, + title="Target Ship data", + description="Target Ship data", + examples=[[create_ship_example()]], + ) + environment: Optional[Environment] = Field( + None, + description="environmental parameters", + examples=[create_environment_example()], + ) + + model_config = ConfigDict( + extra="allow", + title="Test Input Schema", + json_schema_extra={"additionalProperties": True, "omit_default": True}, + ) + diff --git a/src/maritime_schema/types/ship.py b/src/maritime_schema/types/ship.py index eeb08c9..c68dd53 100644 --- a/src/maritime_schema/types/ship.py +++ b/src/maritime_schema/types/ship.py @@ -10,6 +10,8 @@ from pydantic.fields import Field # , FieldInfo from pyproj import Geod +from .state import State + logger = logging.getLogger(__name__) @@ -36,6 +38,19 @@ class AISNavStatus(str, Enum): NOT_DEFINED_DEFAULT = "Not defined (default)" +class ShipState(State): + """State of a ship.""" + nav_status: Optional[AISNavStatus] = Field( + None, description="Navigational status of the ship", examples=["Under way using engine"] + ) + + def __init__(self, nav_status=None, **data: Any): + super().__init__(**data) + self.nav_status = nav_status + if nav_status is None: + self.nav_status = AISNavStatus.NOT_DEFINED_DEFAULT + + class GeneralShipType(str, Enum): WING_IN_GROUND = "Wing in ground" FISHING = "Fishing" @@ -88,3 +103,59 @@ class ShipStatic(BaseModelConfig): name: Optional[str] = Field(None, description="Ship name", examples=["RMS Titanic"]) ship_type: Optional[GeneralShipType] = Field(None, description="General ship type, based on AIS") model_config = ConfigDict(extra="allow") + + +class Ship(BaseModelConfig): + static: ShipStatic = Field( + None, + description="Static ship information which does not change during a scenario.", + examples=[create_ship_static_example()], + ) + initial: Optional[Initial] = Field(None, examples=[create_initial_example()]) + waypoints: Optional[List[Waypoint]] = Field( + None, + description="An array of `Waypoint` objects. Each waypoint object must have a `position` property.
If no turn radius is provided, it will be assumed to be `0`.
Additional data can be added to each waypoint leg. This allows varying parameters on a per-leg basis, such as speed and heading, or navigational status ", + examples=[create_waypoint_example()], + ) + + def __init__(self, **data: Any): + super().__init__(**data) + if not self.waypoints: + self.waypoints = self.generate_waypoints() + + def generate_waypoints(self) -> Union[List[Waypoint], None]: + """Generate waypoints if they don't exist.""" + waypoints = [] + + if self.initial: + # Create waypoints from initial position + geod = Geod(ellps="WGS84") + lon, lat, _ = geod.fwd( + self.initial.position.longitude, + self.initial.position.latitude, + self.initial.cog, + 10000, # distance in meters + ) + position1 = Position(latitude=lat, longitude=lon) + waypoint1 = Waypoint(position=position1) + + position0 = Position( + latitude=self.initial.position.latitude, + longitude=self.initial.position.longitude, + ) + waypoint0 = Waypoint(position=position0, data=None) + + waypoints = [waypoint0, waypoint1] + return waypoints + + else: + return None + + +class OwnShip(Ship): + pass + + +class TargetShip(Ship): + pass + diff --git a/src/maritime_schema/types/state.py b/src/maritime_schema/types/state.py index e0f4a59..a72e612 100644 --- a/src/maritime_schema/types/state.py +++ b/src/maritime_schema/types/state.py @@ -1,4 +1,7 @@ +from turtle import position + + class Position(BaseModelConfig): latitude: float = Field(None, ge=-90, le=90, description="WGS-84 latitude", examples=[51.2131]) @@ -6,3 +9,24 @@ class Position(BaseModelConfig): description="WGS-84 longitude", examples=[11.2131]) model_config = ConfigDict(extra="allow") + + +class DynamicState(MTSBaseModel): + """Dynamic state of an object.""" + sog: Optional[float] = Field( + None, ge=0, description="Speed over ground in m/s", examples=[12.3] + ) + cog: Optional[float] = Field( + None, ge=0, le=360, description="Course over ground in degrees", examples=[180] + ) + heading: Optional[float] = Field( + None, ge=0, le=360, description="Heading in degrees", examples=[180] + ) + model_config = ConfigDict(extra="allow") + + +class State(MTSBaseModel): + position: Position = Field(Position, description="Geographical position") + dynamic: DynamicState = Field( + DynamicState, description="Dynamic state of an object" + ) \ No newline at end of file