maritime-expert

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: 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 &#x3C;= 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 &#x3C;= time &#x3C;= 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

Safety Notice

Copy this and send it to your AI assistant to learn

Source Transparency

Related Skills

finance-expert

trading-expert

dart-expert

postgresql-expert