Feeding_control_system/service/mould_service.py

from datetime import datetime, timedelta
from common.sqlite_handler import SQLiteHandler
from typing import Optional, List
from .api_http_client import api_http_client
from busisness.models import ArtifactInfo, TaskInfo, LoginRequest, LEDInfo
from config.ini_manager import ini_manager


class MouldService:
    """模具服务类，提供模具相关的API调用"""
    
    def __init__(self):
        """初始化模具服务"""
        self._api_client = api_http_client
        self._host = ini_manager.api_base_url
       
    def get_task_info(self, task_id: str) -> Optional[TaskInfo]:
        """
        获取任务单信息
        
        Args:
            task_id: 任务单编号
        
        Returns:
            任务单信息对象，如果失败返回None
        """
        url = f"{self._host}/api/ext/artifact/task?TaskId={task_id}"
        
        try:
            # 调用API获取数据
            response_data = self._api_client.get(url, auth=True)
            
            # 检查响应状态
            if response_data.get('Code') != 200:
                print(f"获取任务单信息失败: {response_data.get('Message')}")
                return None
            
            # 解析数据
            data = response_data.get('Data', {})
            if not data:
                print(f"未获取到任务单 {task_id} 的信息")
                return None
            
            # 转换为任务单信息对象
            task_info = TaskInfo(**data)
            return task_info
            
        except Exception as e:
            print(f"请求任务单信息异常: {e}")
            return None
    
    def get_not_pour_artifacts(self) -> Optional[List[ArtifactInfo]]:
        """
        获取已入模绑定未浇筑的管片信息
        
        Returns:
            未浇筑管片列表，如果失败返回None
        """
        url = f"{self._host}/api/ext/mould/not_pour_hidden"
        
        try:
            # 调用API获取数据
            response_data = self._api_client.get(url, auth=True)
            
            # 检查响应状态
            if response_data.get('Code') != 200:
                print(f"获取未浇筑管片信息失败: {response_data.get('Message')}")
                return None
            
            # 解析数据
            data_list = response_data.get('Data', [])
            if not data_list:
                print("当前没有未浇筑的管片")
                return []
            
            # 转换为管片信息对象列表
            artifacts = [ArtifactInfo(**item) for item in data_list]
            return artifacts
            
        except Exception as e:
            print(f"请求未浇筑管片信息异常: {e}")
            return None

    def get_pouring_led(self) -> Optional[LEDInfo]:
        """
        获取生产动态信息
        
        Returns:
            LEDInfo对象，如果失败返回None
        """
        url = f"{self._host}/api/ext/produce/pouring_led"
        
        try:
            # 调用API获取数据
            response_data = self._api_client.get(url, auth=True)
            
            # 检查响应状态
            if response_data.get('Code') != 200:
                print(f"获取生产动态信息失败: {response_data.get('Message')}")
                return None
            
            # 解析数据
            data = response_data.get('Data', {})
            if not data:
                print(f"未获取到 pouring_led 信息")
                return None
            
            # 转换为管片信息对象列表 - 使用更安全的字段过滤方式
            # 只提取 LEDInfo 类中定义的字段，忽略多余字段和缺失字段
            led_info = LEDInfo(
                TaskID=data.get('TaskID', ''),
                PlateVolume=data.get('PlateVolume', ''),
                MouldCode=data.get('MouldCode', ''),
                ProduceStartTime=data.get('ProduceStartTime', ''),
                ArtifactID=data.get('ArtifactID', ''),
                RingTypeCode=data.get('RingTypeCode', ''),
                PlateIDSerial=data.get('PlateIDSerial', ''),
                CheckResult=data.get('CheckResult', ''),
                UpperWeight=data.get('UpperWeight', 0.0),
                Temper=data.get('Temper', ''),
                WorkshopTemperature=data.get('WorkshopTemperature', ''),
                LowBucketWeighingValue=data.get('LowBucketWeighingValue', ''),
                VibrationFrequency=data.get('VibrationFrequency', ''),
                TotMete=data.get('TotMete', ''),
                BetonVolumeAlready=data.get('BetonVolumeAlready', ''),
                WaterTemperature=data.get('WaterTemperature', ''),
                FormulaProportion=data.get('FormulaProportion', ''),
                DayStrengthValue=data.get('DayStrengthValue', ''),
                NihtStrengthValue=data.get('NihtStrengthValue', '')
            )
            return led_info
            
        except Exception as e:
            print(f"请求 pouring_led 信息异常: {e}")
            return None
    
app_web_service = MouldService()

# if __name__ == "__main__":
#     # 创建模具服务实例
#     mould_service = MouldService()
#     led_info = mould_service.get_pouring_led()
#     if led_info:
#         print(led_info)
    
    # db = SQLiteHandler.get_instance("db/three.db", max_readers=50, busy_timeout=4000)
    # 测试获取未浇筑管片信息
    # not_poured = mould_service.get_not_pour_artifacts()
    # if not_poured:
        # for item in not_poured:
        #     artifact = db.fetch_one("SELECT * FROM ArtifactTask WHERE ArtifactID = ?", (item.ArtifactID,))
        #     if not artifact:
        #         dict={
        #             "ArtifactID": item.ArtifactID,
        #             "ArtifactActionID": item.ArtifactActionID,
        #             "ArtifactIDVice1": item.ArtifactIDVice1,
        #             "ProduceRingNumber": item.ProduceRingNumber,
        #             "MouldCode": item.MouldCode,
        #             "SkeletonID": item.SkeletonID,
        #             "RingTypeCode": item.RingTypeCode,
        #             "SizeSpecification": item.SizeSpecification,
        #             "BuriedDepth": item.BuriedDepth,
        #             "BlockNumber": item.BlockNumber,
        #             "BetonVolume": item.BetonVolume,
        #             "BetonTaskID": item.BetonTaskID,
        #             "HoleRingMarking": item.HoleRingMarking,
        #             "GroutingPipeMarking": item.GroutingPipeMarking,
        #             "PolypropyleneFiberMarking": item.PolypropyleneFiberMarking,
        #             "Status": 1,
        #             "Source": 1
        #         }
        #         db.insert("ArtifactTask", dict)

        #         dict={
        #             "TaskID": item.BetonTaskID,
        #             "ProjectName": "上海市轨道交通19号线工程盾构区间管片生产2标",
        #             "ProduceMixID": "20251030-02",
        #             "VinNo": "",
        #             "BetonVolume": item.BetonVolume,
        #             "MouldCode": item.MouldCode,
        #             "SkeletonID": item.SkeletonID,
        #             "RingTypeCode": item.RingTypeCode,                    
        #             "SizeSpecification": item.SizeSpecification,
        #             "BuriedDepth": item.BuriedDepth,
        #             "BlockNumber": item.BlockNumber,
        #             "Mode": 1,
        #             "Status": 1,
        #             "Source": 1,
        #             "OptTime": str(datetime.now() - timedelta(minutes=5)),
        #             "CreateTime": str(datetime.now())
        #         }
        #         db.insert("PDRecord", dict)
    # for i in range(2, 5):
    #     row = db.fetch_one("SELECT * FROM ArtifactTask WHERE ID = ?", (i,))
    #     if row:
    #          dict={
    #                 "TaskID": row["BetonTaskID"],
    #                 "ProjectName": "上海市轨道交通19号线工程盾构区间管片生产2标",
    #                 "ProduceMixID": "20251030-02",
    #                 "VinNo": "",
    #                 "BetonVolume": row["BetonVolume"],
    #                 "MouldCode": row["MouldCode"],
    #                 "SkeletonID": row["SkeletonID"],
    #                 "RingTypeCode": row["RingTypeCode"],                    
    #                 "SizeSpecification": row["SizeSpecification"],
    #                 "BuriedDepth": row["BuriedDepth"],
    #                 "BlockNumber": row["BlockNumber"],
    #                 "Mode": 1,
    #                 "Status": 1,
    #                 "Source": 1,
    #                 "OptTime": str(datetime.now() - timedelta(minutes=5)),
    #                 "CreateTime": str(datetime.now())
    #             }
    #          db.insert("PDRecord", dict)