一、项目背景
江苏省作为中国重要的制造业基地，钢铁行业在其中占据重要地位。随着工业4.0的推进，钢铁厂需要通过智能化手段提升生产效率、降低设备故障率、优化质量控制。开源Odoo MES系统与SKF Observer Phoenix API的双向对接，能够实现设备状态监测、生产数据采集和质量控制的自动化与智能化，为江苏钢铁行业提供高效、透明的生产管理解决方案。
二、集成目标
设备状态监测：通过SKF Observer Phoenix API获取设备的振动、温度等状态数据，实时同步到Odoo MES系统。
生产数据采集：利用Odoo MES系统采集生产现场的设备运行数据和生产进度数据，实现生产过程的可视化。
质量控制：同步生产过程中的质量检测数据，确保产品质量符合标准。
三、技术架构
- Odoo MES模块：
• 设备管理：用于设备状态监测和维护工单管理。
• 生产管理：用于生产数据采集和生产进度跟踪。
• 质量管理：用于质量检测数据的记录和分析。
- SKF Observer Phoenix 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/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)}")
- 真实案例：连云港兴鑫钢铁通过建设超低排放一体化管控平台，集成生产运行信号和质量监测数据，提升了智能化管理水平。
（三）质量控制集成
- 数据模型设计
# 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)}
- 真实案例：某江苏钢铁厂通过部署数据采集系统，实现了生产过程的实时监控和数据分析，提高了生产效率。
五、安全与配置
API密钥管理：在Odoo的系统参数中存储SKF API密钥，通过加密字段保护。
HTTPS加密：所有API调用均通过HTTPS传输，确保数据安全。
IP白名单：限制SKF API仅允许Odoo服务器的IP访问。
六、部署与测试
- 部署步骤：
• 安装Odoo自定义模块。
• 配置定时任务，如每30分钟同步一次设备状态数据。
• 在SKF Observer API中注册Odoo的Webhook URL。
- 测试案例：
• 设备状态同步测试：模拟设备振动异常，验证是否自动生成维护工单。
• 生产数据采集测试：实时采集生产数据，验证数据的完整性和准确性。
• 质量数据同步测试：同步质量检测数据，验证是否正确记录在Odoo中。
七、总结与展望
通过Odoo MES系统与SKF Observer Phoenix API的双向集成，江苏钢铁厂实现了设备状态的实时监测、生产数据的实时采集以及质量数据的同步，提升了生产效率和智能化管理水平。未来可以进一步扩展功能，如集成数字孪生技术，实现生产过程的可视化监控。
让转型不迷航——邹工转型手札