from wsgiref import headers from dotenv import load_dotenv, find_dotenv from pathlib import Path # Python 3.6+ only import configparser, json, pytz import requests,os import pillow_avif from utils.common import message_queue import hashlib from pathlib import Path import os,stat TIME_ZONE = pytz.timezone("Asia/Shanghai") from numpy import true_divide from databases import ( create_all_database, DeviceConfig, SysConfigs, CRUD, batch_insert_sys_configs, SqlQuery, batch_insert_device_configs, batch_insert_device_configsNew, ) # 追加配置参数 machine_type 拍照机设备类型;0鞋;1服装 MACHINE_TYPE = 0 # 初始化数据表 create_all_database() session = SqlQuery() device_config_crud = CRUD(DeviceConfig) all_devices = device_config_crud.read_all(session) if len(all_devices) == 0: # 如果配置表中一条数据都没有，就将初始化数据全部插入到数据表中 action_tabs = json.load(open("action_tabs.json", encoding="utf-8")) actions = json.load(open("action.json", encoding="utf-8")) batch_insert_device_configs(session, action_tabs, actions) sys_config_crud = CRUD(SysConfigs) all_sys_configs = sys_config_crud.read_all(session) if len(all_sys_configs) == 0: # 如果配置表中一条数据都没有，就将初始化数据全部插入到数据表中 sys_config_json = json.load(open("sys_configs.json", encoding="utf-8")) batch_insert_sys_configs(session, sys_config_json) else: # 查询数据库中缺失哪个数据，进行补充 sys_config_json = json.load(open("sys_configs.json", encoding="utf-8")) for sys_config in sys_config_json: photos = CRUD(SysConfigs) item = photos.read(session, conditions={"key": sys_config.get("key")}) if item == None: config = SysConfigs(**sys_config) session.add(config) session.commit() # 合并事务提交 session.close() # 初始化数据表---结束 def get_config_by_items(config_dict): __config_dict = {} for i, k in config_dict: __config_dict[i] = k return __config_dict keys = ["面料", "里料"] def getSysConfigs(key, item, default=None): session = SqlQuery() crud = CRUD(SysConfigs) one_item = crud.read(session, conditions={"key": key}) config = json.loads(one_item.value) session.close() if item == "image_out_format": default_format = config.get(item, default) if default_format == "" or default_format == None: return "png" return config.get(item, default) def get_dict_value(_dict, key, default=None): if key in _dict: return _dict[key] else: return default # 鞋设备为1 服装设备为5 MCU_CODE = 1 MACHINE_LEVEL = ( "二档" if getSysConfigs("other_configs", "device_speed", "二档") == "" else getSysConfigs("other_configs", "device_speed", "二档") ) IS_TEST = False IS_MCU = True IS_LIN_SHI_TEST = False PhotographSeconds = float( 0.5 if getSysConfigs("take_photo_configs", "camera_delay", "0.5") == "" else getSysConfigs("take_photo_configs", "camera_delay", "0.5") ) # 拍照停留时间 MAX_PIXIAN_SIZE = 12000000 def moveSpeed(level: str = None): config = { "一档": { "camera_high_motor": { "max_speed": 10000, "up_speed": 800, "down_speed": 700, }, "turntable_steering": { "max_speed": 6000, "up_speed": 500, "down_speed": 400, }, }, "二档": { "camera_high_motor": { "max_speed": 7000, "up_speed": 600, "down_speed": 500, }, "turntable_steering": { "max_speed": 4500, "up_speed": 350, "down_speed": 300, }, }, "三档": { "camera_high_motor": { "max_speed": 3500, "up_speed": 400, "down_speed": 300, }, "turntable_steering": { "max_speed": 3000, "up_speed": 200, "down_speed": 200, }, }, } if level is None: return config[MACHINE_LEVEL] else: return config[level] config = configparser.ConfigParser() config_name = "config.ini" config.read(config_name, encoding="utf-8") # 应用名称 APP_NAME = config.get("app", "app_name") # 应用版本号 APP_VERSION = config.get("app", "version") # 是否开启调试模式 IS_DEBUG = config.get("app", "debug") IS_UPLOAD_HLM = True if config.get("app", "is_upload") == "true" else False # 应用端口号 PORT = config.get("app", "port") # 应用线程数 APP_WORKS = config.get("app", "works") # 应用host地址 APP_HOST = config.get("app", "host") # 应用服务启动名称 APP_RUN = config.get("app", "app_run") # 日志名称 LOG_FILE_NAME = config.get("log", "log_file_name") # 最大字节数 MAX_BYTES = config.get("log", "max_bytes") print("Max bytes is", MAX_BYTES) # 备份数量 BACKUP_COUNTS = config.get("log", "backup_counts") # 远程服务器地址 HLM_HOST = config.get("log", "hlm_host") PROJECT = config.get("app", "project") # ---------------------------------- mcu_config_dict = config.items("mcu_config") _mcu_config_dict = {} for i, k in mcu_config_dict: _mcu_config_dict[i] = int(k) # print(_mcu_config_dict) _config_mcu_config = get_config_by_items(config.items("mcu_config")) LEFT_FOOT_ACTION = _mcu_config_dict["left_foot_action"] LEFT_FOOT_PHOTOGRAPH = _mcu_config_dict["left_foot_photograph"] LEFT_FOOT_ACTION_1 = _mcu_config_dict["left_foot_action_1"] LEFT_FOOT_ACTION_2 = _mcu_config_dict["left_foot_action_2"] RIGHT_FOOT_ACTION = _mcu_config_dict["right_foot_action"] RIGHT_FOOT_PHOTOGRAPH = _mcu_config_dict["right_foot_photograph"] RIGHT_FOOT_ACTION_1 = _mcu_config_dict["right_foot_action_1"] RIGHT_FOOT_ACTION_2 = _mcu_config_dict["right_foot_action_2"] NEXT_STEP = int(get_dict_value(_config_mcu_config, "next_step", 6)) # 下一步 MOVE_UP = _mcu_config_dict["move_up"] MOVE_DOWN = _mcu_config_dict["move_down"] STOP = _mcu_config_dict["stop"] # camera_config_dict = config.items("camera_config") # _camera_config_dict = {} # for i, k in mcu_config_dict: # _camera_config_dict[i] = int(k) # _camera_config_dict = get_config_by_items(camera_config_dict) # LOW_ISO = _camera_config_dict["low_iso"] # HIGH_ISO = _camera_config_dict["high_iso"] DOMAIN = ( "https://dev2.valimart.net" if config.get("app", "env") != "dev" else "https://dev2.pubdata.cn" ) Company = "惠利玛" is_test_plugins = true_divide OUT_PIC_MODE = "." + getSysConfigs("basic_configs", "image_out_format", "png") # ".png" OUT_PIC_SIZE = ( [1600] if getSysConfigs("basic_configs", "main_image_size", [1600]) == "" else getSysConfigs("basic_configs", "main_image_size", [1600]) ) # 主图大小 Mode = getSysConfigs("other_configs", "product_type", "鞋类") # 程序执行类 OUT_PIC_FACTOR = float( 1 if getSysConfigs("basic_configs", "image_sharpening", "1") == "" else getSysConfigs("basic_configs", "image_sharpening", "1") ) # 图片锐化 RESIZE_IMAGE_MODE = 1 GRENERATE_MAIN_PIC_BRIGHTNESS = 254 # 色阶是否调整到位判断 RUNNING_MODE = getSysConfigs("other_configs", "running_mode", "普通模式") DEFAULT_CUTOUT_MODE = getSysConfigs("other_configs", "cutout_mode", "普通抠图") CUTOUT_MODE = ( 0 if getSysConfigs("other_configs", "cutout_mode", "普通抠图") == "普通抠图" else 1 ) import importlib OUT_PIC_QUALITY = "普通" GRENERATE_MAIN_PIC_BRIGHTNESS = int( getSysConfigs("other_configs", "grenerate_main_pic_brightness", 254) ) # 色阶是否调整到位判断 SHADOW_PROCESSING = int( getSysConfigs("other_configs", "shadow_processing", 0) ) # 0表示要直线和曲线，1 表示只要直线 LOWER_Y = int( getSysConfigs("other_configs", "lower_y", 4) ) # 鞋底以下多少距离作为阴影蒙版 CHECK_LOWER_Y = int( getSysConfigs("other_configs", "check_lower_y", 4) ) # 检测亮度区域，倒数第几行 IS_GET_GREEN_MASK = ( True if getSysConfigs("other_configs", "is_get_green_mask", "否") == "是" else False ) # 是否进行绿幕抠图 IMAGE_SAVE_MAX_WORKERS = int( getSysConfigs("other_configs", "image_save_max_workers", 20) ) # 批量保存的线程大小 COLOR_GRADATION_CYCLES = int( getSysConfigs("other_configs", "color_gradation_cycles", 23) ) # 色阶处理循环次数 def recordDataPoint(token=None, page="", uuid=None, data=""): """记录日志""" if token == None: return if page == "": return if uuid == None: return headers = {"Content-Type": "application/json", "authorization": "Bearer " + token} params = { "site": 1, "type": 5, "channel": "aigc-camera", "uuid": uuid, "page": page, "page_url": "/", "describe": {"action": page, "data": data}, "time": 0, } return requests.post( headers=headers, data=json.dumps(params), url=DOMAIN + "/api/record/point" ) async def sendSocketMessage( code=0, msg="", data=None, device_status=2, msg_type="" ): data = { "code": code, "msg": msg, "status": device_status, "data": data, "msg_type": msg_type, } await message_queue.put(data) def calculate_md5(filepath): # 打开文件，以二进制只读模式打开 with open(filepath, "rb") as f: # 创建MD5哈希对象 md5hash = hashlib.md5() # 循环读取文件的内容并更新哈希对象 for chunk in iter(lambda: f.read(4096), b""): md5hash.update(chunk) # 返回MD5哈希的十六进制表示 return md5hash.hexdigest() __output_dir = config.get("output_config", "output_dir") path_obj = Path(os.path.abspath(__output_dir)) OUTPUT_DIR = path_obj.absolute() print("OUTPUT_DIR",__output_dir,OUTPUT_DIR) def handle_remove_readonly(func, path, exc): os.chmod(path, stat.S_IWRITE) func(path)