kishorkukreja/awesome-supply-chain/energy-logistics

import numpy as np
from scipy.optimize import linprog
def optimize_pipeline_flow(pipeline_network, supply, demand, capacities, costs):
    """
    Optimize flow through pipeline network
    Parameters:
    - pipeline_network: dict of {(source, destination): pipeline_id}
    - supply: dict of {source: available_volume}
    - demand: dict of {destination: required_volume}
    - capacities: dict of {pipeline_id: max_flow_rate}
    - costs: dict of {pipeline_id: cost_per_barrel}
    """
    from pulp import *
    # Create problem
    prob = LpProblem("Pipeline_Flow", LpMinimize)
    # Decision variables: flow through each pipeline
    flows = {}
    for (src, dest), pipe_id in pipeline_network.items():
        flows[pipe_id] = LpVariable(
            f"Flow_{src}_to_{dest}",
            lowBound=0,
            upBound=capacities[pipe_id]
        )
    # Objective: minimize transportation cost
    prob += lpSum([costs[pipe] * flows[pipe]
                   for pipe in flows])
    # Constraints: supply limits
    for source, max_supply in supply.items():
        outbound = [flows[pipe_id]
                   for (src, dest), pipe_id in pipeline_network.items()
                   if src == source]
        if outbound:
            prob += lpSum(outbound) <= max_supply
    # Constraints: demand satisfaction
    for destination, required in demand.items():
        inbound = [flows[pipe_id]
                  for (src, dest), pipe_id in pipeline_network.items()
                  if dest == destination]
        if inbound:
            prob += lpSum(inbound) >= required
    # Solve
    prob.solve(PULP_CBC_CMD(msg=0))
    return {
        'status': LpStatus[prob.status],
        'total_cost': value(prob.objective),
        'flows': {pipe: flows[pipe].varValue for pipe in flows}
    }
# Example usage
network = {
    ('Field_A', 'Terminal_1'): 'Pipe_1',
    ('Field_B', 'Terminal_1'): 'Pipe_2',
    ('Terminal_1', 'Refinery_1'): 'Pipe_3',
    ('Terminal_1', 'Refinery_2'): 'Pipe_4'
}
supply = {
    'Field_A': 100000,  # barrels/day
    'Field_B': 150000
}
demand = {
    'Refinery_1': 120000,
    'Refinery_2': 80000
}
capacities = {
    'Pipe_1': 110000,
    'Pipe_2': 160000,
    'Pipe_3': 130000,
    'Pipe_4': 90000
}
costs = {
    'Pipe_1': 0.50,  # $/barrel
    'Pipe_2': 0.45,
    'Pipe_3': 0.60,
    'Pipe_4': 0.55
}
result = optimize_pipeline_flow(network, supply, demand, capacities, costs)
print(f"Optimal daily cost: ${result['total_cost']:,.2f}")

def batch_schedule_pipeline(batches, pipeline_capacity, transit_times):
    """
    Schedule multiple product batches through pipeline
    Parameters:
    - batches: list of {product, volume, source, destination, priority}
    - pipeline_capacity: barrels/hour
    - transit_times: dict of {(source, destination): hours}
    """
    from collections import deque
    schedule = []
    current_time = 0
    # Sort by priority and volume
    sorted_batches = sorted(batches,
                           key=lambda x: (x['priority'], -x['volume']))
    for batch in sorted_batches:
        # Calculate transit time
        transit = transit_times.get(
            (batch['source'], batch['destination']), 24
        )
        # Calculate pump time
        pump_time = batch['volume'] / pipeline_capacity
        # Schedule
        schedule.append({
            'batch_id': batch['product'],
            'start_time': current_time,
            'pump_hours': pump_time,
            'arrival_time': current_time + pump_time + transit,
            'volume': batch['volume']
        })
        current_time += pump_time
    return schedule
# Example
batches = [
    {'product': 'Crude_Light', 'volume': 50000,
     'source': 'Terminal_A', 'destination': 'Refinery_1', 'priority': 1},
    {'product': 'Crude_Heavy', 'volume': 75000,
     'source': 'Terminal_A', 'destination': 'Refinery_2', 'priority': 2},
]
schedule = batch_schedule_pipeline(batches, pipeline_capacity=2500,
                                   transit_times={('Terminal_A', 'Refinery_1'): 20})

def optimize_tanker_fleet(shipments, vessels, ports, costs):
    """
    Optimize tanker routing and scheduling
    Parameters:
    - shipments: list of {origin, destination, volume, earliest, latest}
    - vessels: list of {vessel_id, capacity, speed, position, available_time}
    - ports: dict of {port: {lat, lon}}
    - costs: dict with fuel_cost, charter_rate, port_fees
    """
    import numpy as np
    from pulp import *
    prob = LpProblem("Tanker_Routing", LpMinimize)
    # Variables: assign vessel v to shipment s
    x = {}
    for v, vessel in enumerate(vessels):
        for s, shipment in enumerate(shipments):
            if vessel['capacity'] >= shipment['volume']:
                x[v, s] = LpVariable(f"Assign_{v}_{s}", cat='Binary')
    # Objective: minimize total cost
    total_cost = []
    for (v, s), var in x.items():
        vessel = vessels[v]
        shipment = shipments[s]
        # Distance calculation (simplified)
        distance = calculate_sea_distance(
            ports[shipment['origin']],
            ports[shipment['destination']]
        )
        # Voyage cost
        voyage_time = distance / vessel['speed']  # hours
        fuel_cost = costs['fuel_cost'] * distance * vessel['capacity'] * 0.001
        charter_cost = costs['charter_rate'] * voyage_time
        port_cost = costs['port_fees'] * 2  # origin + destination
        total_cost.append((fuel_cost + charter_cost + port_cost) * var)
    prob += lpSum(total_cost)
    # Constraints: each shipment covered once
    for s in range(len(shipments)):
        prob += lpSum([x[v, s] for v, _ in x.keys() if _ == s]) == 1
    # Constraints: vessel availability
    for v in range(len(vessels)):
        assignments = [x[v, s] for _, s in x.keys() if _ == v]
        if assignments:
            prob += lpSum(assignments) <= 1  # One voyage at a time
    prob.solve(PULP_CBC_CMD(msg=0))
    return {
        'status': LpStatus[prob.status],
        'total_cost': value(prob.objective),
        'assignments': [(v, s) for (v, s) in x if x[v, s].varValue > 0.5]
    }
def calculate_sea_distance(port1, port2):
    """Calculate great circle distance between ports"""
    from math import radians, sin, cos, sqrt, atan2
    lat1, lon1 = radians(port1['lat']), radians(port1['lon'])
    lat2, lon2 = radians(port2['lat']), radians(port2['lon'])
    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))
    return 3959 * c  # miles

def optimize_truck_routes(deliveries, depot_location, truck_capacity, max_hours):
    """
    Vehicle routing for fuel delivery trucks
    Uses Clarke-Wright savings algorithm
    """
    import numpy as np
    # Calculate distances
    def distance(loc1, loc2):
        return np.sqrt((loc1[0] - loc2[0])**2 + (loc1[1] - loc2[1])**2)
    # Calculate savings for combining routes
    savings = []
    n = len(deliveries)
    for i in range(n):
        for j in range(i+1, n):
            save = (distance(depot_location, deliveries[i]['location']) +
                   distance(depot_location, deliveries[j]['location']) -
                   distance(deliveries[i]['location'], deliveries[j]['location']))
            savings.append((save, i, j))
    # Sort by savings (descending)
    savings.sort(reverse=True)
    # Build routes
    routes = [[i] for i in range(n)]
    route_loads = [deliveries[i]['volume'] for i in range(n)]
    for save, i, j in savings:
        # Find routes containing i and j
        route_i = next((r for r in routes if i in r), None)
        route_j = next((r for r in routes if j in r), None)
        if route_i != route_j and route_i and route_j:
            # Check if merge is feasible
            combined_load = route_loads[routes.index(route_i)] + \
                           route_loads[routes.index(route_j)]
            if combined_load <= truck_capacity:
                # Merge routes
                route_i.extend(route_j)
                route_loads[routes.index(route_i)] = combined_load
                routes.remove(route_j)
    return {
        'num_trucks': len(routes),
        'routes': routes,
        'utilization': [route_loads[routes.index(r)] / truck_capacity
                       for r in routes]
    }

import numpy as np
import pandas as pd
class EnergyInventoryOptimizer:
    """
    Optimize inventory levels for energy products
    considering price volatility and storage costs
    """