In [74]:
from dist_time import *
import pandas as pd
from collections.abc import Iterator, Iterable, Generator
from collections import defaultdict, namedtuple
import re
from sqlalchemy import create_engine
from sqlalchemy.types import Integer, Text, DateTime, Float, VARCHAR
from sqlalchemy import Table, MetaData, Column
import itertools as it
import numpy as np
np.random.seed=7
class MySQLConfig:
HOST = '10.0.149.62'
USER = "root"
PASS = "root123"
DB = "cm_erzhou_v1_1"
CHARSET = 'utf8'
engine = create_engine(
f'mysql+pymysql://{USER}:{PASS}@{HOST}/{DB}?charset={CHARSET}',
isolation_level="READ UNCOMMITTED")
CREATE_TABLE_DB_ENG = MetaData(bind=MySQLConfig.engine)
def creat_table(db, table_name, col_list):
machine_table_sche = \
Table(
table_name,
db,
*col_list
)
machine_table_sche.create(checkfirst=True)
def to_mysql(table_name: str, data: pd.DataFrame):
try:
data.to_sql(name=table_name,
con=MySQLConfig.engine,
if_exists='append',
index=0)
print(f"mysql write table {table_name} succeed!")
except Exception as exc:
print(f"mysql write table {table_name} failed, error: {exc}.")
raise Exception
def dict_to_dataframe(pd: pd, *, parameter_list: list=None):
data = defaultdict(list)
for paramer in parameter_list:
data['process_time'].extend(list(paramer.values()))
data ['parameter_name'].extend(list(paramer.keys()))
return pd.DataFrame(data)
def df_to_excel(file_name, sheet, df):
file = f"d:/Git/git.jupyter/jupyter_app/ipython/py36_erzhou_input/dist_time/out/{file_name}.xlsx"
df.to_excel(file, sheet_name=sheet, index=False)
def check_dic_value(data: dict=None):
type_dic_value = set()
if data is None:
return
else:
for _,v in data.items():
if isinstance(v, list):
type_dic_value.add('L')
elif isinstance(v, str):
type_dic_value.add('S')
elif isinstance(v, int):
type_dic_value.add('I')
else:
raise ValueError('字典值类型不是字符或列表形式！')
if len(type_dic_value) !=1:
raise ValueError('字典值类型既有字符也有列表！')
return {type_dic_value.pop(): data}
def dict_list_to_dataframe(pd: pd.DataFrame, parameter_list: list=None, colum_name: list=None):
if parameter_list:
re_list = []
for data in parameter_list:
# print(data[1])
ckeck_data = check_dic_value(data[0])
for k, _ in ckeck_data.items():
if k == 'L':
re_data = it.chain(
*[i for i in map(
lambda x, y : [[x, v, data[1]] for v in y if isinstance(y, list)],
ckeck_data[k].keys(), ckeck_data[k].values())
]
)
re_list.extend(re_data)
# return pd.DataFrame(list(re_data), columns=['equipment_port_last', 'equipment_port_next'])
elif k == 'S':
re_data = [[k, v, data[1]] for k, v in ckeck_data[k].items()]
re_list.extend(re_data)
elif k == 'I':
re_data = [[k, v, data[1]] for k, v in ckeck_data[k].items()]
re_list.extend(re_data)
return pd.DataFrame(re_list, columns=colum_name)
else:
return parameter_list
def double_list_to_list_tuple(dic_data: dict=None):
list_tuple = []
for k, v in dic_data.items():
lst = list(v)
if isinstance(lst[0], list):
if isinstance(lst[1], list):
list_tuple.extend([(k, x, y) for x, y in it.zip_longest(lst[0], lst[1])])
elif isinstance(lst[1], str):
list_tuple.extend([(k, x, y) for x, y in [(e, lst[1]) for e in lst[0]]])
else:
list_tuple.extend([(k, x, y) for x, y in [(e, 'equal') for e in lst]])
# raise ValueError(f'权值数据异常，请核实：{k} 的权值！')
return list_tuple
def dict_double_list_to_dataframe(pd: pd.DataFrame, parameter_list: list=None, colum_name: list=None):
if parameter_list:
re_list = []
for data in parameter_list:
ckeck_data = check_dic_value(data[0])
for k, _ in ckeck_data.items():
if k == 'L':
weight_data = double_list_to_list_tuple(ckeck_data[k])
re_data = [(x, y, z, data[1]) for x, y , z in weight_data]
re_list.extend(re_data)
elif k == 'S':
re_data = [[k, v, data[1]] for k, v in ckeck_data[k].items()]
re_list.extend(re_data)
elif k == 'I':
re_data = [[k, v, data[1]] for k, v in ckeck_data[k].items()]
re_list.extend(re_data)
return pd.DataFrame(re_list,
columns=colum_name
)
else:
return parameter_list
In [2]:
PARAM_DICT = {
'ULD_Truck_Capacity':[
(R_Reload_Truck_Capacity, 'truck_capacity'), # 卡车容量配置表
(uld_capacity, 'uld_capacity'), # uld容量配置表
],
'City_Airport_Code': [
(city_apt, 'city_airport_code'), # 城市编码-机场编码表
],
'S_Road_City2Apt': [
(S_Road_city2apt, 's_road_city2apt'), #
],
'Flight_ID_Unload_des': [
(flight_unload, 'flight_unload'), # 飞机卸货区字典
],
'Secondary_Reload_Sort': [
(R_Reload_R_Secondary, 'secondary_reload'), # Secondary_Reload_Sort- R
(D_Reload_D_Secondary, 'secondary_reload'), # Secondary_Reload_Sort- D
(I_Reload_I_Secondary, 'secondary_reload'), # Secondary_Reload_Sort- I
],
'S_Secondary_Reload_Sort': [
(S_Secondary_apt, 'small_secondary_reload'), # small_secondary_reload sort- S
],
'Reload_Setting': [
(R_Secondary_rule, 'reload'), # reload- R
(D_Secondary_rule, 'reload'), # reload- D
(I_Secondary_rule, 'reload'), # reload- I
],
'S_Reload_Setting': [
(S_Secondary_rule, 'small_reload'), # reload- S
],
'apt_reload': [
(apt_reload, 'reload_ot'), # 目的地编码与装货口对应表
]
}
# ---------------------------------------------excel db insert methods ----------------------------------------------------------
# df_to_excel(file_name='', sheet='', df=)
# creat_table(db=CREATE_TABLE_DB_ENG, table_name='', col_list=)
# to_mysql(table_name='', data=)
In [62]:
apt_reload_colum_name = ['dest_zone_code', 'equipment_port', 'allocate_weight', 'reload_type']
apt_reload_df = dict_double_list_to_dataframe(
pd, parameter_list=PARAM_DICT['apt_reload'], colum_name=apt_reload_colum_name)
reload_num = np.random.choice(apt_reload_df[apt_reload_df.dest_zone_code == '870'].equipment_port.tolist())
reload_num
# p = re.compile(r"Reload\d+")
# p.findall(reload_num)
Out[62]:
In [109]:
apt_col = ['dest_zone_code', 'equipment_port', 'allocate_weight']
apt = namedtuple('apt', apt_col)
apt_df = pd.read_sql_table(
con=MySQLConfig.engine,
table_name='i_apt_reload',
columns=apt_col)
apt_df_sample = apt_df.query('dest_zone_code == "7112l"')
apt_values = map(apt._make, apt_df_sample.values)
next(apt_values)
Out[109]: