【実務で活かす数理最適化tips】Pydanticを活用して意図しないエラーやInfeasibleを未然に回避しよう！

import logging
from src.preprocess import load_and_validate_data
from src.postprocess import process_results
from src.solver import CVRPSolver

# ロギングの設定
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

def main() -> None:
    locations_file = "data/locations.csv"
    vehicles_file = "data/vehicles.csv"
    
    logger.info("データを読み込みます")
    try:
        input_data = load_and_validate_data(locations_file, vehicles_file)
        logger.info("データの読み込みと検証が成功しました")
    except Exception as e:
        logger.error(f"データの検証エラー: {e}")
        return

    logger.info("最適化を実行します")
    solver = CVRPSolver(input_data)
    routes = solver.solve()

    logger.info("結果を表示します")    
    process_results(routes, input_data)

if __name__ == "__main__":
    main()

import logging
from src.schemas import CVRPInput

logger = logging.getLogger(__name__)

def process_results(routes: dict[int, list[int]], input_data: CVRPInput) -> None:
    """
    VRP最適化の結果を計算して表示します。
    """
    try:
        total_dist = 0
        node_map = {loc.location_id: loc for loc in input_data.locations}
        
        for id, route in routes.items():
            route_dist = 0
            load = 0
            
            # 表示用メトリクス計算
            for k in range(len(route) - 1):
                u, v = route[k], route[k+1]
                # ユークリッド距離計算
                d = ((node_map[u].x - node_map[v].x)**2 + (node_map[u].y - node_map[v].y)**2)**0.5
                route_dist += d
                load += node_map[v].demand
            
            total_dist += route_dist
            logger.info(f"車両 {id}: {route} | 距離: {route_dist:.2f} | 積載量: {load}")
        logger.info(f"総距離: {total_dist:.2f}")

    except:
        logger.warning("後処理でエラーが発生しました")

import pandas as pd
import logging
from src.schemas import CVRPInput, Location, Vehicle

logger = logging.getLogger(__name__)

def load_and_validate_data(locations_path: str, vehicles_path: str) -> CVRPInput:
    """
    位置情報と車両情報のCSVファイルを読み込み、Pydanticモデルにパースし、
    バリデーション済みのCVRPInputオブジェクトを返す
    """
    # CSV読み込み
    df_loc = pd.read_csv(locations_path)
    df_veh = pd.read_csv(vehicles_path)
    
    # 位置データをバリデーションしてリストに格納
    locations = []
    for _, row in df_loc.iterrows():
        locations.append(Location(
            location_id=row['location_id'],
            x=row['x'],
            y=row['y'],
            demand=row['demand']
        ))
        
    # 車両データをバリデーションしてリストに格納
    vehicles = []
    for _, row in df_veh.iterrows():
        vehicles.append(Vehicle(
            vehicle_id=row['vehicle_id'],
            capacity=row['capacity']
        ))
        
    # CVRPInputオブジェクトを作成してバリデーションを行う
    input_data = CVRPInput(locations=locations, vehicles=vehicles)
    return input_data

from pydantic import BaseModel, Field, field_validator, model_validator

# 位置情報を表すクラス
class Location(BaseModel):
    location_id: int
    x: float
    y: float
    demand: float = Field(ge=0)

# 車両情報を表すクラス
class Vehicle(BaseModel):
    vehicle_id: int
    capacity: float = Field(gt=0)

# 配送計画問題の入力データを表すクラス
class CVRPInput(BaseModel):
    locations: list[Location]
    vehicles: list[Vehicle]

    @field_validator('locations')
    @classmethod
    def validate_depot_exists(cls, v):
        if not any(loc.location_id == 0 for loc in v):
            raise ValueError("デポ（location_id = 0）の情報がlocations.csvに存在する必要があります。")
        return v

    @model_validator(mode='after')
    def validate_capacity_feasibility(self):
        # 最大荷物量と最大車両容量を計算して比較する。最大荷物量が最大車両容量を超える場合は実行不可能。
        max_demand = max((loc.demand for loc in self.locations), default=0)
        max_capacity = max((veh.capacity for veh in self.vehicles), default=0)
        if max_demand > max_capacity:
            raise ValueError(f"実行不可能: 最大需要 ({max_demand}) が最大車両容量 ({max_capacity}) を超えています。")
        return self

import pulp
import logging
import itertools

from src.schemas import CVRPInput

logger = logging.getLogger(__name__)

# 容量制約付き配送計画問題を定式化して解く最適化モデル
class CVRPSolver:
    def __init__(self, input_data: CVRPInput):
        self.locations = input_data.locations
        self.vehicles = input_data.vehicles
        self.num_vehicles = len(self.vehicles)
        self.prob = pulp.LpProblem("CVRP", pulp.LpMinimize)
        
        # パラメータ設定
        self.V = [loc.location_id for loc in self.locations]
        self.V_0 = [i for i in self.V if i != 0]
        self.R = [v.vehicle_id for v in self.vehicles]
        
        self.q = {loc.location_id: loc.demand for loc in self.locations}
        self.Q = {v.vehicle_id: v.capacity for v in self.vehicles}
        self.dist = self._calculate_distance_matrix()

        self._add_variables()  # 変数の設定
        self._add_objective()  # 目的関数の設定
        self._add_constraints()  # 制約条件の設定

    def _calculate_distance_matrix(self) -> dict[tuple[int, int], float]:
        """任意の2地点間のユークリッド距離を計算する"""
        dist = {}
        node_map = {loc.location_id: loc for loc in self.locations}
        for i in self.V:
            for j in self.V:
                d = ((node_map[i].x - node_map[j].x)**2 + (node_map[i].y - node_map[j].y)**2)**0.5
                dist[(i, j)] = d
        return dist

    def _powerset(self, locations: list[int]) -> list]:
        """地点集合の部分集合を生成する（部分巡回路除去制約で使用）"""
        return list(
            itertools.chain.from_iterable(itertools.combinations(locations, r)
            for r in range(len(locations)+1))
        )

    def _add_variables(self):
        # 変数: x[r, i, j] (車両rがiからjへ移動するか)
        self.x = pulp.LpVariable.dicts(
            "x", 
            [(r, i, j) for r in self.R for i in self.V for j in self.V if i != j], 
            cat="Binary"
        )

    def _add_objective(self):
        # 目的関数: 総移動距離の最小化
        self.prob += pulp.lpSum(
            self.dist[i, j] * self.x[r, i, j] 
            for r in self.R for i in self.V for j in self.V if i != j
        )

    def _add_constraints(self):
        # 1. 各顧客には必ず1回訪れる
        for j in self.V_0:
            self.prob += pulp.lpSum(self.x[r, i, j] for r in self.R for i in self.V if i != j) == 1

        # 2. 各車両はデポを1回出発する
        for r in self.R:
            self.prob += pulp.lpSum(self.x[r, 0, j] for j in self.V_0) == 1

        # 3. フロー保存制約 (入ったら出る)
        for r in self.R:
            for j in self.V:
                self.prob += (
                    pulp.lpSum(self.x[r, i, j] for i in self.V if i != j) - 
                    pulp.lpSum(self.x[r, j, k] for k in self.V if j != k) == 0
                )

        # 4. 容量制約
        for r in self.R:
            self.prob += pulp.lpSum(
                self.q[j] * self.x[r, i, j] for i in self.V for j in self.V_0 if i != j
            ) <= self.Q[r]

        # 5. 部分巡回路除去制約 (Powerset)
        subsets = self._powerset(self.V_0)
        for S in subsets:
            if len(S) >= 2:
                self.prob += pulp.lpSum(
                    self.x[r, i, j] for r in self.R for i in S for j in S if i != j
                ) <= len(S) - 1

    def solve(self) -> dict[int, list[int]]:
        status = self.prob.solve(pulp.PULP_CBC_CMD(msg=0))
        if status == pulp.LpStatusOptimal:
            return self._extract_routes()
        raise ValueError("最適化結果が得られませんでした")

    def _extract_routes(self) -> dict[int, list[int]]:
        routes = {}
        for r in self.R:
            path = []
            curr = 0
            
            # デポから開始してルート構築
            while True:
                next_node = None
                for j in self.V:
                    if curr != j:
                        if pulp.value(self.x[r, curr, j]) > 0.5:
                            next_node = j
                            break
                
                if next_node is None:
                    break
                
                path.append(next_node)
                curr = next_node
                if curr == 0:
                    break
            
            if path:
                routes[r] = [0] + path
            else:
                pass
                
        return routes