jikimo/test
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化sf获取cnc程序和ftp

Browse Source
This commit is contained in:
jinling.yang
2023-01-09 19:54:02 +08:00
parent c023a13e3a
commit b7723f82d0
24 changed files with 473 additions and 224 deletions
Download Patch File Download Diff File Expand all files Collapse all files

202
sf_manufacturing/models/mrp_workorder.py
View File

@@ -4,6 +4,7 @@ import math
import requests
import logging
import base64
import hashlib
# import subprocess
from datetime import datetime
from dateutil.relativedelta import relativedelta
@@ -19,6 +20,7 @@ class ResMrpWorkOrder(models.Model):
_order = 'sequence'
workcenter_id = fields.Many2one('mrp.workcenter', required=False)
users_ids = fields.Many2many("res.users", 'users_workorder', related="workcenter_id.users_ids")
processing_panel = fields.Char('加工面')
sequence = fields.Integer(string='工序')
routing_type = fields.Selection([
@@ -29,8 +31,26 @@ class ResMrpWorkOrder(models.Model):
('后置三元质量检测', '后置三元质量检测'),
('解除装夹', '解除装夹'),
], string="工序类型")
@api.onchange('users_ids')
def get_user_permissions(self):
uid = self.env.uid
for workorder in self:
if workorder.users_ids:
list_user_id = []
for item in workorder.users_ids:
list_user_id.append(item.id)
if uid in list_user_id:
workorder.user_permissions = True
else:
workorder.user_permissions = False
else:
workorder.user_permissions = False
user_permissions = fields.Boolean('用户权限', compute='get_user_permissions')
programming_no = fields.Char('编程单号')
is_programming = fields.Boolean('是否编程', default=False)
work_state = fields.Char('业务状态')
programming_state = fields.Char('编程状态')
cnc_worksheet = fields.Binary(
'工作指令', readonly=True)
material_center_point = fields.Char(string='配料中心点')
@@ -106,7 +126,11 @@ class ResMrpWorkOrder(models.Model):
print("(%.2f,%.2f)" % (x, y))
self.material_center_point = ("(%.2f,%.2f,%.2f)" % (x, y, z))
self.X_deviation_angle = jdz
return self.material_center_point
# 将补偿值写入CNC加工工单
workorder = self.env['mrp.workorder'].browse(self.ids)
work = workorder.production_id.workorder_ids
work.compensation_value_x = eval(self.material_center_point)[0]
work.compensation_value_y = eval(self.material_center_point)[1]
def json_workorder_str(self, k, production, route):
workorders_values_str = [0, '', {
@@ -116,6 +140,7 @@ class ResMrpWorkOrder(models.Model):
'name': route.route_workcenter_id.name,
'processing_panel': k,
'routing_type': route.routing_type,
'work_state': '' if not route.routing_type == '获取CNC加工程序' else '待发起',
'workcenter_id': self.env['mrp.routing.workcenter'].get_workcenter(route.workcenter_ids.ids),
'date_planned_start': False,
'date_planned_finished': False,
@@ -233,35 +258,72 @@ class ResMrpWorkOrder(models.Model):
else:
return True
# def fetchCNCing(self):
# return None
# cnc程序获取
def fetchCNC(self):
cnc = self.env['mrp.workorder'].search(
[('routing_type', '=', 'CNC加工'), ('production_id', '=', self.production_id.id)], limit=1)
res = {'model_code': cnc.product_id.barcode, 'production_no': self.production_id.name,
'machine_tool_code': cnc.workcenter_id.machine_tool_id.code,
'material_code': cnc.env['sf.production.materials'].search(
[('id', '=', cnc.product_id.materials_id.id)]).materials_no,
'material_type_code': cnc.env['sf.materials.model'].search(
[('id', '=', cnc.product_id.materials_type_id.id)]).materials_no,
'machining_precision': cnc.product_id.model_machining_precision,
'embryo_long': cnc.product_id.bom_ids.bom_line_ids.product_id.length,
'embryo_height': cnc.product_id.bom_ids.bom_line_ids.product_id.height,
'embryo_width': cnc.product_id.bom_ids.bom_line_ids.product_id.width,
'order_no': cnc.production_id.origin,
'user': self.env.user.name,
'model_file': base64.b64encode(cnc.product_id.model_file).decode('utf-8')
}
logging.info('res:%s' % res)
configsettings = self.env['res.config.settings'].get_values()
config_header = Common.get_headers(self, configsettings['token'], configsettings['sf_secret_key'])
url = '/api/intelligent_programming/create'
config_url = configsettings['sf_url'] + url
# res_str = json.dumps(res)
ret = requests.post(config_url, json={}, data=res, headers=config_header)
ret = ret.json()
result = json.loads(ret['result'])
if result['status'] == 1:
return self.write({'programming_no': result['programming_no'], 'is_programming': True})
try:
cnc = self.env['mrp.workorder'].search(
[('routing_type', '=', 'CNC加工'), ('production_id', '=', self.production_id.id)], limit=1)
logging.info('fetchCNC-cnc:%s' % cnc)
# if cnc.product_id.upload_model_file:
# logging.info('fetchCNC-upload_model_file:%s' % cnc.product_id.upload_model_file)
# attachments = cnc.product_id.upload_model_file[0]
# logging.info('fetchCNC-attachment1:%s' % attachments)
# logging.info('fetchCNC-attachment1:%s' % cnc.product_id.upload_model_file[0])
# logging.info('fetchCNC-attachment2:%s' % cnc.product_id.upload_model_file[0].datas)
# logging.info('fetchCNC-attachment:%s' % attachments.datas)
# base64_data = base64.b64encode(attachments.datas)
# logging.info('fetchCNC-attachment1:%s' % attachments)
# base64_datas = base64_data.decode('utf-8')
# model_code = hashlib.sha1(base64_datas.encode('utf-8')).hexdigest()
# logging.info('fetchCNC-model_code:%s' % model_code)
logging.info('fetchCNC-model_code1:%s' % cnc.product_id.model_code)
res = {'model_code': '' if not cnc.product_id.model_code else cnc.product_id.model_code,
'production_no': self.production_id.name,
'machine_tool_code': cnc.workcenter_id.machine_tool_id.code,
'material_code': cnc.env['sf.production.materials'].search(
[('id', '=', cnc.product_id.materials_id.id)]).materials_no,
'material_type_code': cnc.env['sf.materials.model'].search(
[('id', '=', cnc.product_id.materials_type_id.id)]).materials_no,
'machining_processing_panel': cnc.product_id.model_processing_panel,
'machining_precision': cnc.product_id.model_machining_precision,
'embryo_long': cnc.product_id.bom_ids.bom_line_ids.product_id.length,
'embryo_height': cnc.product_id.bom_ids.bom_line_ids.product_id.height,
'embryo_width': cnc.product_id.bom_ids.bom_line_ids.product_id.width,
'order_no': cnc.production_id.origin,
'user': self.env.user.name,
'model_file': '' if not cnc.product_id.model_file else base64.b64encode(
cnc.product_id.model_file).decode('utf-8')
}
logging.info('res:%s' % res)
configsettings = self.env['res.config.settings'].get_values()
config_header = Common.get_headers(self, configsettings['token'], configsettings['sf_secret_key'])
url = '/api/intelligent_programming/create'
config_url = configsettings['sf_url'] + url
# res_str = json.dumps(res)
ret = requests.post(config_url, json={}, data=res, headers=config_header)
ret = ret.json()
logging.info('fetchCNC-ret:%s' % ret)
if ret['status'] == 1:
self.write(
{'programming_no': ret['programming_no'], 'programming_state': '编程中', 'work_state': '编程中'})
else:
logging.info('fetchCNC-error:%s' % cnc)
raise UserError('行业资源库解析失败')
except Exception as e:
logging.info('fetchCNC error:%s' % e)
raise UserError(e)
# return {
# 'name': _("工单"),
# 'view_mode': 'form',
# 'res_model': 'mrp.workorder',
# 'res_id': self.id,
# 'type': 'ir.actions.act_window',
# 'target': 'new'
# }
def json_workorder_str1(self, k, production, route):
workorders_values_str = [0, '', {
@@ -271,6 +333,7 @@ class ResMrpWorkOrder(models.Model):
'name': route.route_workcenter_id.name,
'processing_panel': k,
'routing_type': route.routing_type,
'work_state': '' if not route.routing_type == '获取CNC加工程序' else '待发起',
'workcenter_id': self.env['mrp.routing.workcenter'].get_workcenter(route.workcenter_ids.ids),
'date_planned_start': False,
'date_planned_finished': False,
@@ -349,44 +412,73 @@ class CNCprocessing(models.Model):
workorder_id = fields.Many2one('mrp.workorder', string="工单")
# mrs下发编程单创建CNC加工
def cnc_processing_create(self, obj):
workorder = self.env['mrp.workorder'].search([('production_id.name', '=', obj['production_order_no']),
('processing_panel', '=', obj['processing_panel']),
('routing_type', '=', 'CNC加工')])
vals = {
'workorder_id': workorder.id,
'sequence_number': obj['sequence_number'],
'program_name': obj['program_name'],
'cutting_tool_name': obj['cutting_tool_name'],
'cutting_tool_no': obj['cutting_tool_no'],
'processing_type': obj['processing_type'],
'margin_x_y': obj['margin_x_y'],
'margin_z': obj['margin_z'],
'depth_of_processing_z': obj['depth_of_processing_z'],
'cutting_tool_extension_length': obj['cutting_tool_extension_length'],
'cutting_tool_handle_type': obj['cutting_tool_handle_type'],
'estimated_processing_time': obj['estimated_processing_time'],
'remark': obj['remark']
}
return self.env['sf.cnc.processing'].create(vals)
def cnc_processing_create(self, cnc_workorder, ret):
for obj in ret['programming_list']:
workorder = self.env['mrp.workorder'].search([('production_id.name', '=', ret['production_order_no']),
('processing_panel', '=', obj['processing_panel']),
('routing_type', '=', 'CNC加工')])
cnc_processing = self.env['sf.cnc.processing'].create({
'workorder_id': workorder.id,
'sequence_number': obj['sequence_number'],
'program_name': obj['program_name'],
'cutting_tool_name': obj['cutting_tool_name'],
'cutting_tool_no': obj['cutting_tool_no'],
'processing_type': obj['processing_type'],
'margin_x_y': obj['margin_x_y'],
'margin_z': obj['margin_z'],
'depth_of_processing_z': obj['depth_of_processing_z'],
'cutting_tool_extension_length': obj['cutting_tool_extension_length'],
'cutting_tool_handle_type': obj['cutting_tool_handle_type'],
'estimated_processing_time': obj['estimated_processing_time'],
'remark': obj['remark']
})
self.get_cnc_processing_file(ret['folder_name'], cnc_processing)
cnc_workorder.state = 'done'
cnc_workorder.work_state = '已编程'
cnc_workorder.programming_state = '已编程'
cnc_workorder.time_ids.date_end = datetime.now()
def get_cnc_processing_file(self, folder_name, cnc_processing):
logging.info('folder_name:%s' % folder_name)
serverdir = os.path.join('/', folder_name, 'return', cnc_processing.processing_panel)
logging.info('serverdir:%s' % serverdir)
for root, dirs, files in os.walk(server_dir):
for f in files:
logging.info('f:%s' % f)
if os.path.splitext(f)[1] == ".pdf":
full_path = os.path.join(server_dir, root, f)
logging.info('pdf:%s' % full_path)
if full_path != False:
if not cnc_processing.workorder_id.cnc_worksheet:
cnc_processing.workorder_id.cnc_worksheet = base64.b64encode(
open(full_path, 'rb').read())
else:
if cnc_processing.program_name == f.split('.')[0]:
cnc_file_path = os.path.join(server_dir, root, f)
logging.info('cnc_file_path:%s' % cnc_file_path)
cnc_processing.with_user(request.env.ref("base.user_admin")).write_file(cnc_file_path,
cnc_processing)
# 创建附件(nc文件)
def attachment_create(self, name, data):
attachment = self.env['ir.attachment'].create({
'datas': base64.b64encode(data),
'type': 'binary',
'public': True,
'description': '程序文件',
'name': name
})
return attachment
# 将FTP的nc文件下载到临时目录
def download_file_tmp(self, model_code, processing_panel):
remotepath = os.path.join('/', model_code, 'return', processing_panel)
serverdir = os.path.join('/tmp', model_code, 'return', processing_panel)
ftp = FtpController()
ftp.download_file_tree(remotepath, serverdir)
return serverdir
def download_file_tmp(self, production_no, processing_panel):
remotepath = os.path.join('/', production_no, 'return', processing_panel)
serverdir = os.path.join('/tmp', production_no, 'return', processing_panel)
ftp_resconfig = self.env['res.config.settings'].get_values()
ftp = FtpController(str(ftp_resconfig['ftp_host']), int(ftp_resconfig['ftp_port']), ftp_resconfig['ftp_user'],
ftp_resconfig['ftp_password'])
download_state = ftp.download_file_tree(remotepath, serverdir)
return download_state
# 将nc文件存到attach的datas里
def write_file(self, nc_file_path, cnc):