当前位置：首页 > article >正文

《手札·数转篇》开源Odoo软件与SKF Observer API钢铁厂双向集成方案

article 2025/2/12 10:24:15

一、项目背景
在钢铁厂的生产环境中，设备的稳定运行、质量控制以及数据采集是确保生产效率和产品质量的关键。开源Odoo软件提供了强大的维护模块、质量模块和数据采集（数采）模块，结合SKF Observer API，可以实现设备状态监测、质量数据同步以及生产数据的实时采集与分析，从而提升钢铁厂的智能化管理水平。

二、集成目标

在这里插入图片描述

设备维护模块集成：通过Odoo维护模块与SKF Observer API对接，实时获取设备状态数据（如振动、温度等），并根据数据生成维护工单。
质量模块集成：同步生产过程中的质量检测数据，确保原材料和成品的质量符合标准。
数采模块集成：实时采集生产现场的设备运行数据，包括设备状态、生产参数等，提升生产过程的透明度。

三、技术架构

Odoo模块：

• 维护模块：用于设备状态监测和维护工单管理。

• 质量模块：用于质量检测数据的记录和分析。

• 数采模块：用于实时采集生产现场的设备运行数据。

SKF Observer API：

• 提供设备状态监测数据（如振动、温度等）。

• 支持通过API接收维护工单状态更新。

中间层：

• 使用Python脚本作为调度器，定时拉取SKF数据并触发Odoo业务逻辑。

四、集成方案

（一）设备维护模块集成

数据模型设计

   # models/maintenance_equipment.py
   from odoo import models, fields

   class MaintenanceEquipment(models.Model):
       _inherit = 'maintenance.equipment'

       skf_id = fields.Char('SKF设备ID')
       vibration_threshold = fields.Float('振动阈值（mm/s）')
       last_sync_time = fields.Datetime('最后同步时间')

定时任务实现

   # models/maintenance_sync.py
   from odoo import models, api
   import requests
   import logging

   class MaintenanceSync(models.Model):
       _name = 'maintenance.sync'

       @api.model
       def cron_sync_equipment_status(self):
           equipments = self.env['maintenance.equipment'].search([('skf_id', '!=', False)])
           skf_api_key = self.env['ir.config_parameter'].sudo().get_param('skf.api_key')

           for equipment in equipments:
               url = f"https://api.skf.com/observer/v1/devices/{equipment.skf_id}/sensor_data"
               params = {'start_time': equipment.last_sync_time.isoformat() if equipment.last_sync_time else '2024-01-01T00:00:00Z'}
               headers = {'Authorization': f'Bearer {skf_api_key}'}

               try:
                   response = requests.get(url, headers=headers, params=params)
                   data = response.json()
                   for entry in data.get('data', []):
                       if entry['vibration'] > equipment.vibration_threshold:
                           self.env['maintenance.request'].create({
                               'name': f"设备{equipment.name}振动异常",
                               'equipment_id': equipment.id,
                               'description': f"振动值：{entry['vibration']} mm/s，超出阈值{equipment.vibration_threshold} mm/s"
                           })
                   equipment.last_sync_time = fields.Datetime.now()
               except Exception as e:
                   logging.error(f"同步失败: {str(e)}")

真实案例：某钢铁厂通过部署设备在线监测系统，成功减少了设备突发停机事件，提高了设备的运行效率。

（二）质量模块集成

数据模型设计

   # models/quality_check.py
   from odoo import models, fields

   class QualityCheck(models.Model):
       _name = 'quality.check'

       product_id = fields.Many2one('product.product', '产品')
       check_date = fields.Datetime('检测日期')
       result = fields.Selection([('pass', '合格'), ('fail', '不合格')], '检测结果')
       notes = fields.Text('备注')

数据同步实现

   # controllers/quality_sync.py
   from odoo import http
   import requests
   import json

   class QualitySyncController(http.Controller):
       @http.route('/quality/sync', type='json', auth='user')
       def sync_quality_data(self):
           skf_api_key = http.request.env['ir.config_parameter'].sudo().get_param('skf.api_key')
           url = "https://api.skf.com/observer/v1/quality_data"
           headers = {'Authorization': f'Bearer {skf_api_key}'}

           try:
               response = requests.get(url, headers=headers)
               data = response.json()
               for entry in data.get('quality_checks', []):
                   product = http.request.env['product.product'].search([('default_code', '=', entry['product_code'])])
                   if product:
                       http.request.env['quality.check'].create({
                           'product_id': product.id,
                           'check_date': entry['check_date'],
                           'result': entry['result'],
                           'notes': entry['notes']
                       })
               return {'success': True}
           except Exception as e:
               return {'error': str(e)}

真实案例：连云港兴鑫钢铁通过建设超低排放一体化管控平台，集成生产运行信号和质量监测数据，提升了智能化管理水平。

（三）数采模块集成

数据模型设计

   # models/production_data.py
   from odoo import models, fields

   class ProductionData(models.Model):
       _name = 'production.data'

       equipment_id = fields.Many2one('maintenance.equipment', '设备')
       timestamp = fields.Datetime('时间戳')
       production_rate = fields.Float('生产速率')
       quality_index = fields.Float('质量指数')

数据采集实现

   # models/production_sync.py
   from odoo import models, api
   import requests
   import logging

   class ProductionSync(models.Model):
       _name = 'production.sync'

       @api.model
       def cron_sync_production_data(self):
           equipments = self.env['maintenance.equipment'].search([('skf_id', '!=', False)])
           skf_api_key = self.env['ir.config_parameter'].sudo().get_param('skf.api_key')

           for equipment in equipments:
               url = f"https://api.skf.com/observer/v1/production_data/{equipment.skf_id}"
               headers = {'Authorization': f'Bearer {skf_api_key}'}

               try:
                   response = requests.get(url, headers=headers)
                   data = response.json()
                   for entry in data.get('production_data', []):
                       self.env['production.data'].create({
                           'equipment_id': equipment.id,
                           'timestamp': entry['timestamp'],
                           'production_rate': entry['production_rate'],
                           'quality_index': entry['quality_index']
                       })
               except Exception as e:
                   logging.error(f"数据采集失败: {str(e)}")