SCMOPT REST API & Validiation

Pydanticによるデータフレーム検証

Excelファイルやcsvファイルを読み込んだ際に,データの妥当性の検証を行う必要がある. そのために,fastAPI (https://fastapi.tiangolo.com/) に含まれるPydantic(https://pydantic-docs.helpmanual.io/)を利用する.

Pydanticは,オブジェクトタイプが間違えていても,できるだけ自動変換する.

MELOS-GF

Weiszfeld法のREST(Representational State Transfer) APIを FastAPI とpydanticを用いて作成する.

MelosGfのモデルクラスは,顧客クラスCustomer,移動時間クラス Time から構成され,以下のように定義される.

クラス定義

データ検証用に顧客リスト,時間リストのクラスも作っておく.

class Customer[source]

Customer(name:str, lat:float, lon:float, weight:float=None) :: BaseModel

class CustomerList[source]

CustomerList(field_list:List[Customer]=[]) :: BaseModel

class Time[source]

Time(from_node:int, from_name:str=None, to_node:int, to_name:str=None, time:int, distance:int=None) :: BaseModel

class TimeList[source]

TimeList(field_list:List[Time]=[]) :: BaseModel

class MelosGf[source]

MelosGf(num_facilities:ConstrainedIntValue=1, epsilon:ConstrainedFloatValue=1e-05, max_iter:ConstrainedIntValue=10, seed:int=1, customers:List[Customer]=[], times:List[Time]=[]) :: BaseModel

Melos Green Field モデル

顧客クラスの使用例

PydanticのBaseClassから派生させたクラス

引数は辞書で与え,入力データにエラーがある場合には,ValidationErrorが発生する.

クラスインスタンスには,以下のメソッドがある.

  • json
  • schema
  • dict
from pprint import pprint
example  = {
                "name": "響本店",
                "lat": "43.06417", #文字列はfloatに変換される
                "lon": 141.34694,
                "weight": 1 #整数もfloatに変換される
            }
cust = Customer(**example) #Pydantic BaseClassは辞書から生成できる.(**で展開)
print(cust) #インスタンス文字列
pprint(cust.json()) #JSON文字列 https://www.json.org/json-ja.html
pprint(cust.schema()) #JSONスキーマ
print(cust.dict()) #辞書に変換
print(jsonable_encoder(cust)) #JSONで読める形式(この場合は辞書) https://fastapi.tiangolo.com/tutorial/encoder/
name='響本店' lat=43.06417 lon=141.34694 weight=1.0
('{"name": "\\u97ff\\u672c\\u5e97", "lat": 43.06417, "lon": 141.34694, '
 '"weight": 1.0}')
{'example': {'lat': 43.06417, 'lon': 141.34694, 'name': '響本店', 'weight': 100.0},
 'properties': {'lat': {'description': '緯度', 'title': 'Lat', 'type': 'number'},
                'lon': {'description': '経度', 'title': 'Lon', 'type': 'number'},
                'name': {'description': '名称',
                         'title': 'Name',
                         'type': 'string'},
                'weight': {'description': '重み',
                           'title': 'Weight',
                           'type': 'number'}},
 'required': ['name', 'lat', 'lon'],
 'title': 'Customer',
 'type': 'object'}
{'name': '響本店', 'lat': 43.06417, 'lon': 141.34694, 'weight': 1.0}
{'name': '響本店', 'lat': 43.06417, 'lon': 141.34694, 'weight': 1.0}
example2  = {
                "name": "響本店",
                "lat": "43.06417"
            }
try:
    cust2 = Customer(**example2)
except ValidationError as e:
    print(e.json())
[
  {
    "loc": [
      "lon"
    ],
    "msg": "field required",
    "type": "value_error.missing"
  }
]

顧客リストクラス

データフレームを辞書に変換してCustomerListクラスに渡す.

逆にモデルからデータを得ることもできる.

モデルの属性から顧客リストを得た後に, FastAPIのjsonable_encoder関数で辞書のリストに変換し,そこからデータフレームを生成する.

sheet = pd.read_excel("melos-gf.xlsx", sheet_name=None) #read all sheets 
cust_df = sheet["melos-gf"] #extract dataframe 
model = CustomerList(**{"field_list":cust_df.to_dict("records")})
pprint(jsonable_encoder(model.field_list)[:3]) #JSONで読める形式(この場合はリスト)
pd.DataFrame(jsonable_encoder(model.field_list)).head() #データフレームに戻す
[{'lat': 43.06417,
  'lon': 141.34694,
  'name': '札幌市',
  'weight': 9632.389999999998},
 {'lat': 40.82444, 'lon': 140.74, 'name': '青森市', 'weight': 8607.569999999998},
 {'lat': 39.70361, 'lon': 141.1525, 'name': '盛岡市', 'weight': 9614.31}]
name lat lon weight
0 札幌市 43.06417 141.34694 9632.39
1 青森市 40.82444 140.74000 8607.57
2 盛岡市 39.70361 141.15250 9614.31
3 仙台市 38.26889 140.87194 11050.32
4 秋田市 39.71861 140.10250 2280.27

データフレームを検証する関数 valid_test

エラーがあるとJSON形式でエラーを捉えることがでいるので,それをエラーデータフレームに変換して返す.

引数:

  • Cls: 検証用のクラス
  • df: 検証したいデータフレーム

返値:

  • データにエラーがある場合には,エラーの情報を含んだデータフレームを返し,エラーがない場合にはNoneを返す.

valid_test[source]

valid_test(Cls:BaseModel, df:DataFrame)

valid_test関数の使用例

sheet = pd.read_excel("melos-gf.xlsx", sheet_name=None) #read all sheets 
cust_df = sheet["melos-gf"] #extract dataframe 
time_df = sheet["time"]

#エラー挿入
cust_df.iloc[0,1] = None
cust_df.iloc[0,2] = np.nan
cust_df.iloc[0,4] = "1.2345" #大文字だとエラー

ret = valid_test(CustomerList, cust_df)
print(ret)

ret = valid_test(TimeList, time_df)
print(ret)
   row     col                       message
0    0    name  none is not an allowed value
1    0     lat  none is not an allowed value
2    0  weight    value is not a valid float
None

すべてのデータフレームを検証する関数 test_all

引数:

  • df_list: データフレームのリスト
  • cls_list: 検証用のクラスのリスト
  • name_list: データの名称のリスト

返値:

  • errorsheet: エラーがあるデータの名称のリスト
  • error_df: エラーの情報を含んだデータフレームのリスト

test_all[source]

test_all(df_list, cls_list, name_list)

test_all関数の使用例

sheet = pd.read_excel("melos-gf.xlsx", sheet_name=None) 
#エラー挿入
sheet["melos-gf"].iloc[0,1] = None
sheet["melos-gf"].iloc[0,2] = np.nan
sheet["time"].iloc[0,5] = "1.2345" #大文字だとエラー

cust_df = sheet["melos-gf"]
time_df = sheet["time"]

df_list = [cust_df, time_df]
cls_list =[CustomerList, TimeList]
name_list =["Customer", "Time"]

error_sheet, error_df = test_all(df_list, cls_list, name_list)

print(error_sheet)
print(error_df)
['## Error in Customer Data', '## Error in Time Data']
[   row   col                       message
0    0  name  none is not an allowed value
1    0   lat  none is not an allowed value,    row   col                       message
0    0  time  value is not a valid integer]

REST API用のモデルをもとにWeiszfeld法を適用し,結果の辞書を返す関数 solve_weiszfeld

これはサーバー側で呼ばれ,結果はJSONに変換されて返される.

solve_weiszfeld[source]

solve_weiszfeld(model)

モデルをもとにWeiszfeld法を適用し,結果の辞書を返す関数

solve_weiszfeld関数の使用例

データを入れた辞書 test をMelosGfクラスのコンストラクタにわたすと,モデルのインスタンスが生成される. モデルをsolve_weizfeldに渡すと求解され,解の情報を保存した辞書が返される.

sheet = pd.read_excel("melos-gf.xlsx", sheet_name=None) 
cust_df = sheet["melos-gf"]
time_df = sheet["time"]

test ={"num_facilities":5, "seed":3, "customers":cust_df.to_dict("records"), "times":time_df.to_dict("records")}
try:
    model = MelosGf(**test)  # データ検証をこれで行う! Exceptionがあれば,それを表示する.
except ValidationError as e:
    print(e)
ret = solve_weiszfeld(model)
print(ret)
{'X': [34.65360063314167, 32.90941699070985, 37.75022939546075, 35.413513788771525, 34.68623309930219], 'Y': [133.92046323477393, 130.63627299431, 140.46771832282408, 136.90252457051528, 135.5198043043494], 'partition': {30: 0, 31: 0, 32: 0, 33: 0, 35: 0, 36: 0, 37: 0, 38: 0, 34: 1, 39: 1, 40: 1, 41: 1, 42: 1, 43: 1, 44: 1, 45: 1, 46: 1, 0: 2, 1: 2, 2: 2, 3: 2, 4: 2, 5: 2, 6: 2, 7: 2, 8: 2, 9: 2, 10: 2, 11: 2, 12: 2, 14: 2, 13: 3, 15: 3, 16: 3, 17: 3, 18: 3, 19: 3, 20: 3, 21: 3, 22: 3, 23: 3, 24: 4, 25: 4, 26: 4, 27: 4, 28: 4, 29: 4}, 'cost': 49104387.09080069}

fastAPIでの求解

appはコード内で定義されたFastAPIのインスタンスである.

@app.post("/melosgf/")
def solve_melos_gf(model: MelosGf):
    return solve_weiszfeld(model)

呼び出し例

データを入れた辞書を,hson.dumps関数でJSONファイルに変換してサーバー(以下では,ローカルサーバー http://127.0.0.1:8000 を仮定)に渡すと,サーバー側でモデルインスタンスに変換される. それをもとに,最適化計算を行い,結果を返す.結果はレスポンスオブジェクトなので,text属性やjson()関数で中身を得る.

URL = "http://127.0.0.1:8000/melosgf"
ret = requests.post(URL, data=json.dumps(test))
ret.text

MELOS

モデル Melos

予測 forecast とSCBASも兼用

class Product[source]

Product(name:str, average_demand:float=None, weight:float=None, volume:float=None, cust_value:float=None, dc_value:float=None, plnt_value:float=None, fixed_cost:float=None, inv_cost:float=None, safety_inventory:float=None, initial_inventory:float=None, target_inventory:float=None) :: BaseModel

class ProductList[source]

ProductList(field_list:List[Product]=[]) :: BaseModel

class TotalDemand[source]

TotalDemand(cust:str, prod:str, demand:float) :: BaseModel

class Demand[source]

Demand(date:date, cust:str, prod:str, demand:float, sales:float=None, promo_0:int=None, promo_1:int=None) :: BaseModel

class DemandList[source]

DemandList(field_list:List[Demand]=[]) :: BaseModel

class Promotion[source]

Promotion(date:date, promo_0:int=None, promo_1:int=None) :: BaseModel

class PromotionList[source]

PromotionList(field_list:List[Promotion]=[]) :: BaseModel

class Dc[source]

Dc(name:str, lat:float, lon:float, lb:float, ub:float, fc:float, vc:float) :: BaseModel

class DcList[source]

DcList(field_list:List[Dc]=[]) :: BaseModel

class Plant[source]

Plant(name:str, lat:float, lon:float, lb:float=0.0, ub:float=None) :: BaseModel

class PlantList[source]

PlantList(field_list:List[Plant]=[]) :: BaseModel

class PlantProduct[source]

PlantProduct(prod:str, plnt:str, ub:float, lead_time:int) :: BaseModel

class PlantProductList[source]

PlantProductList(field_list:List[PlantProduct]=[]) :: BaseModel

class Trans[source]

Trans(from_node:str, to_node:str, dist:int, time:int, cost:float, kind:str) :: BaseModel

class Melos[source]

Melos(dc_lb:int=None, dc_ub:int=None, single_sourcing:bool, max_cpu:int, customers:List[Customer]=[], dcs:List[Dc]=[], plants:List[Plant]=[], products:List[Product]=[], plnt_prod:List[PlantProduct]=[], total_demand:List[TotalDemand]=[], trans:List[Trans]=[]) :: BaseModel

MELOS モデル

データフレーム検証

sheet = pd.read_excel("melos.xlsx", sheet_name=None) #read all sheets 
sheet_list = ["Cust", "Prod", "demand", "DC", "Plnt", "Plnt-Prod", "time"]
cls_list =[CustomerList, ProductList, DemandList, DcList, PlantList, PlantProductList, TimeList]
for s,c in zip(sheet_list, cls_list):
    ret = valid_test(c, sheet[s])
    if ret is not None:
        print(f"Error in Sheet {s}")
        print(ret)
sheet = pd.read_excel("melos.xlsx", sheet_name=None) #read all sheets 
prod_df = sheet["Prod"] #extract dataframe 
model = ProductList(**{"field_list":prod_df.to_dict("records")})
#print(type(jsonable_encoder(model.field_list))) #JSONで読める形式(この場合はリスト)
#pd.DataFrame(jsonable_encoder(model.field_list)).head()

prod_df = sheet["Cust"]
model = CustomerList(**{"field_list":prod_df.to_dict("records")})
#pd.DataFrame(jsonable_encoder(model.field_list)).head()

demand_df = sheet["demand"]
model = DemandList(**{"field_list":demand_df.to_dict("records")})

MELOSをモデルから解く関数 solve_melos

solve_melos[source]

solve_melos(model:Melos)

MELOSをモデルから解く関数

REST API呼び出し例

cust_df = pd.read_csv(folder + "Cust.csv")
plnt_df = pd.read_csv(folder + "Plnt.csv")
total_demand_df = pd.read_csv(folder + "total_demand.csv")
trans_df = pd.read_csv(folder + "trans_cost.csv")
dc_df = pd.read_csv(folder + "DC.csv")
prod_df = pd.read_csv(folder + "Prod.csv")
plnt_prod_df = pd.read_csv(folder + "Plnt-Prod.csv")

test ={"dc_lb":None, "dc_ub":None,"single_sourcing":True,"max_cpu":10,"customers":cust_df.to_dict("records"), 
       "dcs":dc_df.to_dict("records"),"plants":plnt_df.to_dict("records"),  
       "products":prod_df.to_dict("records"),"plnt_prod": plnt_prod_df.to_dict("records"), "total_demand":total_demand.to_dict("records"),
       "trans":trans_df.to_dict("records")}
model = Melos(**test) 
ret = solve_melos(model)
URL = "http://127.0.0.1:8000/melos"
ret = requests.post(URL, data=json.dumps(test))
ret.text[:100]

OptSeq

モデル OptSeq

class OptSeq[source]

OptSeq(text_model:str, best_act:str=None, max_cpu:int=1, initial_flag:bool=False, seed:int=1) :: BaseModel

OptSeq モデル

データフレーム検証用のクラス

class Activity[source]

Activity(name:str, duedate:Union[str, int], backward:bool, weight:int, autoselect:bool) :: BaseModel

class ActivityList[source]

ActivityList(field_list:List[Activity]=[]) :: BaseModel

class Mode[source]

Mode(name:str, duration:int, breakable:ConstrainedStrValue, parallel:ConstrainedStrValue, state:ConstrainedStrValue) :: BaseModel

class ModeList[source]

ModeList(field_list:List[Mode]=[]) :: BaseModel

class Resource[source]

Resource(name:str, capacity:Union[int, ConstrainedStrValue]) :: BaseModel

class ResourceList[source]

ResourceList(field_list:List[Resource]=[]) :: BaseModel

class ActivityMode[source]

ActivityMode(activity:str, mode:str) :: BaseModel

class ActivityModeList[source]

ActivityModeList(field_list:List[ActivityMode]=[]) :: BaseModel

class ModeResource[source]

ModeResource(mode:str, resource:str, type:str=None, requirement:Union[ConstrainedStrValue, int]) :: BaseModel

class ModeResourceList[source]

ModeResourceList(field_list:List[ModeResource]=[]) :: BaseModel

class Temporary[source]

Temporary(pred:str, succ:str, type:str, delay:int) :: BaseModel

class TemporaryList[source]

TemporaryList(field_list:List[Temporary]=[]) :: BaseModel

class Nonrenewable[source]

Nonrenewable(name:str, rhs:int, direction:str, weight:Union[str, int]) :: BaseModel

class NonrenewableList[source]

NonrenewableList(field_list:List[Nonrenewable]=[]) :: BaseModel

class LeftHandSide[source]

LeftHandSide(res_name:str, term:int, act_name:str, mode_name:str) :: BaseModel

class LeftHandSideList[source]

LeftHandSideList(field_list:List[LeftHandSide]=[]) :: BaseModel

class State[source]

State(state_name:str, time_value:ConstrainedStrValue) :: BaseModel

class StateList[source]

StateList(field_list:List[State]=[]) :: BaseModel

sheet = pd.read_excel("optseq.xlsx", sheet_name=None)
sheet_list = ["act", "mode", "res", "act_mode", "mode_res", "temp", "non_res", "non_lhs", "state"]
cls_list =[ActivityList, ModeList, ResourceList, ActivityModeList, ModeResourceList, TemporaryList, NonrenewableList, LeftHandSideList, StateList]
sheet["mode"].iloc[0,4] = "{4:5}"
for s,c in zip(sheet_list, cls_list):
    ret = valid_test(c, sheet[s])
    if ret is not None:
        print(f"Error in Sheet {s}")
        print(ret)
model = optseq.ex1()
act_df, res_df, mode_df, act_mode_df, mode_res_df, temp_df, non_res_df, non_lhs_df, state_df = optseq.convert(model)
act_df.head()
name duedate backward weight autoselect
0 Act[1] inf False 1 False
1 Act[2] inf False 1 False
2 Act[3] inf False 1 False
3 Act[4] inf False 1 False
4 Act[5] inf False 1 False
ret = valid_test(ActivityList, act_df)
print(ret)
None

モデルから求解する関数 solve_optseq

solve_optseq[source]

solve_optseq(model)

以下をローカルのソルバーと置き換える.すべて実行環境はlinuxとする.

REST API呼び出し例

ex1_model = optseq.ex1()
test = {"text_model": ex1_model.update(), "max_cpu": 1}
model = OptSeq(**test)
solve_optseq(model)
URL = "http://127.0.0.1:8000/optseq"
ret = requests.post(URL, data=json.dumps(test))
ret.json()

SCOP

モデル SCOP

class Scop[source]

Scop(text_model:str, best_sol:str=None, max_cpu:int=1, initial_flag:bool=False, seed:int=1, target:int=0) :: BaseModel

SCOP モデル

モデルから求解する関数 solve_scop

solve_scop[source]

solve_scop(model)

以下をローカルのソルバーと置き換える.すべて実行環境はlinuxとする.

REST API呼び出し例

workers=['A','B','C']
Jobs   =[0,1,2]
Cost={ ('A',0):15, ('A',1):20, ('A',2):30,
       ('B',0): 7, ('B',1):15, ('B',2):12,
       ('C',0):25, ('C',1):10, ('C',2):13 }

m=scop.Model()
x={}
for i in workers:
    x[i]=m.addVariable(name=i,domain=Jobs)

xlist=[]
for i in x:
    xlist.append(x[i])

con1=scop.Alldiff('AD',xlist,weight='inf')

con2=scop.Linear('linear_constraint',weight=1,rhs=0,direction='<=')
for i in workers:
    for j in Jobs:
        con2.addTerms(Cost[i,j],x[i],j)

m.addConstraint(con1)
m.addConstraint(con2)

test = {"text_model": m.update(), "max_cpu": 1}
model = Scop(**test)
solve_scop(model)
URL = "http://127.0.0.1:8000/scop"
ret = requests.post(URL, data=json.dumps(test))
ret.json()

METRO

モデル Metro

class Node[source]

Node(id:int=None, name:str=None, location:Union[NoneType, JsonWrapperValue, Tuple[float, float]]=None) :: BaseModel

class NodeList[source]

NodeList(field_list:List[Node]=[]) :: BaseModel

class Break[source]

Break(id:int=None, time_windows:Union[JsonWrapperValue, Tuple[int, int]], service:int, description:str=None) :: BaseModel

class BreakList[source]

BreakList(field_list:List[Break]=[]) :: BaseModel

class Vehicle[source]

Vehicle(id:int=None, description:str=None, name:str=None, start:Union[NoneType, JsonWrapperValue, JsonWrapperValue, Tuple[float, float], tuple]=None, start_index:int=None, end:Union[NoneType, JsonWrapperValue, JsonWrapperValue, Tuple[float, float], tuple]=None, end_index:int=None, capacity:Union[JsonWrapperValue, List[int]], time_window:Union[JsonWrapperValue, Tuple[int, int]], skills:Union[JsonWrapperValue, List[int]], breaks:Union[JsonWrapperValue, List[Break]]) :: BaseModel

class VehicleList[source]

VehicleList(field_list:List[Vehicle]=[]) :: BaseModel

class Job[source]

Job(id:int=None, description:str=None, location:Union[NoneType, JsonWrapperValue, Tuple[float, float]]=None, location_index:int=None, service:int, delivery:Union[JsonWrapperValue, List[int]], pickup:Union[JsonWrapperValue, List[int]], skills:Union[JsonWrapperValue, List[int]], priority:int, time_windows:Union[JsonWrapperValue, List[Tuple[int, int]]]) :: BaseModel

class JobList[source]

JobList(field_list:List[Job]=[]) :: BaseModel

class ShipmentStep[source]

ShipmentStep(id:int=None, description:str=None, location:Union[NoneType, JsonWrapperValue, Tuple[float, float]]=None, location_index:int=None, service:int, time_windows:Union[JsonWrapperValue, List[Tuple[int, int]]]) :: BaseModel

class Shipment[source]

Shipment(pickup:ShipmentStep=None, delivery:ShipmentStep=None, pickup_point:str, pickup_service:int, pickup_time_windows:Union[JsonWrapperValue, List[Tuple[int, int]]], pickup_location:Union[NoneType, JsonWrapperValue, Tuple[float, float]]=None, pickup_index:int=None, delivery_point:str, delivery_service:int, delivery_time_windows:Union[JsonWrapperValue, List[Tuple[int, int]]], delivery_location:Union[NoneType, JsonWrapperValue, Tuple[float, float]]=None, delivery_index:int=None, amount:Union[JsonWrapperValue, List[int]], skills:Union[JsonWrapperValue, List[int]], priority:int) :: BaseModel

class ShipmentList[source]

ShipmentList(field_list:List[Shipment]=[]) :: BaseModel

class Metro[source]

Metro(vehicles:List[Vehicle], jobs:List[Job]=None, shipments:List[Shipment]=None, breaks:List[Break]=[], matrix:List[List[int]]=[]) :: BaseModel

データフレーム検証用のクラス

正規表現を用いる.

データフレーム検証

sheet = pd.read_excel("metro.xlsx", sheet_name=None) #read all sheets 

sheet_list = ["break", "vehicle", "job", "shipment", "node", "time"]
cls_list =[BreakList, VehicleList, JobList, ShipmentList, NodeList, TimeList]
for s,c in zip(sheet_list, cls_list):
    ret = valid_test(c, sheet[s])
    if ret is not None:
        print(f"Error in Sheet {s}")
        print(ret)

モデルから求解する関数 solve_metro

solve_metro[source]

solve_metro(model, matrix=False, threads=4, explore=1)

REST API呼び出し例

no ="06"
node_df = pd.read_csv(folder +"metroIV/node"+no+".csv")
job_df = pd.read_csv(folder +"metroIV/job"+no+".csv") 
shipment_df = pd.read_csv(folder +"metroIV/shipment"+no+".csv") 
vehicle_df = pd.read_csv(folder +"metroIV/vehicle"+no+".csv") 
time_df = pd.read_csv(folder+"metroIV/time"+no+".csv", index_col=0)
break_df = pd.read_csv(folder +"metroIV/break.csv", index_col=0)
model = metro.build_model_for_vrp(job_df, shipment_df, vehicle_df, break_df, time_df)

model = Metro(**model) 
#ret = solve_metro(model)
#jsonable_encoder(model.shipments[0])
URL = "http://127.0.0.1:8000/metro"
ret = requests.post(URL, data=json.dumps(model)) 
ret.text[:100]

MESSA

モデル Messa

class Stage[source]

Stage(name:str, net_replenishment_time:int, max_guaranteed_LT:int, processing_time:int, replenishment_LT:int, guaranteed_LT:int, z:float, average_demand:float, sigma:float, h:float, b:float, x:float=None, y:float=None, capacity:float) :: BaseModel

class StageList[source]

StageList(field_list:List[Stage]=[]) :: BaseModel

class Bom[source]

Bom(child:str, parent:str, units:float, allocation:float=None) :: BaseModel

class BomList[source]

BomList(field_list:List[Bom]=[]) :: BaseModel

class Messa[source]

Messa(stages:List[Stage], boms:List[Bom]) :: BaseModel

データフレーム検証

sheet = pd.read_excel("messa.xlsx", sheet_name=None) #read all sheets 
sheet_list = ["stage", "bom"]
cls_list =[StageList, BomList]
for s,c in zip(sheet_list, cls_list):
    ret = valid_test(c, sheet[s])
    if ret is not None:
        print(f"Error in Sheet {s}")
        print(ret)

モデルから求解する関数 solve_messa

solve_messa[source]

solve_messa(model:Messa)

SSAをモデルから解く関数

REST API呼び出し例

folder_bom = "../data/bom/"
stage_df = pd.read_csv(folder_bom + "ssa03.csv")
bom_df = pd.read_csv(folder_bom + "ssa_bom03.csv")
#best_cost, stage_df, bom_df, fig = optinv.solve_SSA(stage_df, bom_df)
test = {"stages": stage_df.to_dict("records"), "boms": bom_df.to_dict("records")}
model = Messa(**test)
URL = "http://127.0.0.1:8000/messa"
ret = requests.post(URL, data=json.dumps(test)) 
#jsonable_encoder(model.boms[0])

OptInv

モデルはMessaと同じ

モデルから求解する関数 solve_optinv

solve_optinv[source]

solve_optinv(model:Messa)

periodic_inv_optをモデルから解く関数

REST API呼び出し例

URL = "http://127.0.0.1:8000/optinv"
ret = requests.post(URL, data=json.dumps(test)) 

SENDO

モデル Sendo

class Sendo[source]

Sendo(cost_per_dis:float=10, cost_per_time:float=5000, capacity:int=10000, max_cpu:int=60, scaling:bool=False, k:int=10, alpha:float=0.5, max_iter:int=10, osrm:bool=False, dcs:List[Dc], demand:List[List[float]]) :: BaseModel

データフレーム検証

sheet = pd.read_excel("sendo.xlsx", sheet_name=None) #read all sheets 
dc_df = sheet["DC"]
od_df = sheet["od"]
ret = valid_test(DcList, dc_df)
print(ret)
None
#name2 = tuple(od_df.name)
#assert name1 == name2
#長さのみ検証
n,_ = dc_df.shape
rows, cols = od_df.shape
if rows==n and cols-1 == n:
    print("OK")
OK

モデルから求解する関数 solve_sendo

solve_sendo[source]

solve_sendo(model:Sendo)

Solve SENDO using Sendo model

REST API呼び出し例

dc_df = pd.read_csv(folder + "DC.csv", index_col=0) #ub =base capacity, vc = transfer cost
od_df = pd.read_csv(folder + "od.csv", index_col=0)

n= 10
dc_df = dc_df.iloc[:n,:]
od_df = od_df.iloc[:n,:n]

#リストのリストに変換
L =[]
for row in od_df.values:
    L.append( list(row) )
test ={
    "cost_per_dis":20, "cost_per_time":8000, "capacity":1000., "max_cpu":10, "scaling":False, "k":10, "alpha":0.5, "max_iter":10, "osrm": False,
    "dcs": dc_df.to_dict("records"), "demand":L}
#model = Sendo(**test)
URL = "http://127.0.0.1:8000/sendo"
ret = requests.post(URL, data=json.dumps(test)) 

OptShift

モデル OptShift

class Period[source]

Period(id:int, description:str) :: BaseModel

class PeriodList[source]

PeriodList(field_list:List[Period]=[]) :: BaseModel

class ShiftBreak[source]

ShiftBreak(period:int, break_time:int) :: BaseModel

class ShiftBreakList[source]

ShiftBreakList(field_list:List[ShiftBreak]=[]) :: BaseModel

class Day[source]

Day(id:int, day:date, day_of_week:str, day_type:str) :: BaseModel

class DayList[source]

DayList(field_list:List[Day]=[]) :: BaseModel

class ShiftJob[source]

ShiftJob(id:int='ジョブのインデックス(休憩は必ず0)', description:str='ジョブの説明') :: BaseModel

class ShiftJobList[source]

ShiftJobList(field_list:List[ShiftJob]=[]) :: BaseModel

class Staff[source]

Staff(name:str, wage_per_period:int, max_period:int, max_day:int, job_set:str, day_off:str) :: BaseModel

class StaffList[source]

StaffList(field_list:List[Staff]=[]) :: BaseModel

class Requirement[source]

Requirement(day_type:str, job:int, period:int, requirement:int) :: BaseModel

class RequirementList[source]

RequirementList(field_list:List[Requirement]=[]) :: BaseModel

class Parameter[source]

Parameter(theta:int=1, lb_penalty:int=10000, job_change_penalty:int=10, break_penalty:int=10000, max_day_penalty:int=5000, time_limit:int=10, random_seed:int=1) :: BaseModel

class OptShift[source]

OptShift(periods:List[Period], breaks:List[ShiftBreak], days:List[Day], jobs:List[ShiftJob], staffs:List[Staff], requirements:List[Requirement], parameters:Parameter) :: BaseModel

データフレーム検証

sheet = pd.read_excel("optshift.xlsx", sheet_name = None)
sheet_list = ["period", "break", "day", "job", "staff", "requirement"]
cls_list =[PeriodList, ShiftBreakList, DayList, ShiftJobList, StaffList, RequirementList]
for s,c in zip(sheet_list, cls_list):
    ret = valid_test(c, sheet[s])
    if ret is not None:
        print(f"Error in Sheet {s}")
        print(ret)
folder = "../data/shift/"
period_df = pd.read_csv(folder+"period.csv", index_col=0)
break_df = pd.read_csv(folder+"break.csv", index_col=0)
day_df = pd.read_csv(folder+"day.csv", index_col=0)
job_df = pd.read_csv(folder+"job.csv", index_col=0)
staff_df = pd.read_csv(folder+"staff.csv", index_col=0)
requirement_df = pd.read_csv(folder+"requirement.csv", index_col=0)

test ={"periods": period_df.to_dict("records"),
      "breaks": break_df.to_dict("records"),
       "days": day_df.to_dict("records"),
       "jobs": job_df.to_dict("records"),
       "staffs": staff_df.to_dict("records"),
       "requirements": requirement_df.to_dict("records"),
       "parameters": {"random_seed": 2, "time_limit": 1}
      }
model = OptShift(**test)
jsonable_encoder(model.requirements[1])
{'day_type': 'weekday', 'job': 1, 'period': 1, 'requirement': 2}

モデルから求解する関数

solve_optshift[source]

solve_optshift(model)

Solve OPTSHIFT using OptShift model

REST API呼び出し例

URL = "http://127.0.0.1:8000/optshift/"
ret = requests.post(URL, data=json.dumps(test)) 
ret.text[:30]

OptLot

モデル OptLot

class Production[source]

Production(name:str, ProdTime:float, SetupTime:float, ProdCost:float, SetupCost:float) :: BaseModel

class ProductionList[source]

ProductionList(field_list:List[Production]=[]) :: BaseModel

class PlantDemand[source]

PlantDemand(prod:str, period:int, demand:float) :: BaseModel

class PlantDemandList[source]

PlantDemandList(field_list:List[PlantDemand]=[]) :: BaseModel

class LotResource[source]

LotResource(name:str, period:int, capacity:float) :: BaseModel

class LotResourceList[source]

LotResourceList(field_list:List[LotResource]=[]) :: BaseModel

sheet = pd.read_excel("optlot.xlsx", sheet_name =None)
sheet_list = ["lotprod", "production", "bom", "plnt-demand", "resource"]
cls_list =[ProductList, ProductionList, BomList, PlantDemandList, LotResourceList]
for s,c in zip(sheet_list, cls_list):
    ret = valid_test(c, sheet[s])
    if ret is not None:
        print(f"Error in Sheet {s}")
        print(ret)