# -*- coding: utf-8 -*- import re import ast import json import logging from datetime import datetime from odoo import http from odoo.http import request def convert_to_seconds(time_str): # 修改正则表达式，使 H、M、S 部分可选 pattern = r"(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?" match = re.match(pattern, time_str) if match: # 提取各时间单位，如果某个单位缺失则默认设为0 hours = int(match.group(1)) if match.group(1) else 0 minutes = int(match.group(2)) if match.group(2) else 0 seconds = int(match.group(3)) if match.group(3) else 0 # 计算总秒数 total_seconds = hours * 3600 + minutes * 60 + seconds if total_seconds == 0: # return None pattern = r"(?:(\d+)小时)?(?:(\d+)分钟)?(?:(\d+)秒)?" match = re.match(pattern, time_str) if match: # 提取各时间单位，如果某个单位缺失则默认设为0 hours = int(match.group(1)) if match.group(1) else 0 minutes = int(match.group(2)) if match.group(2) else 0 seconds = int(match.group(3)) if match.group(3) else 0 # 计算总秒数 total_seconds = hours * 3600 + minutes * 60 + seconds return total_seconds else: return None return total_seconds class Sf_Dashboard_Connect(http.Controller): @http.route('/api/get_machine_datas/list', type='http', auth='public', methods=['GET', 'POST'], csrf=False, cors="*") def get_machine_datas_list(self, **kw): """ 拿到机床数据返回给大屏展示 :param kw: :return: """ res = {'status': 1, 'message': '成功', 'data': []} logging.info('前端请求机床数据的参数为:%s' % kw) # 获取当前时间的时间戳 current_timestamp = datetime.now().timestamp() print(current_timestamp) # tem_list = [ # "XT-GNJC-WZZX-X800-Y550-Z550-T24-A5-1", "XT-GNJC-LSZX-X800-Y550-Z550-T24-A3-3", # "XT-GNJC-LSZX-X800-Y550-Z550-T24-A3-4", "XT-GNJC-LSZX-X800-Y550-Z550-T24-A3-5", # "XT-GNJC-LSZX-X800-Y550-Z550-T24-A3-6", "XT-GNJC-LSZX-X800-Y550-Z550-T24-A3-7", # "XT-GNJC-LSZX-X800-Y550-Z550-T24-A3-8", "XT-GNJC-WZZX-X800-Y550-Z550-T24-A5-2", # "XT-GNJC-GSZG-X600-Y400-Z350-T21-A3-9", "XT-GNJC-GSZG-X600-Y400-Z350-T21-A3-10", # "XT-GNJC-GSZG-X600-Y400-Z350-T21-A3-11", "XT-GNJC-GSZG-X600-Y400-Z350-T21-A3-12", # "XT-GNJC-GSZG-X600-Y400-Z350-T21-A3-13", "XT-GNJC-GSZG-X600-Y400-Z350-T21-A3-14" # ] try: equipment_obj = request.env['maintenance.equipment'].sudo() # 获取请求的机床数据 machine_list = ast.literal_eval(kw['machine_list']) for item in machine_list: machine_data = equipment_obj.search([('code', '=', item)]) # 机床上线时间段 first_online_duration = current_timestamp - int(machine_data.first_online_time.timestamp()) power_off_time = None power_off_rate = None if machine_data.machine_power_on_time: power_off_time = first_online_duration - convert_to_seconds(machine_data.machine_power_on_time) power_off_rate = round((power_off_time / first_online_duration), 3) else: power_off_time = False power_off_rate = False if machine_data: res['data'].append({ 'active': machine_data.status, 'id': machine_data.id, 'name': machine_data.name, 'brand': machine_data.type_id.name, 'code': machine_data.code, 'status': machine_data.status, 'run_status': machine_data.run_status, 'run_time': machine_data.run_time, 'system_date': machine_data.system_date, 'system_time': machine_data.system_time, 'cut_time': machine_data.cut_time, 'cut_status': machine_data.cut_status, 'program': machine_data.program, 'program_name': machine_data.program_name, 'program_status': machine_data.program_status, 'tool_num': machine_data.tool_num, 'machine_power_on_time': machine_data.machine_power_on_time, 'product_counts': machine_data.product_counts, 'mode': machine_data.mode, 'start_time': machine_data.start_time, 'end_time': machine_data.end_time, 'program_start_time': machine_data.program_start_time, 'program_end_time': machine_data.program_end_time, 'standby_start_time': machine_data.standby_start_time, 'standby_end_time': machine_data.standby_end_time, 'offline_start_time': machine_data.offline_start_time, 'offline_end_time': machine_data.offline_end_time, 'emg_status': machine_data.emg_status, 'current_program': machine_data.current_program, 'current_program_seq': machine_data.current_program_seq, 'x_abs_pos': machine_data.x_abs_pos, 'y_abs_pos': machine_data.y_abs_pos, 'z_abs_pos': machine_data.z_abs_pos, 'feed_speed_set': machine_data.feed_speed_set, 'act_feed_speed': machine_data.act_feed_speed, 'spindle_speed_set': machine_data.spindle_speed_set, 'act_spindle_speed': machine_data.act_spindle_speed, 'spindle_load': machine_data.spindle_load, 'x_axis_load': machine_data.x_axis_load, 'y_axis_load': machine_data.y_axis_load, 'z_axis_load': machine_data.z_axis_load, 'rapid_feed': machine_data.rapid_feed, 'feed_rate': machine_data.feed_rate, 'x_mach_coord': machine_data.x_mach_coord, 'y_mach_coord': machine_data.y_mach_coord, 'z_mach_coord': machine_data.z_mach_coord, 'x_rel_coord': machine_data.x_rel_coord, 'y_rel_coord': machine_data.y_rel_coord, 'z_rel_coord': machine_data.z_rel_coord, 'x_dis_coord': machine_data.x_dis_coord, 'y_dis_coord': machine_data.y_dis_coord, 'z_dis_coord': machine_data.z_dis_coord, 'alarm_time': machine_data.alarm_time, 'alarm_msg': machine_data.alarm_msg, 'clear_time': machine_data.clear_time, # 计算出来的数据 # 开动率：运行时间/通电时间 'run_rate': machine_data.run_rate, # 关机时长：初次上线时间 - 通电时间 'power_off_time': power_off_time, # 关机率：关机时长/初次上线时间 'power_off_rate': power_off_rate, 'first_online_duration': first_online_duration, # 停机时间：关机时间 - 运行时间 # 停机时长：关机时间 - 初次上线时间 }) return json.JSONEncoder().encode(res) except Exception as e: logging.info('前端请求机床数据失败，原因:%s' % e) res['status'] = -1 res['message'] = '前端请求机床数据失败，原因:%s' % e return json.JSONEncoder().encode(res) # @http.route('/api/logs/list', type='http', auth='public', methods=['GET', 'POST'], csrf=False, # cors="*") # def logs_list(self, **kw): # """ # 拿到日志数据返回给大屏展示 # :param kw: # :return: # """ # res = {'status': 1, 'message': '成功', 'data': []} # logging.info('前端请求日志数据的参数为:%s' % kw) # # try: # # 获取请求的日志数据 # logs_obj = request.env['maintenance.equipment.oee.log.detail'].sudo() # # 获取请求的机床数据 # machine_list = ast.literal_eval(kw['machine_list']) # begin_time_str = kw['begin_time'].strip('"') # end_time_str = kw['end_time'].strip('"') # # begin_time = datetime.strptime(begin_time_str, '%Y-%m-%d %H:%M:%S') # end_time = datetime.strptime(end_time_str, '%Y-%m-%d %H:%M:%S') # # print('begin_time: %s' % begin_time) # for item in machine_list: # log_datas = logs_obj.search( # [('equipment_code', '=', item), ('time', '>=', begin_time), ('time', '<=', end_time)]) # print('log_datas: %s' % log_datas) # for log_data in log_datas: # res['data'].append({ # 'equipment_code': log_data.equipment_code, # 'time': log_data.time.strftime('%Y-%m-%d %H:%M:%S'), # 'state': log_data.state # # }) # # return json.JSONEncoder().encode(res) # # except Exception as e: # logging.info('前端请求日志数据失败，原因:%s' % e) # res['status'] = -1 # res['message'] = '前端请求日志数据失败，原因:%s' % e # return json.JSONEncoder().encode(res) @http.route('/api/logs/list', type='http', auth='public', methods=['GET', 'POST'], csrf=False, cors="*") def logs_list(self, **kw): """ 拿到日志数据返回给大屏展示 :param kw: :return: """ res = {'status': 1, 'message': '成功', 'data': {}} logging.info('前端请求日志数据的参数为:%s' % kw) try: # 获取请求的日志数据 logs_obj = request.env['maintenance.equipment.oee.log.detail'].sudo() # 获取请求的机床数据 machine_list = ast.literal_eval(kw['machine_list']) begin_time_str = kw['begin_time'].strip('"') end_time_str = kw['end_time'].strip('"') begin_time = datetime.strptime(begin_time_str, '%Y-%m-%d %H:%M:%S') end_time = datetime.strptime(end_time_str, '%Y-%m-%d %H:%M:%S') print('begin_time: %s' % begin_time) for item in machine_list: log_datas = logs_obj.search( [('equipment_code', '=', item), ('time', '>=', begin_time), ('time', '<=', end_time)]) print('log_datas: %s' % log_datas) # 将数据按照 equipment_code 进行分组 if item not in res['data']: res['data'][item] = [] for log_data in log_datas: res['data'][item].append({ 'time': log_data.time.strftime('%Y-%m-%d %H:%M:%S'), 'state': log_data.state, 'production_name': log_data.production_name, }) return json.dumps(res) # 注意使用 json.dumps 而不是直接用 json.JSONEncoder().encode() except Exception as e: logging.info('前端请求日志数据失败，原因:%s' % e) res['status'] = -1 res['message'] = '前端请求日志数据失败，原因:%s' % e return json.dumps(res) # 返回CNC机床列表 @http.route('/api/CNCList', type='http', auth='public', methods=['GET', 'POST'], csrf=False, cors="*") def CNCList(self, **kw): """ 获取CNC机床列表 :param kw: :return: """ # logging.info('CNCList:%s' % kw) try: res = {'Succeed': True} # cnc_list = request.env['sf.cnc.equipment'].sudo().search([]) # cnc_list = ["XT-GNJC-WZZX-X800-Y550-Z550-T24-A5-1", "XT-GNJC-LSZX-X800-Y550-Z550-T24-A3-3", # "XT-GNJC-LSZX-X800-Y550-Z550-T24-A3-4", "XT-GNJC-LSZX-X800-Y550-Z550-T24-A3-5", # "XT-GNJC-LSZX-X800-Y550-Z550-T24-A3-6", "XT-GNJC-LSZX-X800-Y550-Z550-T24-A3-7", # "XT-GNJC-LSZX-X800-Y550-Z550-T24-A3-8", "XT-GNJC-WZZX-X800-Y550-Z550-T24-A5-2", # "XT-GNJC-GSZG-X600-Y400-Z350-T21-A3-9", "XT-GNJC-GSZG-X600-Y400-Z350-T21-A3-10", # "XT-GNJC-GSZG-X600-Y400-Z350-T21-A3-11", "XT-GNJC-GSZG-X600-Y400-Z350-T21-A3-12", # "XT-GNJC-GSZG-X600-Y400-Z350-T21-A3-13", "XT-GNJC-GSZG-X600-Y400-Z350-T21-A3-14"] cnc_list_obj = request.env['maintenance.equipment'].sudo().search( [('function_type', '!=', False), ('active', '=', True)]) cnc_list = list(map(lambda x: x.code, cnc_list_obj)) print('cnc_list: %s' % cnc_list) res['CNCList'] = cnc_list except Exception as e: res = {'Succeed': False, 'ErrorCode': 202, 'Error': e} logging.info('CNCList error:%s' % e) return json.JSONEncoder().encode(res)