数字化生产管理系统设计

下面提供一个基于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'))
);

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如若转载，请注明出处：http://www.pswp.cn/diannao/94089.shtml
繁体地址，请注明出处：http://hk.pswp.cn/diannao/94089.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

【vue】创建响应式数据ref和reactive的区别

【vue】创建响应式数据ref和reactive的区别

目录 1、所谓响应式数据 2、ref创建基本类型响应式数据 3、reactive创建对象类型响应式数据 4、ref定义对象类型响应式数据 5、总结：ref和reactive对比 6、补充：toRefs与toRef 1、所谓响应式数据 所谓响应式数据就是，在vue页面中&#…
阅读更多...
YOLO12 改进、魔改｜直方图 Transformerm模块HTB ，通过动态范围特征分组、针对性注意力与多尺度融合，提高对遮挡以及多尺度目标的关注能力

YOLO12 改进、魔改｜直方图 Transformerm模块HTB ，通过动态范围特征分组、针对性注意力与多尺度融合，提高对遮挡以及多尺度目标的关注能力

在恶劣天气（如雨、雪、雾）下的图像恢复任务中，传统 Transformer 模型为降低计算量，常将自注意力限制在固定空间范围或仅在通道维度操作，导致难以捕捉长距离空间特征，尤其无法有效处理天气退化像素&#xff…
阅读更多...
涉水救援机器人cad【12张】三维图＋设计书明说

涉水救援机器人cad【12张】三维图＋设计书明说

涉水救援机器人设计 摘 要 随着城市化进度的加快，各种水上游乐设备的增多，水上灾害和溺水事件频繁发生，水上救援任务困难重重，特别是在一些水流湍急的环境下进行救援。传统的水上救援工作比较缓慢，大多数是通过投射救…
阅读更多...
电子病历：现代HIS系统不可或缺的核心组成部分

电子病历：现代HIS系统不可或缺的核心组成部分

一、电子病历在 HIS 系统中扮演了一个什么角色?电子病历在医院信息系统（HIS）中扮演着核心、基础且不可替代的角色，可以说是整个HIS系统的临床信息中枢和业务驱动引擎。它的重要性体现在以下几个方面：1、临床诊疗活动的核心载体&a…
阅读更多...
【深度学习】通俗易懂的基础知识：指数加权平均

【深度学习】通俗易懂的基础知识：指数加权平均

一、什么是指数加权平均？ 指数在数学中表示一个数的幂次运算（如aⁿ中的n），而在统计学中特指随时间变化的几何衰减系数，加权指对不同数据赋予不同权重，使重要数据对结果产生更大影响。指数加权平均指是一种时…
阅读更多...
c++-list

c++-list

C-list std::list是C标准模板库(STL)提供的双向链表容器&#xff0c;它提供了高效的插入和删除操作&#xff0c;特别适合频繁修改的序列。定义在 <list> 头文件中&#xff0c;属于 std 命名空间。该类的接口与常规容器接口基本一致。 模板原型&#xff1a; template <…
阅读更多...
【笔试真题】2024秋招京东后端开发岗位-第一批笔试

【笔试真题】2024秋招京东后端开发岗位-第一批笔试

31.牛牛与切割机 有一个序列 a1,a2,...,ana_1,a_2,...,a_na1​,a2​,...,an​ &#xff0c; 牛牛将对这个序列切割一刀&#xff08;划分分成两个不相交的非空序列&#xff0c;一个序列为 a1,...,apa_1,...,a_pa1​,...,ap​&#xff0c;另一个序列为 ap1,...,ana_{p1},...,a_na…
阅读更多...
【整数转罗马数字】

【整数转罗马数字】

思路计算数字的位数&#xff1a; 通过 while(x) 循环计算输入数字 num 的位数 n。提取各位数字&#xff1a; 将数字 num 的每一位分解并存储到 nums 数组中&#xff0c;顺序为从高位到低位。罗马数字映射&#xff1a; 使用固定数组 Roman 存储罗马数字符号&#xff1a;Roman {…
阅读更多...
spring Scheduled注解详解

spring Scheduled注解详解

spirng Scheduled注解详解 用于标记需要安排执行的方法的注解。必须指定 cron、fixedDelay 或 fixedRate 中的恰好一个属性。 被标注的方法必须不接受任何参数。它通常会具有 void 类型的返回值&#xff1b;如果不是这样&#xff0c;那么在通过调度器调用该方法时&#xff0c;返…
阅读更多...
新升级超值型系列32位单片机MM32G0005

新升级超值型系列32位单片机MM32G0005

灵动微推出的新型MM32G0005系列基于ArmCortex - M0内核&#xff0c;具备高可靠性、低功耗、高性价比等特性。Flash升级至64KB&#xff0c;SRAM为4KB&#xff0c;还有1KB Data Flash。Flash全温擦写次数超过10万次。采用24Pin封装&#xff0c;最多有22个IO。QFN20和TSSOP20封装与…
阅读更多...
Spark SQL 的详细介绍

Spark SQL 的详细介绍

Spark SQL 是 Apache Spark 生态系统中用于处理结构化数据的模块&#xff0c;它将 SQL 查询与 Spark 的分布式计算能力相结合&#xff0c;提供了一种高效、灵活的方式来处理结构化和半结构化数据。以下是对 Spark SQL 的详细介绍&#xff1a;1. 核心定位与优势结构化数据处理&a…
阅读更多...
【FreeRTOS】空闲任务与钩子函数原理、实现与功能详解

【FreeRTOS】空闲任务与钩子函数原理、实现与功能详解

一、FreeRTOS空闲任务概述FreeRTOS中的空闲任务(Idle Task)是系统自动创建的一个特殊任务&#xff0c;具有最低优先级(优先级0)。当没有其他更高优先级的任务运行时&#xff0c;调度器就会运行空闲任务。空闲任务的主要功能系统资源回收&#xff1a;自动清理被删除任务的内存和…
阅读更多...
imx6ull-驱动开发篇6——Linux 设备树语法

imx6ull-驱动开发篇6——Linux 设备树语法

目录 前言 设备树 设备树概念 DTS、 DTB 和 DTC DTS 语法 .dtsi 头文件 设备节点 /根节点​​ 节点命名与标签 节点层次结构​ 属性数据类型​ 标准属性 compatible 属性 model 属性 status 属性 #address-cells 和#size-cells 属性 reg 属性 ranges 属性 n…
阅读更多...
ansible简单playbook剧本例子2

ansible简单playbook剧本例子2

1. 准备主机组[rootansible-master ansible_quickstart]# vim inventory/hosts[web:vars] ansible_port22 ansible_passwordAdmin123456[web] 192.168.100.1822.准备剧本 vim hello.yml--- - hosts: webremote_user: roottasks:- name: Ping the target hostsping:- name: 获取…
阅读更多...
EmpService 和 EmpMapper接口的作用

EmpService 和 EmpMapper接口的作用

在这个项目中&#xff0c;EmpService 和 EmpMapper 都定义接口&#xff0c;是基于面向接口编程&#xff08;Interface Oriented Programming&#xff0c;IOP&#xff09;的设计思想&#xff0c;这两种接口在项目中承担着不同的职责&#xff0c;具体说明如下&#xff1a; EmpSer…
阅读更多...
【语音技术】什么是动态实体

【语音技术】什么是动态实体

目录 动态实体的定义和维度 1.1 动态实体的资源 1.2 生效维度 1.2.1 应用级 1.2.2 用户级 1.2.3 自定义级 2. 动态实体的上传及使用 2.1 WebAPI 2.1.1 授权认证 2.1.2 上传资源接口 2.1.2.1 参数说明 2.1.2.2 返回说明 2.1.3 查询打包状态 2.1.3.1 参数说明 2.1.…
阅读更多...
STM32学习记录--Day3

STM32学习记录--Day3

今天了解了下I2C&#xff1a;1.I2C电路结构I2C通信示意图&#xff1a;数据传输阶段​​​​主→从模式​​&#xff08;写操作&#xff09;&#xff1a;主机控制SCL时钟&#xff08;把SCL拉低&#xff09;主机向SDA线发送数据&#xff08;每次8位1位ACK&#xff09;​​主←从模…
阅读更多...
裂变数据看板：5个核心指标决定活动生死​

裂变数据看板：5个核心指标决定活动生死​

数据是裂变活动的“指南针”。本文详解曝光量、转化率、裂变系数等5大核心指标&#xff0c;结合工具与案例&#xff0c;教你用数据驱动活动优化&#xff0c;避免“自嗨式裂变”。​为什么数据是裂变的“生死线”&#xff1f;&#xff08;认知重构&#xff09; 很多企业裂变活动…
阅读更多...
iOS 类存储 与 C# 类存储 的差异

iOS 类存储 与 C# 类存储 的差异

C# 中类的代码&#xff08;包括方法、属性等成员&#xff09;的存储机制与 Objective-C 有显著差异&#xff0c;其核心依赖于 ​CLR&#xff08;公共语言运行时&#xff09;的方法表&#xff08;Method Table&#xff09;和虚拟方法表&#xff08;vtable&#xff09;机制&#…
阅读更多...
Selenium自动化：轻松实现网页操控

Selenium自动化：轻松实现网页操控

selenium自动化 1 什么是 Selenium 自动化 Selenium 是一个用于 Web 应用程序测试的工具&#xff0c;支持多种浏览器&#xff08;如 Chrome、Firefox、Edge 等&#xff09;。WebDriver 是 Selenium 的核心组件&#xff0c;用于控制浏览器行为并执行自动化操作。元素定位是通过…
阅读更多...
最新文章

Copyright @ 2022~2023