SCMOPT REST API & Validiation
Pydanticによるデータフレーム検証
Excelファイルやcsvファイルを読み込んだ際に，データの妥当性の検証を行う必要がある． そのために，fastAPI (https://fastapi.tiangolo.com/) に含まれるPydantic (https://pydantic-docs.helpmanual.io/) を利用する．
Pydanticは，オブジェクトタイプが間違えていても，できるだけ自動変換する．
example = {
"name": "響本店",
"lat": "43.06417", #文字列はfloatに変換される
"lon": 141.34694,
"weight": 1 #整数もfloatに変換される
}
cust = Customer(**example) #Pydantic BaseClassは辞書から生成できる．（**で展開）
print("文字列:", cust) #インスタンス文字列
print("JSON:", cust.json()) #JSON文字列 https://www.json.org/json-ja.html
print("スキーマ:", cust.schema()) #JSONスキーマ
print("辞書:", cust.dict()) #辞書に変換
print("JSONで読める形式:", jsonable_encoder(cust)) #JSONで読める形式（この場合は辞書） https://fastapi.tiangolo.com/tutorial/encoder/
example2 = {
"name": "響本店",
"lat": "43.06417"
}
try:
cust2 = Customer(**example2)
except ValidationError as e:
print(e.json())
cust_json = cust.json()
cust_dict = json.loads(cust_json)
new_customer = parse_obj_as(Customer, cust_dict) #以下でも同じ
#Customer(**cust_dict)
new_customer
cust_df, time_df = read_xl("melos-gf.xlsx")
#read_xl("error.xlsx") #FileNotFoundErrorが発生
cust_df.head()
#time_df.head()
customer_list, time_list = make_obj_list([cust_df, time_df], [Customer, Time])
print(customer_list[0])
print(time_list[10])
cust_dic = {} #customer dic.
for c in customer_list:
cust_dic[c.name] = c
print("Pop from customer dic.=", cust_dic.popitem() )
time_dic = {} #time dic.
for t in time_list:
time_dic[t.from_name, t.to_name] = t.time
print("Pop from time dic.=", time_dic.popitem() )
from pprint import pprint
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() #データフレームに戻す
cust_df.iloc[0,1] = None
cust_df.iloc[0,2] = np.nan
cust_df.iloc[0,3] = "１．２３４５" #大文字だとエラー
time_df.iloc[0,5] = "１．２３４５"
ret = valid_test(CustomerList, cust_df)
print(ret)
ret = valid_test(TimeList, time_df)
print(ret)
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[0])
print(error_df[0])
print(error_sheet[1])
print(error_df[1])
sheet = pd.read_excel("melos-gf.xlsx", engine="openpyxl", 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) #初期値X0,Y0を入れても良い
print(ret)
呼び出し例
データを入れた辞書を，json.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
instance_pool = InstancePool()
instance_pool.add(model)
instance_pool.to_pickle("an_instance_pool")
a_solution = Solution(name="melos_gf_sol", data = ret, objval=ret["cost"])
#a_solution.to_pickle()
#a_solution.from_pickle()
a_solution
solution_pool = SolutionPool()
solution_pool.add(a_solution)
solution_pool
cust_df, prod_df, demand_df, dc_df, plnt_df, plnt_prod_df, time_df = read_xl("melos.xlsx")
sheet_list = ["Cust", "Prod", "demand", "DC", "Plnt", "Plnt-Prod", "time"]
df_list = [cust_df, prod_df, demand_df, dc_df, plnt_df, plnt_prod_df, time_df]
cls_list =[CustomerList, ProductList, DemandList, DcList, PlantList, PlantProductList, TimeList]
for s,df,c in zip(sheet_list, df_list, cls_list):
ret = valid_test(c, df)
if ret is not None:
print(f"Error in Sheet {s}")
print(ret)
# 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()
model = CustomerList(**{"field_list":cust_df.to_dict("records")})
#pd.DataFrame(jsonable_encoder(model.field_list)).head()
model = DemandList(**{"field_list":demand_df.to_dict("records")})
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]
sheet = pd.read_excel("optseq.xlsx", engine="openpyxl", 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()
ret = valid_test(ActivityList, act_df)
print(ret)
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()
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()
sheet = pd.read_excel("metro.xlsx", engine="openpyxl", 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)
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) #JSONのもとになる辞書を返す
#model = Metro(**model)
#ret = solve_metro(model)
#jsonable_encoder(model.shipments[0])
#Vehicle(**model["vehicles"][0])
#model["shipments"][0]
URL = "http://127.0.0.1:8000/metro"
ret = requests.post(URL, data=json.dumps(model))
ret.text[:100]
sheet = pd.read_excel("messa.xlsx", engine="openpyxl", 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)
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])
URL = "http://127.0.0.1:8000/optinv"
ret = requests.post(URL, data=json.dumps(test))
sheet = pd.read_excel("sendo.xlsx", engine="openpyxl", sheet_name=None) #read all sheets
dc_df = sheet["DC"]
od_df = sheet["od"]
ret = valid_test(DcList, dc_df)
print(ret)
#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")
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))
sheet = pd.read_excel("optshift.xlsx", engine="openpyxl", 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])
URL = "http://127.0.0.1:8000/optshift/"
ret = requests.post(URL, data=json.dumps(test))
ret.text[:30]
sheet = pd.read_excel("optlot.xlsx", engine="openpyxl", 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)
#https://stackoverflow.com/questions/66168517/generate-dynamic-model-using-pydantic 参照
from pydantic import create_model
d = {}
for i in range(3):
d[f"Skill{i}"] = (bool, ...) # (型,...) ...は既定値なしを意味する
Customer = create_model("Customer", **d)
print(Customer.schema())
#fastapiでの利用は？ 漢字より英語？
class Item(BaseModel):
名称: str
在庫費用: float
在庫量下限: float
在庫量上限: float
初期在庫量: float
最終在庫量: float
class ItemList(BaseModel):
field_list: List[Item]= Field(description="品目リスト", default=[])
print(Item.schema())
wb = load_workbook("optlot-master2-ex1.xlsx")
item_df, process_df, resource_df, bom_df, usage_df = lot.read_dfs_from_excel_lot(wb)
cls_list =[ItemList]
sheet_list = [item_df]
for s,c in zip(sheet_list, cls_list):
ret = valid_test(c, s)
if ret is not None:
print(f"Error in Sheet {s}")
print(ret)
item_df.iloc[0,0] = None
item_df