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: int
    max_speed_kts: float
    current_position: Tuple[float, float]
    heading: float
    speed_kts: float
    status: VesselStatus

@dataclass
class Voyage:
    """Voyage information"""
    voyage_id: str
    vessel_imo: str
    departure_port: str
    destination_port: str
    scheduled_departure: datetime
    scheduled_arrival: datetime
    actual_departure: Optional[datetime]
    actual_arrival: Optional[datetime]
    cargo_manifest: List[dict]
    route_waypoints: List[Tuple[float, float]]
    estimated_fuel_consumption: float

class VesselTrackingSystem:
    """Maritime vessel tracking and monitoring"""

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

    def process_ais_message(self, ais_data: dict) -> dict:
        """Process AIS position report"""
        mmsi = ais_data['mmsi']
        vessel = self._get_vessel_by_mmsi(mmsi)

        if not vessel:
            return {'error': 'Vessel not found', 'mmsi': mmsi}

        # Update vessel position
        vessel.current_position = (ais_data['latitude'], ais_data['longitude'])
        vessel.heading = ais_data.get('heading', 0)
        vessel.speed_kts = ais_data.get('speed', 0)
        vessel.status = VesselStatus(ais_data.get('status', 'underway'))

        # Store AIS message
        self.ais_messages.append({
            'timestamp': datetime.now(),
            'mmsi': mmsi,
            'position': vessel.current_position,
            'speed': vessel.speed_kts,
            'heading': vessel.heading
        })

        # Check for anomalies
        anomalies = self._detect_anomalies(vessel, ais_data)

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

    def _detect_anomalies(self, vessel: Vessel, ais_data: dict) -> List[dict]:
        """Detect unusual vessel behavior"""
        anomalies = []

        # Speed anomaly
        if vessel.speed_kts > vessel.max_speed_kts * 1.1:
            anomalies.append({
                'type': 'excessive_speed',
                'severity': 'medium',
                'message': f'Speed {vessel.speed_kts} kts exceeds maximum'
            })

        # Draft anomaly
        if 'draft' in ais_data and ais_data['draft'] > vessel.draft_m * 1.2:
            anomalies.append({
                'type': 'excessive_draft',
                'severity': 'high',
                'message': 'Draft exceeds vessel specifications'
            })

        # Unexpected stop
        if vessel.status == VesselStatus.AT_ANCHOR and vessel.speed_kts > 0.5:
            anomalies.append({
                'type': 'anchor_drag',
                'severity': 'critical',
                'message': 'Vessel moving while at anchor'
            })

        return anomalies

    def calculate_eta(self, voyage_id: str) -> dict:
        """Calculate estimated time of arrival"""
        voyage = self.voyages.get(voyage_id)
        if not voyage:
            return {'error': 'Voyage not found'}

        vessel = self.vessels.get(voyage.vessel_imo)
        if not vessel:
            return {'error': 'Vessel not found'}

        # Calculate remaining distance
        dest_coords = self._get_port_coordinates(voyage.destination_port)
        remaining_distance_nm = self._calculate_distance(
            vessel.current_position,
            dest_coords
        )

        # Calculate ETA based on current speed
        if vessel.speed_kts > 0:
            hours_remaining = remaining_distance_nm / vessel.speed_kts
            eta = datetime.now() + timedelta(hours=hours_remaining)
        else:
            # Use average speed if vessel is stopped
            avg_speed = vessel.max_speed_kts * 0.7  # Assume 70% of max
            hours_remaining = remaining_distance_nm / avg_speed
            eta = datetime.now() + timedelta(hours=hours_remaining)

        # Calculate delay
        delay_hours = (eta - voyage.scheduled_arrival).total_seconds() / 3600

        return {
            'voyage_id': voyage_id,
            'vessel_name': vessel.vessel_name,
            'destination': voyage.destination_port,
            'current_position': vessel.current_position,
            'remaining_distance_nm': remaining_distance_nm,
            'current_speed_kts': vessel.speed_kts,
            'estimated_arrival': eta.isoformat(),
            'scheduled_arrival': voyage.scheduled_arrival.isoformat(),
            'delay_hours': delay_hours,
            'on_schedule': delay_hours <= 0
        }

    def optimize_route(self,
                      start_position: Tuple[float, float],
                      destination: str,
                      vessel_type: VesselType,
                      departure_time: datetime) -> dict:
        """Optimize vessel route considering weather and fuel"""
        dest_coords = self._get_port_coordinates(destination)

        # Calculate great circle route
        gc_distance = self._calculate_distance(start_position, dest_coords)

        # Get weather forecast
        weather = self._get_weather_forecast(start_position, dest_coords, departure_time)

        # Calculate fuel consumption for different routes
        routes = [
            {
                'name': 'Great Circle',
                'distance_nm': gc_distance,
                'waypoints': self._generate_waypoints(start_position, dest_coords, 10)
            },
            {
                'name': 'Weather Optimized',
                'distance_nm': gc_distance * 1.05,  # 5% longer to avoid weather
                'waypoints': self._generate_weather_route(start_position, dest_coords, weather)
            }
        ]

        # Calculate fuel and time for each route
        for route in routes:
            avg_speed = 18.0  # knots
            transit_time = route['distance_nm'] / avg_speed
            fuel_consumption = self._estimate_fuel_consumption(
                route['distance_nm'],
                vessel_type,
                avg_speed
            )

            route['transit_time_hours'] = transit_time
            route['fuel_consumption_mt'] = fuel_consumption
            route['estimated_fuel_cost'] = fuel_consumption * 500  # $500/MT

        # Recommend optimal route
        recommended = min(routes, key=lambda r: r['estimated_fuel_cost'])

        return {
            'routes': routes,
            'recommended_route': recommended['name'],
            'savings': {
                'fuel_mt': routes[0]['fuel_consumption_mt'] - recommended['fuel_consumption_mt'],
                'cost_usd': routes[0]['estimated_fuel_cost'] - recommended['estimated_fuel_cost']
            }
        }

    def _calculate_distance(self, point1: Tuple[float, float], point2: Tuple[float, float]) -> float:
        """Calculate great circle distance in nautical miles"""
        from math import radians, sin, cos, sqrt, atan2

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

        dlat = lat2 - lat1
        dlon = lon2 - lon1

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

        distance_km = 6371 * c
        distance_nm = distance_km * 0.539957

        return distance_nm

    def _get_vessel_by_mmsi(self, mmsi: str) -> Optional[Vessel]:
        """Get vessel by MMSI"""
        for vessel in self.vessels.values():
            if vessel.mmsi == mmsi:
                return vessel
        return None

    def _get_port_coordinates(self, port_code: str) -> Tuple[float, float]:
        """Get port coordinates"""
        ports = {
            'USNYC': (40.6694, -74.0450),  # New York
            'NLRTM': (51.9244, 4.4777),    # Rotterdam
            'SGSIN': (1.2644, 103.8227),   # Singapore
            'CNSHA': (31.2304, 121.4737)   # Shanghai
        }
        return ports.get(port_code, (0.0, 0.0))

    def _generate_waypoints(self, start: Tuple[float, float], end: Tuple[float, float], count: int) -> List[Tuple[float, float]]:
        """Generate waypoints along great circle route"""
        waypoints = []
        for i in range(count + 1):
            fraction = i / count
            lat = start[0] + (end[0] - start[0]) * fraction
            lon = start[1] + (end[1] - start[1]) * fraction
            waypoints.append((lat, lon))
        return waypoints

    def _get_weather_forecast(self, start: Tuple[float, float], end: Tuple[float, float], time: datetime) -> dict:
        """Get weather forecast for route"""
        # Would integrate with weather API
        return {'wind_speed': 15, 'wave_height': 2.5}

    def _generate_weather_route(self, start: Tuple[float, float], end: Tuple[float, float], weather: dict) -> List[Tuple[float, float]]:
        """Generate weather-optimized route"""
        # Simplified - would use sophisticated weather routing
        return self._generate_waypoints(start, end, 12)

    def _estimate_fuel_consumption(self, distance_nm: float, vessel_type: VesselType, speed_kts: float) -> float:
        """Estimate fuel consumption in metric tons"""
        # Fuel consumption rates (MT per day at cruising speed)
        daily_consumption = {
            VesselType.CONTAINER: 80,
            VesselType.BULK_CARRIER: 30,
            VesselType.TANKER: 50
        }

        base_consumption = daily_consumption.get(vessel_type, 40)

        # Speed factor (fuel increases with cube of speed)
        speed_factor = (speed_kts / 18.0) ** 3

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

        return 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=None,
            cargo_operations=[]
        )

        self.allocations.append(allocation)

        return {
            'allocation_id': allocation.allocation_id,
            'berth_id': suitable_berth['berth_id'],
            'scheduled_arrival': eta.isoformat(),
            'scheduled_departure': allocation.scheduled_departure.isoformat(),
            'estimated_hours_at_berth': time_at_berth
        }

    def track_container(self, container_number: str) -> dict:
        """Track container through port"""
        # Container tracking using IoT sensors
        container_data = {
            'container_number': container_number,
            'status': 'in_yard',
            'location': 'Block A, Row 12, Tier 3',
            'last_move': datetime.now() - timedelta(hours=2),
            'vessel_loaded': None,
            'customs_cleared': True,
            'temperature': 5.0  # For reefer containers
        }

        return container_data

    def optimize_yard_operations(self, expected_moves: int) -> dict:
        """Optimize container yard operations"""
        # Simplified yard optimization
        # In production, would use complex algorithms

        return {
            'expected_moves': expected_moves,
            'optimal_sequence': 'calculated',
            'estimated_time_hours': expected_moves * 0.1,  # 6 minutes per move
            'crane_allocation': {
                'crane_1': expected_moves // 2,
                'crane_2': expected_moves // 2
            }
        }

    def _find_suitable_berth(self, cargo_type: str, eta: datetime) -> Optional[dict]:
        """Find suitable berth for vessel"""
        # Check berth availability and suitability
        for berth_id, berth in self.berths.items():
            if cargo_type in berth['cargo_types']:
                # Check if berth is available
                if self._is_berth_available(berth_id, eta):
                    return berth
        return None

    def _is_berth_available(self, berth_id: str, time: datetime) -> bool:
        """Check if berth is available at given time"""
        for allocation in self.allocations:
            if allocation.berth_id == berth_id:
                if allocation.scheduled_arrival <= time <= allocation.scheduled_departure:
                    return False
        return True

    def _estimate_port_time(self, cargo_type: str) -> float:
        """Estimate time vessel will spend in port (hours)"""
        port_times = {
            'container': 24,
            'bulk': 48,
            'tanker': 18,
            'general_cargo': 36
        }
        return port_times.get(cargo_type, 24)

    def _generate_allocation_id(self) -> str:
        import uuid
        return f"BERTH-{uuid.uuid4().hex[:8].upper()}"

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',
            'vessel': 'MV EXAMPLE',
            'departure_port': 'CNSHA',
            'destination_port': 'USNYC',
            'eta': (datetime.now() + timedelta(days=18)).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

maritime-expert

Maritime Expert

Core Concepts

Maritime Systems