CameraMachine/python/settings.py

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,
)
USER_TOKEN = None  # 可以通过外部设置或从配置文件中加载
USER_ENV = None  # 可以通过外部设置或从配置文件中加载
# 追加配置参数 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 updateSysConfigs(params):
    session = SqlQuery()
    sysConfig = CRUD(SysConfigs)
    # 如果value是字典或列表，需要转换为JSON字符串
    if "value" in params and isinstance(params["value"], (dict, list)):
        params["value"] = json.dumps(params["value"], ensure_ascii=False)
    # 走编辑逻辑
    # kws = params.__dict__
    sysConfig.updateConditions(session, conditions={"key": params["key"]}, **params)
    session.close()

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"
)
def getDoman(env):
    return "https://dev2.valimart.net" if 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)
)  # 色阶处理循环次数
# grenerate_main_pic_brightness 亮度范围 默认值254 范围 200 -255 步长 1
# opacity 透明度 默认值 0.5 范围 0.5-1 步长 0.1
# IMAGE_MASK_CONFIG = int(
#     getSysConfigs("basic_configs", "image_mask_config", {"mode":0,"opacity":0.5,"grenerate_main_pic_brightness":254})
# )  # 色阶处理循环次数

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"
    )
def syncPhotoRecord(data,action_type=0):
    """同步图片记录"""
    headers = {
        "Authorization": f"Bearer {USER_TOKEN}",
        "content-type": "application/json",
    }
    machine_type = MACHINE_TYPE
    url = getDoman(USER_ENV) + f"/api/ai_image/camera_machine/photo_records_handler"
    postData = {"data":data,"machine_type":machine_type,'action_type':action_type}
    response = requests.post(url=url,data=json.dumps(postData, ensure_ascii=False, default=str), headers=headers)
    print("用户token",response.content)
def checkRecordSyncStatus():
    # check_photo_record_sync_status
    headers = {
        "Authorization": f"Bearer {USER_TOKEN}",
        "content-type": "application/json",
    }
    machine_type = MACHINE_TYPE
    url = getDoman(USER_ENV) + f"/api/ai_image/camera_machine/photo_records_handler?machine_type={machine_type}"
    try:
        response = requests.get(url=url, headers=headers).json()
        return response.get("data").get('status',True)
    except Exception as e:
        return True

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)

try:
    __scan_dir = config.get("scan_config", "scan_dir")
    if __scan_dir == "" or __scan_dir is None:
        SCAN_DIR = None
    else:
        scan_path_obj = Path(os.path.abspath(__scan_dir))
        SCAN_DIR = scan_path_obj.absolute()
        if not os.path.exists(SCAN_DIR):
            # 创建多级目录
            os.makedirs(SCAN_DIR, exist_ok=True)
except:
    SCAN_DIR = None
    print('扫码配置不存在')
print("SCAN_DIR",SCAN_DIR)
CUSTOMER_TEMPLATE_URL = config.get("customer_template", "template_url")


def hex_to_rgb(hex_color):
    """
    将十六进制颜色值转换为RGB颜色值
    :param hex_color: 十六进制颜色值，例如 "#FF0000" 或 "FF0000"
    :return: RGB元组，例如 (255, 0, 0)
    """
    # 移除可能存在的 # 符号
    hex_color = hex_color.lstrip('#')
    
    # 确保是有效的十六进制字符串
    if len(hex_color) != 6:
        raise ValueError("十六进制颜色值格式不正确，应为6位十六进制数")
    
    # 将十六进制字符串转换为RGB值
    try:
        r = int(hex_color[0:2], 16)
        g = int(hex_color[2:4], 16)
        b = int(hex_color[4:6], 16)
        return (r, g, b)
    except ValueError:
        return (255, 255, 255)


def rgb_to_hex(r, g, b):
    """
    将RGB颜色值转换为十六进制颜色值
    :param r: 红色分量 (0-255)
    :param g: 绿色分量 (0-255)
    :param b: 蓝色分量 (0-255)
    :return: 十六进制颜色值，例如 "#FF0000"
    """
    # 验证RGB值在有效范围内
    if not all(0 <= val <= 255 for val in [r, g, b]):
        return "#FFFFFF"
    
    return "#{:02X}{:02X}{:02X}".format(r, g, b)