Flask转发 [Siemens.Sistar.Api.dll] Braumat API---＞DLL to restAPI

import os
import clr  # 导入 pythonnet 提供的 clr 模块
from typing import Anyimport func_timeout
from func_timeout import func_set_timeout
from func_timeout.exceptions import FunctionTimedOut# 加载 DLL 文件
current_path = os.getcwd()
dll_path = os.path.join(current_path, "Siemens.Sistar.Api.dll")  # DLL 文件的路径
clr.AddReference(dll_path)# 导入 C# 命名空间中的类
from Siemens.Sistar.Api import SistarBatch# 实例化类
sistar_batch_instance = SistarBatch('python_web_api_batch')
sistar_batch_params_instance = SistarBatch.ParameterList()from System import UInt16, UInt32, Double, Boolean, DateTime, String, Int32BatchStartMode = {'Immediate': SistarBatch.BatchStartMode.Immediate,'StartTime': SistarBatch.BatchStartMode.StartTime,'ByEvent': SistarBatch.BatchStartMode.ByEvent,'StartTimeAuto': SistarBatch.BatchStartMode.StartTimeAuto
}BatchStatus = {'Locked': SistarBatch.BatchStatus.Locked,'ReadyForRelease': SistarBatch.BatchStatus.ReadyForRelease,'Released': SistarBatch.BatchStatus.Released
}@func_set_timeout(timeout=5)
def create_batch(site: int,area: int,year: int,order: int,batch: int,recipeCategory: int,recipe: int,line: int,productId: int,size: float,useDefaultSize: bool,startMode: str,plannedStartTimeYear: int,plannedStartTimeMonth: int,plannedStartTimeDay: int,plannedStartTimeHour: int,plannedStartTimeMin: int,plannedStartTimeSec: int,status: str,parameterList: Any,useDefaultParameterValues: bool,errorMessage: str,doRepeat: bool):"""创建工单:param site: 1:param area: 1:param year: 2024:param order: 1:param batch:115:param recipeCategory:1:param recipe:9:param line:1:param productId:0:param size:100:param useDefaultSize:True:param startMode: 'Immediate':param plannedStartTimeYear: 2024:param plannedStartTimeMonth: 12:param plannedStartTimeDay: 1:param status: 'ReadyForRelease':param parameterList:  sistar_batch_params_instance:param useDefaultParameterValues: True:param errorMessage: '':param doRepeat: False:return:"""error_message = errorMessageresult = sistar_batch_instance.CreateBatch(UInt16(site), UInt16(area), UInt16(year), UInt32(order), UInt32(batch),UInt32(recipeCategory),UInt32(recipe),UInt32(line),UInt32(productId), Double(size), Boolean(useDefaultSize),BatchStartMode[startMode],DateTime(Int32(plannedStartTimeYear), Int32(plannedStartTimeMonth),Int32(plannedStartTimeDay), Int32(plannedStartTimeHour),Int32(plannedStartTimeMin), Int32(plannedStartTimeSec)),BatchStatus[status],parameterList, Boolean(useDefaultParameterValues), String(error_message),Boolean(doRepeat))return result@func_set_timeout(timeout=5)
def create_batch_ex(site: int,area: int,year: int,order: int,batch: int,recipeCategory: int,recipe: int,line: int,productId: int,useDefaultProductId: bool,size: float,useDefaultSize: bool,startMode: str,plannedStartTimeYear: int,plannedStartTimeMonth: int,plannedStartTimeDay: int,plannedStartTimeHour: int,plannedStartTimeMin: int,plannedStartTimeSec: int,status: str,parameterList: Any,useDefaultParameterValues: bool,batcName: str,errorMessage: str,doRepeat: bool):"""创建工单Ex:param site: 1:param area: 1:param year: 2024:param order: 1:param batch:115:param recipeCategory:1:param recipe:9:param line:1:param productId:0:param size:100:param useDefaultSize:True:param startMode: 'Immediate':param plannedStartTimeYear: 2024:param plannedStartTimeMonth: 12:param plannedStartTimeDay: 1:param status: 'ReadyForRelease':param parameterList:  sistar_batch_params_instance:param useDefaultParameterValues: True:param errorMessage: '':param doRepeat: False:return:"""error_message = errorMessageresult = sistar_batch_instance.CreateBatchEx(UInt16(site), UInt16(area), UInt16(year), UInt32(order), UInt32(batch),UInt32(recipeCategory),UInt32(recipe),UInt32(line),UInt32(productId), Boolean(useDefaultProductId), Double(size),Boolean(useDefaultSize),BatchStartMode[startMode],DateTime(Int32(plannedStartTimeYear), Int32(plannedStartTimeMonth),Int32(plannedStartTimeDay), Int32(plannedStartTimeHour),Int32(plannedStartTimeMin), Int32(plannedStartTimeSec)),BatchStatus[status],parameterList, Boolean(useDefaultParameterValues), String(batcName),String(error_message),Boolean(doRepeat))return result@func_set_timeout(timeout=5)
def delete_batch(site: int,area: int,year: int,order: int,batch: int,recipeCategory: int,errorMessage: str,doRepeat: bool):error_message = errorMessageresult = sistar_batch_instance.DeleteBatch(UInt16(site), UInt16(area),UInt16(year), UInt32(order), UInt32(batch),UInt32(recipeCategory),String(error_message),Boolean(doRepeat))return result@func_set_timeout(timeout=5)
def get_last_error():result = sistar_batch_instance.GetLastError()return result@func_set_timeout(timeout=5)
def get_last_full_error_string():result = sistar_batch_instance.GetLastFullErrorString()return result@func_set_timeout(timeout=5)
def set_batch_parameters(site: int,area: int,year: int,order: int,batch: int,recipeCategory: int,parameterList: Any,errorMessage: str,doRepeat: bool):error_message = errorMessageresult = sistar_batch_instance.SetBatchParameters(UInt16(site), UInt16(area), UInt16(year), UInt32(order),UInt32(batch),UInt32(recipeCategory), parameterList,String(error_message),Boolean(doRepeat))return result@func_set_timeout(timeout=5)
def set_batch_size(site: int,area: int,year: int,order: int,batch: int,recipeCategory: int,size: float,errorMessage: str,doRepeat: bool):error_message = errorMessageresult = sistar_batch_instance.SetBatchSize(UInt16(site), UInt16(area), UInt16(year), UInt32(order), UInt32(batch),UInt32(recipeCategory), Double(size),String(error_message),Boolean(doRepeat))return result@func_set_timeout(timeout=5)
def set_batch_start_data(site: int,area: int,year: int,order: int,batch: int,recipeCategory: int,startMode: str,plannedStartTimeYear,plannedStartTimeMonth,plannedStartTimeDay,plannedStartTimeHour: int,plannedStartTimeMin: int,plannedStartTimeSec: int,errorMessage: str,doRepeat: bool):error_message = errorMessageresult = sistar_batch_instance.SetBatchStartData(UInt16(site), UInt16(area), UInt16(year), UInt32(order),UInt32(batch),UInt32(recipeCategory), BatchStartMode[startMode],DateTime(Int32(plannedStartTimeYear), Int32(plannedStartTimeMonth),Int32(plannedStartTimeDay), Int32(plannedStartTimeHour),Int32(plannedStartTimeMin), Int32(plannedStartTimeSec)),String(error_message),Boolean(doRepeat))return result@func_set_timeout(timeout=5)
def set_batch_status(site: int,area: int,year: int,order: int,batch: int,recipeCategory: int,status: str,errorMessage: str,doRepeat: bool):error_message = errorMessageresult = sistar_batch_instance.SetBatchStatus(UInt16(site), UInt16(area), UInt16(year), UInt32(order),UInt32(batch),UInt32(recipeCategory), BatchStatus[status],String(error_message),Boolean(doRepeat))return result@func_set_timeout(timeout=5)
def set_timeout(timeout: int):result = sistar_batch_instance.SetTimeout(UInt16(timeout))return result@func_set_timeout(timeout=5)
def add_parameter(paramNumber: int, paramValue: str):result = sistar_batch_params_instance.AddParameter(UInt16(paramNumber), String(paramValue))return result@func_set_timeout(timeout=5)
def get_number_at(index: int):result = sistar_batch_params_instance.GetNumberAt(UInt32(index))return result@func_set_timeout(timeout=5)
def get_size():result = sistar_batch_params_instance.GetSize()return result@func_set_timeout(timeout=5)
def get_value_at(index: int):result = sistar_batch_params_instance.GetValueAt(UInt32(index))return result@func_set_timeout(timeout=5)
def fff():return create_batch(site=1, area=1, year=2024, order=1, batch=115, recipeCategory=1, recipe=9, line=1, productId=0, size=100, useDefaultSize=True, startMode='Immediate', plannedStartTimeYear=2024, plannedStartTimeMonth=12, plannedStartTimeDay=1, plannedStartTimeHour=12, plannedStartTimeMin=12, plannedStartTimeSec=1, status='ReadyForRelease', parameterList=sistar_batch_params_instance, useDefaultParameterValues=True, errorMessage='', doRepeat=False)if __name__ == '__main__':try:f = fff()except func_timeout.exceptions.FunctionTimedOut:print('timeout')

import func_timeout
from flask import requestfrom helpers import success_response, error_response
from . import api_bp
from marshmallow import Schema, fields
from .adapter import create_batch, sistar_batch_params_instance, create_batch_ex, delete_batch, get_last_error, \get_last_full_error_string, set_batch_parameters, set_batch_size, set_batch_start_data, set_batch_status, \set_timeout, add_parameter, get_number_at, get_size, get_value_at
from get_logger import  logger
# 定义输入参数模式@api_bp.route('/create_batch', methods=['POST'])
def create_batch_func():"""site, area, year, order, batch, recipeCategory, recipe, line, productId, size, useDefaultSize,startMode, plannedStartTimeYear, plannedStartTimeMonth, plannedStartTimeDay, status,parameterList, useDefaultParameterValues, errorMessage,doRepeat:return:"""res = request.jsonparameterList = sistar_batch_params_instancetry:logger.info(f"开始调用创建工单,参数:{res}")logger.info(f"开始调用创建工单,参数:{res}")result = create_batch(res.get('site'),res.get('area'),res.get('year'),res.get('order'),res.get('batch'),res.get('recipeCategory'),res.get('recipe'),res.get('line'),res.get('productId'),res.get('size'),res.get('useDefaultSize'),res.get('startMode'),res.get('plannedStartTimeYear'),res.get('plannedStartTimeMonth'),res.get('plannedStartTimeDay'),res.get('plannedStartTimeHour'),res.get('plannedStartTimeMin'),res.get('plannedStartTimeSec'),'Locked',parameterList,res.get('useDefaultParameterValues'),res.get('errorMessage'),res.get('doRepeat'))logger.info(f"调用成功,braumat返回数据:{result}")return success_response(data=result)except func_timeout.exceptions.FunctionTimedOut:logger.error(f"调用超时")return error_response(message='timeout',code=504)except Exception as e:logger.error(f"未捕获错误:{str(e)}")return error_response(message=str(e))@api_bp.route('/create_batch_ex', methods=['POST'])
def create_batch_ex_func():"""site, area, year, order, batch, recipeCategory, recipe, line, productId, size, useDefaultSize,startMode, plannedStartTimeYear, plannedStartTimeMonth, plannedStartTimeDay, status,parameterList, useDefaultParameterValues, errorMessage,doRepeat:return:"""res = request.jsonparameterList = sistar_batch_params_instancetry:# 设置超时时间为 5 秒result = create_batch_ex(res.get('site'),res.get('area'),res.get('year'),res.get('order'),res.get('batch'),res.get('recipeCategory'),res.get('recipe'),res.get('line'),res.get('productId'),res.get('useDefaultProductId'),res.get('size'),res.get('useDefaultSize'),res.get('startMode'),res.get('plannedStartTimeYear'),res.get('plannedStartTimeMonth'),res.get('plannedStartTimeDay'),res.get('plannedStartTimeHour'),res.get('plannedStartTimeMin'),res.get('plannedStartTimeSec'),'Locked',parameterList,res.get('useDefaultParameterValues'),res.get('batcName'),res.get('errorMessage'),res.get('doRepeat'))return success_response(data=result)except func_timeout.exceptions.FunctionTimedOut:return error_response(message='timeout')except Exception as e:return error_response(message=str(e))@api_bp.route('/delete_batch', methods=['POST'])
def delete_batch_func():"""site, area, year, order, batch, recipeCategory, recipe, line, productId, size, useDefaultSize,startMode, plannedStartTimeYear, plannedStartTimeMonth, plannedStartTimeDay, status,parameterList, useDefaultParameterValues, errorMessage,doRepeat:return:"""res = request.jsonparameterList = sistar_batch_params_instancetry:# 设置超时时间为 5 秒logger.info(f"开始调用删除工单,参数:{res}")result = delete_batch(res.get('site'),res.get('area'),res.get('year'),res.get('order'),res.get('batch'),res.get('recipeCategory'),res.get('errorMessage'),res.get('doRepeat'))logger.info(f"调用成功,braumat返回数据:{result}")return success_response(data=result)except func_timeout.exceptions.FunctionTimedOut:logger.error(f"调用超时")return error_response(message='timeout')except Exception as e:logger.error(f"未捕获错误:{str(e)}")return error_response(message=str(e))@api_bp.route('/get_last_error', methods=['POST'])
def get_last_error_func():"""site, area, year, order, batch, recipeCategory, recipe, line, productId, size, useDefaultSize,startMode, plannedStartTimeYear, plannedStartTimeMonth, plannedStartTimeDay, status,parameterList, useDefaultParameterValues, errorMessage,doRepeat:return:"""try:# 设置超时时间为 5 秒result = get_last_error()return success_response(data=result)except func_timeout.exceptions.FunctionTimedOut:return error_response(message='timeout')except Exception as e:return error_response(message=str(e))@api_bp.route('/get_last_full_error_string', methods=['POST'])
def get_last_full_error_string_func():"""site, area, year, order, batch, recipeCategory, recipe, line, productId, size, useDefaultSize,startMode, plannedStartTimeYear, plannedStartTimeMonth, plannedStartTimeDay, status,parameterList, useDefaultParameterValues, errorMessage,doRepeat:return:"""try:# 设置超时时间为 5 秒result = get_last_full_error_string()return success_response(data=result)except func_timeout.exceptions.FunctionTimedOut:return error_response(message='timeout')except Exception as e:return error_response(message=str(e))@api_bp.route('/set_batch_parameters', methods=['POST'])
def set_batch_parameters_func():"""site, area, year, order, batch, recipeCategory, recipe, line, productId, size, useDefaultSize,startMode, plannedStartTimeYear, plannedStartTimeMonth, plannedStartTimeDay, status,parameterList, useDefaultParameterValues, errorMessage,doRepeat:return:"""res = request.jsonparameterList = sistar_batch_params_instancetry:# 设置超时时间为 5 秒result = set_batch_parameters(res.get('site'),res.get('area'),res.get('year'),res.get('order'),res.get('batch'),res.get('recipeCategory'),parameterList,res.get('errorMessage'),res.get('doRepeat'))return success_response(data=result)except func_timeout.exceptions.FunctionTimedOut:return error_response(message='timeout')except Exception as e:return error_response(message=str(e))@api_bp.route('/set_batch_size', methods=['POST'])
def set_batch_size_func():"""site, area, year, order, batch, recipeCategory, recipe, line, productId, size, useDefaultSize,startMode, plannedStartTimeYear, plannedStartTimeMonth, plannedStartTimeDay, status,parameterList, useDefaultParameterValues, errorMessage,doRepeat:return:"""res = request.jsonparameterList = sistar_batch_params_instancetry:# 设置超时时间为 5 秒result = set_batch_size(res.get('site'),res.get('area'),res.get('year'),res.get('order'),res.get('batch'),res.get('recipeCategory'),res.get('size'),res.get('errorMessage'),res.get('doRepeat'))return success_response(data=result)except func_timeout.exceptions.FunctionTimedOut:return error_response(message='timeout')except Exception as e:return error_response(message=str(e))@api_bp.route('/set_batch_start_data', methods=['POST'])
def set_batch_start_data_func():"""site, area, year, order, batch, recipeCategory, recipe, line, productId, size, useDefaultSize,startMode, plannedStartTimeYear, plannedStartTimeMonth, plannedStartTimeDay, status,parameterList, useDefaultParameterValues, errorMessage,doRepeat:return:"""res = request.jsonparameterList = sistar_batch_params_instancetry:# 设置超时时间为 5 秒result = set_batch_start_data(res.get('site'),res.get('area'),res.get('year'),res.get('order'),res.get('batch'),res.get('recipeCategory'),res.get('startMode'),res.get('plannedStartTimeYear'),res.get('plannedStartTimeMonth'),res.get('plannedStartTimeDay'),res.get('plannedStartTimeHour'),res.get('plannedStartTimeMin'),res.get('plannedStartTimeSec'),res.get('errorMessage'),res.get('doRepeat'))return success_response(data=result)except func_timeout.exceptions.FunctionTimedOut:return error_response(message='timeout')except Exception as e:return error_response(message=str(e))@api_bp.route('/set_batch_status', methods=['POST'])
def set_batch_status_func():"""site, area, year, order, batch, recipeCategory, recipe, line, productId, size, useDefaultSize,startMode, plannedStartTimeYear, plannedStartTimeMonth, plannedStartTimeDay, status,parameterList, useDefaultParameterValues, errorMessage,doRepeat:return:"""res = request.jsonparameterList = sistar_batch_params_instancetry:# 设置超时时间为 5 秒result = set_batch_status(res.get('site'),res.get('area'),res.get('year'),res.get('order'),res.get('batch'),res.get('recipeCategory'),res.get('status'),res.get('errorMessage'),res.get('doRepeat'))return success_response(data=result)except func_timeout.exceptions.FunctionTimedOut:return error_response(message='timeout')except Exception as e:return error_response(message=str(e))@api_bp.route('/set_timeout', methods=['POST'])
def set_timeout_func():"""site, area, year, order, batch, recipeCategory, recipe, line, productId, size, useDefaultSize,startMode, plannedStartTimeYear, plannedStartTimeMonth, plannedStartTimeDay, status,parameterList, useDefaultParameterValues, errorMessage,doRepeat:return:"""res = request.jsonparameterList = sistar_batch_params_instancetry:# 设置超时时间为 5 秒result = set_timeout(res.get('timeout'))return success_response(data=result)except func_timeout.exceptions.FunctionTimedOut:return error_response(message='timeout')except Exception as e:return error_response(message=str(e))@api_bp.route('/add_parameter', methods=['POST'])
def add_parameter_func():"""site, area, year, order, batch, recipeCategory, recipe, line, productId, size, useDefaultSize,startMode, plannedStartTimeYear, plannedStartTimeMonth, plannedStartTimeDay, status,parameterList, useDefaultParameterValues, errorMessage,doRepeat:return:"""res = request.jsonparameterList = sistar_batch_params_instancetry:# 设置超时时间为 5 秒result = add_parameter(res.get('paramNumber'),res.get('paramValue'))return success_response(data=result)except func_timeout.exceptions.FunctionTimedOut:return error_response(message='timeout')except Exception as e:return error_response(message=str(e))@api_bp.route('/get_number_at', methods=['POST'])
def get_number_at_func():"""site, area, year, order, batch, recipeCategory, recipe, line, productId, size, useDefaultSize,startMode, plannedStartTimeYear, plannedStartTimeMonth, plannedStartTimeDay, status,parameterList, useDefaultParameterValues, errorMessage,doRepeat:return:"""res = request.jsonparameterList = sistar_batch_params_instancetry:# 设置超时时间为 5 秒result = get_number_at(res.get('index'))return success_response(data=result)except func_timeout.exceptions.FunctionTimedOut:return error_response(message='timeout')except Exception as e:return error_response(message=str(e))@api_bp.route('/get_size', methods=['POST'])
def get_size_func():"""site, area, year, order, batch, recipeCategory, recipe, line, productId, size, useDefaultSize,startMode, plannedStartTimeYear, plannedStartTimeMonth, plannedStartTimeDay, status,parameterList, useDefaultParameterValues, errorMessage,doRepeat:return:"""res = request.jsonparameterList = sistar_batch_params_instancetry:# 设置超时时间为 5 秒result = get_size()return success_response(data=result)except func_timeout.exceptions.FunctionTimedOut:return error_response(message='timeout')except Exception as e:return error_response(message=str(e))@api_bp.route('/get_value_at', methods=['POST'])
def get_value_at_func():"""site, area, year, order, batch, recipeCategory, recipe, line, productId, size, useDefaultSize,startMode, plannedStartTimeYear, plannedStartTimeMonth, plannedStartTimeDay, status,parameterList, useDefaultParameterValues, errorMessage,doRepeat:return:"""res = request.jsonparameterList = sistar_batch_params_instancetry:# 设置超时时间为 5 秒result = get_value_at(res.get('index'))return success_response(data=result)except func_timeout.exceptions.FunctionTimedOut:return error_response(message='timeout')except Exception as e:return error_response(message=str(e))