maritime-expert

name	maritime-expert
version	1.0.0
description	Expert-level maritime systems, vessel tracking, port operations, cargo management, and maritime logistics
category	domains
tags	["maritime","shipping","logistics","vessel","port","cargo"]
allowed-tools	["Read","Write","Edit"]

Maritime Expert

Expert guidance for maritime systems, vessel tracking, port operations, cargo management, maritime logistics, and shipping industry software.

Core Concepts

Maritime Systems

• Vessel Traffic Services (VTS)
• Port Management Systems
• Cargo Management Systems
• Fleet Management
• Maritime Communication Systems
• Container Terminal Operating Systems (TOS)
• Ship Performance Monitoring

Maritime Technologies

• AIS (Automatic Identification System)
• ECDIS (Electronic Chart Display and Information System)
• Satellite communication (VSAT)
• Weather routing systems
• Ballast water management
• Engine monitoring systems
• Container tracking (IoT)

Standards and Protocols

• IMO regulations (International Maritime Organization)
• SOLAS (Safety of Life at Sea)
• MARPOL (Marine Pollution)
• ISM Code (International Safety Management)
• ISPS Code (International Ship and Port Facility Security)
• UN/EDIFACT for EDI
• NMEA protocols

Vessel Tracking System

from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import List, Optional, Tuple
from decimal import Decimal
from enum import Enum
import numpy as np

class VesselType(Enum):
    CONTAINER = "container"
    BULK_CARRIER = "bulk_carrier"
    TANKER = "tanker"
    RO_RO = "ro_ro"
    CRUISE = "cruise"
    CARGO = "general_cargo"

class VesselStatus(Enum):
    UNDERWAY = "underway"
    AT_ANCHOR = "at_anchor"
    MOORED = "moored"
    NOT_UNDER_COMMAND = "not_under_command"
    RESTRICTED_MANEUVERABILITY = "restricted_maneuverability"

@dataclass
class Vessel:
    """Vessel information"""
    imo_number: str  # International Maritime Organization number
    mmsi: str  # Maritime Mobile Service Identity
    vessel_name: str
    vessel_type: VesselType
    flag: str
    call_sign: str
    length_m: float
    beam_m: float
    draft_m: float
    gross_tonnage: int
    deadweight_tonnage: 
    max_speed_kts: 
    current_position: [, ]
    heading: 
    speed_kts: 
    status: VesselStatus


 :
    
    voyage_id: 
    vessel_imo: 
    departure_port: 
    destination_port: 
    scheduled_departure: datetime
    scheduled_arrival: datetime
    actual_departure: [datetime]
    actual_arrival: [datetime]
    cargo_manifest: []
    route_waypoints: [[, ]]
    estimated_fuel_consumption: 

 :
    

     ():
        .vessels = {}
        .voyages = {}
        .ais_messages = []

     () -> :
        
        mmsi = ais_data[]
        vessel = ._get_vessel_by_mmsi(mmsi)

          vessel:
             {: , : mmsi}

        
        vessel.current_position = (ais_data[], ais_data[])
        vessel.heading = ais_data.get(, )
        vessel.speed_kts = ais_data.get(, )
        vessel.status = VesselStatus(ais_data.get(, ))

        
        .ais_messages.append({
            : datetime.now(),
            : mmsi,
            : vessel.current_position,
            : vessel.speed_kts,
            : vessel.heading
        })

        
        anomalies = ._detect_anomalies(vessel, ais_data)

         {
            : mmsi,
            : vessel.vessel_name,
            : vessel.current_position,
            : vessel.speed_kts,
            : vessel.heading,
            : vessel.status.value,
            : anomalies,
            : datetime.now().isoformat()
        }

     () -> []:
        
        anomalies = []

        
         vessel.speed_kts > vessel.max_speed_kts * :
            anomalies.append({
                : ,
                : ,
                : 
            })

        
           ais_data  ais_data[] > vessel.draft_m * :
            anomalies.append({
                : ,
                : ,
                : 
            })

        
         vessel.status == VesselStatus.AT_ANCHOR  vessel.speed_kts > :
            anomalies.append({
                : ,
                : ,
                : 
            })

         anomalies

     () -> :
        
        voyage = .voyages.get(voyage_id)
          voyage:
             {: }

        vessel = .vessels.get(voyage.vessel_imo)
          vessel:
             {: }

        
        dest_coords = ._get_port_coordinates(voyage.destination_port)
        remaining_distance_nm = ._calculate_distance(
            vessel.current_position,
            dest_coords
        )

        
         vessel.speed_kts > :
            hours_remaining = remaining_distance_nm / vessel.speed_kts
            eta = datetime.now() + timedelta(hours=hours_remaining)
        :
            
            avg_speed = vessel.max_speed_kts *   
            hours_remaining = remaining_distance_nm / avg_speed
            eta = datetime.now() + timedelta(hours=hours_remaining)

        
        delay_hours = (eta - voyage.scheduled_arrival).total_seconds() / 

         {
            : voyage_id,
            : vessel.vessel_name,
            : voyage.destination_port,
            : vessel.current_position,
            : remaining_distance_nm,
            : vessel.speed_kts,
            : eta.isoformat(),
            : voyage.scheduled_arrival.isoformat(),
            : delay_hours,
            : delay_hours <= 
        }

     () -> :
        
        dest_coords = ._get_port_coordinates(destination)

        
        gc_distance = ._calculate_distance(start_position, dest_coords)

        
        weather = ._get_weather_forecast(start_position, dest_coords, departure_time)

        
        routes = [
            {
                : ,
                : gc_distance,
                : ._generate_waypoints(start_position, dest_coords, )
            },
            {
                : ,
                : gc_distance * ,  
                : ._generate_weather_route(start_position, dest_coords, weather)
            }
        ]

        
         route  routes:
            avg_speed =   
            transit_time = route[] / avg_speed
            fuel_consumption = ._estimate_fuel_consumption(
                route[],
                vessel_type,
                avg_speed
            )

            route[] = transit_time
            route[] = fuel_consumption
            route[] = fuel_consumption *   

        
        recommended = (routes, key= r: r[])

         {
            : routes,
            : recommended[],
            : {
                : routes[][] - recommended[],
                : routes[][] - recommended[]
            }
        }

     () -> :
        
         math  radians, sin, cos, sqrt, atan2

        lat1, lon1 = radians(point1[]), radians(point1[])
        lat2, lon2 = radians(point2[]), radians(point2[])

        dlat = lat2 - lat1
        dlon = lon2 - lon1

        a = sin(dlat/)** + cos(lat1) * cos(lat2) * sin(dlon/)**
        c =  * atan2(sqrt(a), sqrt(-a))

        distance_km =  * c
        distance_nm = distance_km * 

         distance_nm

     () -> [Vessel]:
        
         vessel  .vessels.values():
             vessel.mmsi == mmsi:
                 vessel
         

     () -> [, ]:
        
        ports = {
            : (, -),  
            : (, ),    
            : (, ),   
            : (, )   
        }
         ports.get(port_code, (, ))

     () -> [[, ]]:
        
        waypoints = []
         i  (count + ):
            fraction = i / count
            lat = start[] + (end[] - start[]) * fraction
            lon = start[] + (end[] - start[]) * fraction
            waypoints.append((lat, lon))
         waypoints

     () -> :
        
        
         {: , : }

     () -> [[, ]]:
        
        
         ._generate_waypoints(start, end, )

     () -> :
        
        
        daily_consumption = {
            VesselType.CONTAINER: ,
            VesselType.BULK_CARRIER: ,
            VesselType.TANKER: 
        }

        base_consumption = daily_consumption.get(vessel_type, )

        
        speed_factor = (speed_kts / ) ** 

        days_at_sea = (distance_nm / speed_kts) / 
        total_fuel = base_consumption * days_at_sea * speed_factor

         total_fuel

Port Operations System

@dataclass
class BerthAllocation:
    """Berth allocation for vessel"""
    allocation_id: str
    vessel_imo: str
    berth_id: str
    scheduled_arrival: datetime
    scheduled_departure: datetime
    actual_arrival: Optional[datetime]
    actual_departure: Optional[datetime]
    cargo_operations: List[dict]

class PortOperationsSystem:
    """Port and terminal operations management"""

    def __init__(self):
        self.berths = {}
        self.allocations = []
        self.cargo_operations = []

    def allocate_berth(self, vessel_imo: str, eta: datetime, cargo_type: str) -> dict:
        """Allocate berth for arriving vessel"""
        # Find suitable berth
        suitable_berth = self._find_suitable_berth(cargo_type, eta)

        if not suitable_berth:
            return {'error': 'No suitable berth available'}

        # Estimate time at berth
        time_at_berth = self._estimate_port_time(cargo_type)

        allocation = BerthAllocation(
            allocation_id=self._generate_allocation_id(),
            vessel_imo=vessel_imo,
            berth_id=suitable_berth['berth_id'],
            scheduled_arrival=eta,
            scheduled_departure=eta + timedelta(hours=time_at_berth),
            actual_arrival=None,
            actual_departure=,
            cargo_operations=[]
        )

        .allocations.append(allocation)

         {
            : allocation.allocation_id,
            : suitable_berth[],
            : eta.isoformat(),
            : allocation.scheduled_departure.isoformat(),
            : time_at_berth
        }

     () -> :
        
        
        container_data = {
            : container_number,
            : ,
            : ,
            : datetime.now() - timedelta(hours=),
            : ,
            : ,
            :   
        }

         container_data

     () -> :
        
        
        

         {
            : expected_moves,
            : ,
            : expected_moves * ,  
            : {
                : expected_moves // ,
                : expected_moves // 
            }
        }

     () -> []:
        
        
         berth_id, berth  .berths.items():
             cargo_type  berth[]:
                
                 ._is_berth_available(berth_id, eta):
                     berth
         

     () -> :
        
         allocation  .allocations:
             allocation.berth_id == berth_id:
                 allocation.scheduled_arrival <= time <= allocation.scheduled_departure:
                     
         

     () -> :
        
        port_times = {
            : ,
            : ,
            : ,
            : 
        }
         port_times.get(cargo_type, )

     () -> :
         uuid
         

Cargo Management

class CargoManagementSystem:
    """Cargo and freight management"""

    def calculate_stowage_plan(self, containers: List[dict], vessel_capacity: dict) -> dict:
        """Calculate optimal container stowage plan"""
        # Simplified stowage planning
        # In production, would use sophisticated algorithms

        # Sort containers by weight (heaviest on bottom)
        sorted_containers = sorted(containers, key=lambda c: c['weight'], reverse=True)

        stowage_plan = {
            'bay_plans': [],
            'total_containers': len(containers),
            'total_weight': sum(c['weight'] for c in containers),
            'utilization': (len(containers) / vessel_capacity['max_containers']) * 100
        }

        return stowage_plan

    def track_bill_of_lading(self, bl_number: str) -> dict:
        """Track shipment by Bill of Lading"""
        # Track cargo shipment
        return {
            'bl_number': bl_number,
            'status': 'in_transit',
            'current_location': 'At Sea',
            : ,
            : ,
            : ,
            : (datetime.now() + timedelta(days=)).isoformat()
        }

Best Practices

Vessel Operations

• Maintain accurate AIS transmission
• Follow IMO regulations strictly
• Implement fuel optimization
• Conduct regular safety drills
• Maintain proper manning levels
• Use weather routing services
• Implement environmental compliance

Port Operations

• Optimize berth allocation
• Minimize vessel waiting time
• Implement automated gate systems
• Use container tracking technology
• Optimize yard operations
• Maintain equipment reliability
• Ensure security compliance (ISPS)

Cargo Management

• Maintain accurate documentation
• Implement proper stowage planning
• Use standardized EDI messages
• Track cargo in real-time
• Ensure proper handling of dangerous goods
• Maintain cold chain for reefers
• Implement quality control

Safety and Environment

• Follow SOLAS requirements
• Implement ISM Code
• Comply with MARPOL regulations
• Conduct risk assessments
• Maintain pollution prevention
• Implement ballast water management
• Train crew regularly

Anti-Patterns

❌ Inaccurate AIS data transmission ❌ Poor cargo documentation ❌ Inefficient port operations ❌ No weather routing ❌ Inadequate maintenance ❌ Poor crew training ❌ Ignoring environmental regulations ❌ No cargo tracking ❌ Inefficient fuel management

Resources

• IMO (International Maritime Organization): https://www.imo.org/
• ICS (International Chamber of Shipping): https://www.ics-shipping.org/
• BIMCO: https://www.bimco.org/
• Marine Traffic: https://www.marinetraffic.com/
• Port Technology: https://www.porttechnology.org/
• Maritime and Port Authority: https://www.mpa.gov.sg/
• SOLAS Convention: https://www.imo.org/en/About/Conventions/Pages/SOLAS.aspx