數字化生產管理系統設計

下面提供一個基于Python的數字化管理系統框架，使用現代技術棧實現。這個系統將包含設備監控、故障管理、裝配配套表生成、生產計劃管理等功能。

系統架構

數字化生產管理系統
├── 設備監控模塊
├── 故障管理模塊
├── 產品裝配管理模塊
├── 生產計劃管理模塊
├── 數據集成接口
└── 報表導出模塊

技術棧

后端: Python + FastAPI (RESTful API)
前端: Vue.js (可選) 或直接提供API供上層系統調用
數據庫: PostgreSQL (關系型) + Redis (緩存)
任務隊列: Celery (異步任務處理)
文件處理: openpyxl (Excel操作)
監控: Prometheus + Grafana (可選)

核心代碼實現

1. 主應用框架 (main.py)

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from .routers import devices, faults, products, productionapp = FastAPI(title="數字化生產管理系統", version="1.0.0")# 允許跨域
app.add_middleware(CORSMiddleware,allow_origins=["*"],allow_credentials=True,allow_methods=["*"],allow_headers=["*"],
)# 包含各模塊路由
app.include_router(devices.router)
app.include_router(faults.router)
app.include_router(products.router)
app.include_router(production.router)@app.get("/")
async def root():return {"message": "數字化生產管理系統 API"}

2. 設備監控模塊 (devices.py)

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from typing import List
from datetime import datetime
import psycopg2
import redisrouter = APIRouter(prefix="/devices", tags=["設備管理"])# Redis連接
r = redis.Redis(host='localhost', port=6379, db=0)# 數據庫連接配置
DB_CONFIG = {"host": "localhost","database": "production_db","user": "postgres","password": "password"
}class Device(BaseModel):id: strname: strstatus: strlast_heartbeat: datetimeip_address: str@router.get("/", response_model=List[Device])
async def get_online_devices():"""獲取所有在線設備狀態"""conn = psycopg2.connect(**DB_CONFIG)cursor = conn.cursor()cursor.execute("SELECT * FROM devices WHERE status = 'online'")devices = cursor.fetchall()cursor.close()conn.close()return [Device(id=d[0], name=d[1], status=d[2], last_heartbeat=d[3], ip_address=d[4]) for d in devices]@router.get("/{device_id}/status")
async def get_device_status(device_id: str):"""獲取特定設備狀態"""# 先從Redis緩存獲取status = r.get(f"device:{device_id}:status")if status:return {"device_id": device_id, "status": status.decode()}# 緩存沒有則查數據庫conn = psycopg2.connect(**DB_CONFIG)cursor = conn.cursor()cursor.execute("SELECT status FROM devices WHERE id = %s", (device_id,))result = cursor.fetchone()cursor.close()conn.close()if not result:raise HTTPException(status_code=404, detail="Device not found")# 更新緩存r.setex(f"device:{device_id}:status", 30, result[0])return {"device_id": device_id, "status": result[0]}

3. 故障管理模塊 (faults.py)

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from typing import List
from datetime import datetime
import psycopg2router = APIRouter(prefix="/faults", tags=["故障管理"])class Fault(BaseModel):id: intdevice_id: strfault_code: strdescription: strtimestamp: datetimeresolved: boolclass FaultCreate(BaseModel):device_id: strfault_code: strdescription: str@router.post("/", response_model=Fault)
async def report_fault(fault: FaultCreate):"""報告設備故障"""conn = psycopg2.connect(**DB_CONFIG)cursor = conn.cursor()cursor.execute("INSERT INTO faults (device_id, fault_code, description, timestamp, resolved) ""VALUES (%s, %s, %s, %s, %s) RETURNING id",(fault.device_id, fault.fault_code, fault.description, datetime.now(), False))fault_id = cursor.fetchone()[0]conn.commit()# 更新設備狀態為故障cursor.execute("UPDATE devices SET status = 'fault' WHERE id = %s",(fault.device_id,))conn.commit()cursor.close()conn.close()return {"id": fault_id,**fault.dict(),"timestamp": datetime.now(),"resolved": False}@router.get("/unresolved", response_model=List[Fault])
async def get_unresolved_faults():"""獲取未解決的故障列表"""conn = psycopg2.connect(**DB_CONFIG)cursor = conn.cursor()cursor.execute("SELECT * FROM faults WHERE resolved = FALSE")faults = cursor.fetchall()cursor.close()conn.close()return [Fault(id=f[0], device_id=f[1], fault_code=f[2], description=f[3], timestamp=f[4], resolved=f[5]) for f in faults]

4. 產品裝配管理模塊 (products.py)

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from typing import List
import psycopg2
from openpyxl import Workbook
from io import BytesIO
from fastapi.responses import StreamingResponserouter = APIRouter(prefix="/products", tags=["產品裝配管理"])class Component(BaseModel):code: strname: strquantity: intclass ProductAssembly(BaseModel):product_id: strproduct_name: strcomponents: List[Component]@router.get("/{product_id}/assembly", response_model=ProductAssembly)
async def get_product_assembly(product_id: str):"""獲取產品裝配配套表"""conn = psycopg2.connect(**DB_CONFIG)cursor = conn.cursor()# 獲取產品信息cursor.execute("SELECT name FROM products WHERE id = %s", (product_id,))product = cursor.fetchone()if not product:raise HTTPException(status_code=404, detail="Product not found")# 獲取零部件列表cursor.execute("SELECT c.code, c.name, pc.quantity ""FROM product_components pc ""JOIN components c ON pc.component_id = c.id ""WHERE pc.product_id = %s", (product_id,))components = cursor.fetchall()cursor.close()conn.close()return ProductAssembly(product_id=product_id,product_name=product[0],components=[Component(code=c[0], name=c[1], quantity=c[2]) for c in components])@router.get("/{product_id}/assembly/excel")
async def export_product_assembly_excel(product_id: str):"""導出產品裝配配套表為Excel"""assembly = await get_product_assembly(product_id)# 創建Excel文件wb = Workbook()ws = wb.activews.title = "裝配配套表"# 寫入標題ws.append(["產品編號", assembly.product_id])ws.append(["產品名稱", assembly.product_name])ws.append([])ws.append(["零部件代號", "零部件名稱", "數量"])# 寫入零部件數據for comp in assembly.components:ws.append([comp.code, comp.name, comp.quantity])# 保存到內存excel_file = BytesIO()wb.save(excel_file)excel_file.seek(0)# 返回文件下載return StreamingResponse(excel_file,media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",headers={"Content-Disposition": f"attachment; filename=assembly_{product_id}.xlsx"})

5. 生產計劃管理模塊 (production.py)

from fastapi import APIRouter, HTTPException, UploadFile, File
from pydantic import BaseModel
from typing import List
from datetime import date
import psycopg2
import openpyxl
import csv
import iorouter = APIRouter(prefix="/production", tags=["生產計劃管理"])class MasterPlan(BaseModel):id: strproduct_id: strquantity: intstart_date: dateend_date: datepriority: intclass WorkshopPlan(BaseModel):id: strmaster_plan_id: strworkshop_id: strproduct_id: strquantity: intplanned_start: dateplanned_end: dateactual_start: date = Noneactual_end: date = Nonestatus: str = "pending"  # pending, in_progress, completed@router.post("/generate-workshop-plans")
async def generate_workshop_plans(master_plan_id: str):"""根據主生產計劃生成車間級生產計劃"""conn = psycopg2.connect(**DB_CONFIG)cursor = conn.cursor()# 獲取主生產計劃cursor.execute("SELECT * FROM master_production_plans WHERE id = %s", (master_plan_id,))master_plan = cursor.fetchone()if not master_plan:raise HTTPException(status_code=404, detail="Master plan not found")# 這里應該有更復雜的邏輯來分配車間任務# 簡化為直接為每個車間創建相同的任務cursor.execute("SELECT id FROM workshops")workshops = cursor.fetchall()plans = []for workshop in workshops:cursor.execute("INSERT INTO workshop_plans ""(master_plan_id, workshop_id, product_id, quantity, planned_start, planned_end, status) ""VALUES (%s, %s, %s, %s, %s, %s, %s) RETURNING id",(master_plan_id,workshop[0],master_plan[1],  # product_idmaster_plan[2],  # quantitymaster_plan[3],  # start_datemaster_plan[4],  # end_date"pending"))plan_id = cursor.fetchone()[0]plans.append(plan_id)conn.commit()cursor.close()conn.close()return {"message": f"Generated {len(plans)} workshop plans", "plan_ids": plans}@router.post("/import-workshop-plans/excel")
async def import_workshop_plans_excel(file: UploadFile = File(...)):"""通過Excel批量導入車間生產計劃"""contents = await file.read()wb = openpyxl.load_workbook(io.BytesIO(contents))ws = wb.activeplans = []for row in ws.iter_rows(min_row=2, values_only=True):  # 假設第一行是標題if not row[0]:  # 跳過空行continueplan = {"master_plan_id": row[0],"workshop_id": row[1],"product_id": row[2],"quantity": row[3],"planned_start": row[4],"planned_end": row[5]}plans.append(plan)# 保存到數據庫conn = psycopg2.connect(**DB_CONFIG)cursor = conn.cursor()for plan in plans:cursor.execute("INSERT INTO workshop_plans ""(master_plan_id, workshop_id, product_id, quantity, planned_start, planned_end, status) ""VALUES (%s, %s, %s, %s, %s, %s, %s)",(plan["master_plan_id"],plan["workshop_id"],plan["product_id"],plan["quantity"],plan["planned_start"],plan["planned_end"],"pending"))conn.commit()cursor.close()conn.close()return {"message": f"Imported {len(plans)} workshop plans successfully"}@router.get("/workshop-plans/{workshop_id}", response_model=List[WorkshopPlan])
async def get_workshop_plans(workshop_id: str):"""獲取車間生產計劃及執行情況"""conn = psycopg2.connect(**DB_CONFIG)cursor = conn.cursor()cursor.execute("SELECT id, master_plan_id, workshop_id, product_id, quantity, ""planned_start, planned_end, actual_start, actual_end, status ""FROM workshop_plans WHERE workshop_id = %s",(workshop_id,))plans = cursor.fetchall()cursor.close()conn.close()return [WorkshopPlan(id=p[0],master_plan_id=p[1],workshop_id=p[2],product_id=p[3],quantity=p[4],planned_start=p[5],planned_end=p[6],actual_start=p[7],actual_end=p[8],status=p[9]) for p in plans]

6. 數據庫模型 (SQL)

-- 設備表
CREATE TABLE devices (id VARCHAR(50) PRIMARY KEY,name VARCHAR(100) NOT NULL,status VARCHAR(20) NOT NULL CHECK (status IN ('online', 'offline', 'fault', 'maintenance')),last_heartbeat TIMESTAMP,ip_address VARCHAR(50)
);-- 故障表
CREATE TABLE faults (id SERIAL PRIMARY KEY,device_id VARCHAR(50) REFERENCES devices(id),fault_code VARCHAR(50) NOT NULL,description TEXT,timestamp TIMESTAMP NOT NULL,resolved BOOLEAN NOT NULL DEFAULT FALSE
);-- 產品表
CREATE TABLE products (id VARCHAR(50) PRIMARY KEY,name VARCHAR(100) NOT NULL
);-- 零部件表
CREATE TABLE components (id VARCHAR(50) PRIMARY KEY,code VARCHAR(50) NOT NULL,name VARCHAR(100) NOT NULL
);-- 產品-零部件關聯表
CREATE TABLE product_components (product_id VARCHAR(50) REFERENCES products(id),component_id VARCHAR(50) REFERENCES components(id),quantity INTEGER NOT NULL,PRIMARY KEY (product_id, component_id)
);-- 主生產計劃表
CREATE TABLE master_production_plans (id VARCHAR(50) PRIMARY KEY,product_id VARCHAR(50) REFERENCES products(id),quantity INTEGER NOT NULL,start_date DATE NOT NULL,end_date DATE NOT NULL,priority INTEGER NOT NULL
);-- 車間表
CREATE TABLE workshops (id VARCHAR(50) PRIMARY KEY,name VARCHAR(100) NOT NULL
);-- 車間生產計劃表
CREATE TABLE workshop_plans (id VARCHAR(50) PRIMARY KEY,master_plan_id VARCHAR(50) REFERENCES master_production_plans(id),workshop_id VARCHAR(50) REFERENCES workshops(id),product_id VARCHAR(50) REFERENCES products(id),quantity INTEGER NOT NULL,planned_start DATE NOT NULL,planned_end DATE NOT NULL,actual_start DATE,actual_end DATE,status VARCHAR(20) NOT NULL CHECK (status IN ('pending', 'in_progress', 'completed', 'delayed'))
);