从零到一:我的第一个Python完整项目——智能设备巡检管理系统
·
从6月初开始学Python,到7月初完成第一个完整项目。两个月的自学成果,都浓缩在这9个模块里了。
项目做什么的?
一个命令行版的设备巡检管理系统,用于工厂或物业场景。可以管理设备信息、执行巡检任务、记录异常情况、生成统计图表。
用了哪些技术?
面向对象:Device基类 + Sensor/Camera子类,继承和多态
生成器:批量生成模拟设备数据,惰性加载节省内存
迭代器:自定义迭代器,支持按设备类型筛选遍历
闭包:实现告警计数器,统计数据外部不可篡改
装饰器:操作日志系统,自动记录每次操作的时间和函数名
深浅拷贝:设备清单导出时保护原始数据
Pandas:设备数据的统计分析
Matplotlib:自动生成饼状图和柱状图
开发过程印象最深的坑
私有属性 __purchase_price 传参时和名称改写机制冲突,学会了参数名和私有属性名要分开
生成器只能遍历一次,list(generator) 解决了这个问题
InspectionRecord 的参数名从 result 改成 status,整个项目统一了命名
收获
从第一行代码到项目跑通,最大的收获不是学会了某个技术点,而是从零设计一个完整系统的能力——怎么拆分模块、怎么命名、怎么迭代优化。
'''device_model.py'''
class Device:
"""设备基类,包含所有设备的公共属性"""
def __init__(self, device_id, device_name, location, device_type,
purchase_date, inspection_cycle, status, purchase_price):
"""初始化设备基类"""
self.device_id = device_id
self.device_name = device_name
self.location = location
self.device_type = device_type
self.purchase_date = purchase_date
self.inspection_cycle = inspection_cycle
self.status = status
self.__purchase_price = purchase_price
def get_info(self):
"""返回设备信息的字典格式"""
return {
"设备编号": self.device_id,
"设备名称": self.device_name,
"安装位置": self.location,
"设备类型": self.device_type,
"购买日期": self.purchase_date,
"巡检周期(天)": self.inspection_cycle,
"状态": self.status
}
def __str__(self):
"""设备的友好展示"""
info = self.get_info()
return (f"设备编号:{info['设备编号']} 设备名称:{info['设备名称']} "
f"安装位置:{info['安装位置']} 设备类型:{info['设备类型']} "
f"购买日期:{info['购买日期']} 巡检周期(天):{info['巡检周期(天)']} "
f"状态:{info['状态']}")
class Sensor(Device):
"""传感器子类,继承自Device"""
def __init__(self, device_id, device_name, location, device_type,
purchase_date, inspection_cycle, status, purchase_price,
measure_param, precision_level, measure_range):
super().__init__(device_id, device_name, location, device_type,
purchase_date, inspection_cycle, status, purchase_price)
self.measure_param = measure_param
self.precision_level = precision_level
self.measure_range = measure_range
def __str__(self):
base_str = super().__str__()
return (f"{base_str} 测量参数:{self.measure_param} "
f"精度等级:{self.precision_level} "
f"量程范围:{self.measure_range}")
class Camera(Device):
"""摄像头子类,继承自Device"""
def __init__(self, device_id, device_name, location, device_type,
purchase_date, inspection_cycle, status, purchase_price,
resolution, storage_capacity, night_vision):
super().__init__(device_id, device_name, location, device_type,
purchase_date, inspection_cycle, status, purchase_price)
self.resolution = resolution
self.storage_capacity = storage_capacity
self.night_vision = night_vision
def __str__(self):
base_str = super().__str__()
return (f"{base_str} 分辨率:{self.resolution} "
f"存储容量:{self.storage_capacity} 支持夜视:{self.night_vision}")
class InspectionRecord:
"""巡检记录类"""
def __init__(self, record_device_id, inspector, inspection_date,
status, is_resolved, fault_desc=None):
self.record_device_id = record_device_id
self.inspector = inspector
self.inspection_date = inspection_date
self.status = status
self.fault_desc = fault_desc
if is_resolved:
self.is_resolved = '已处理'
else:
self.is_resolved = '未处理'
def __str__(self):
return (f"设备编号:{self.record_device_id} 巡检人:{self.inspector} "
f"巡检日期:{self.inspection_date} 巡检结果:{self.status} "
f"故障描述:{self.fault_desc} 是否已处理:{self.is_resolved}")
if __name__ == '__main__':
# 测试代码
device = Device("DEV001", "温度传感器", "A车间-01", "传感器",
"2024-01-15", 7, "正常", 1500)
sensor = Sensor("DEV002", "压力传感器", "B车间-02", "传感器",
"2024-02-20", 7, "正常", 2000,
"压力", "A级", "0-100MPa")
camera = Camera("DEV003", "监控摄像头", "C车间-入口", "摄像头",
"2024-03-10", 15, "正常", 5000,
"1080P", "256GB", True)
record = InspectionRecord("DEV001", "张三", "2024-06-15", "异常", False, "温度偏高")
print(device)
print(sensor)
print(camera)
print(record)
'''device_generator.py'''
import random
from device_model import Device, Sensor, Camera
#设备名称列表
DEVICE_NAMES = ['温度传感器', '压力传感器', '流量计', '监控摄像头', '红外摄像头', '湿度传感器']
#安装位置列表
LOCATIONS = ['A车间-01', 'B车间-02', 'C车间-入口', 'D仓库-01', '办公楼-大厅', '配电房-01']
#巡检周期
INSPECTION_CYCLES = [7, 15, 30]
#Camera
#分辨率
RESOLUTIONS = ['720P', '1080P', '4K']
#存储内容
STORAGE_CAPACITIES = ['128GB', '256GB', '512GB']
#支持夜视
NIGHT_VISION = [True, False]
#Sensor
#测量参数
SENSOR_PARAMS = ['温度', '压力', '湿度', '流量', '液位']
#精度等级
PRECISION_LEVELS = ['A级', 'B级', 'C级']
#量程范围
MEASURE_RANGES = ['0-50', '0-100', '0-200', '0-500']
def _generate_common_attrs(device_index):
"""生成所有设备的公共属性,返回一个字典"""
# 随机状态
rand = random.random()
if rand < 0.8:
status = '正常'
elif rand < 0.95:
status = '异常'
else:
status = '已报废'
return {
'device_id': f"DEV{device_index:03d}",
'device_name': random.choice(DEVICE_NAMES),
'location': random.choice(LOCATIONS),
'purchase_date': f"{random.randint(2022, 2026)}-{random.randint(1, 12):02d}-{random.randint(1, 28):02d}",
'inspection_cycle': random.choice(INSPECTION_CYCLES),
'status': status,
'purchase_price': random.randint(1000, 10000)
}
def generate_devices(count):
for i in range(count):
attrs = _generate_common_attrs(i + 1)
#在这里生成一个随机设备
if random.random() <0.5:
yield Sensor(device_id=attrs['device_id'],
device_name=attrs['device_name'],
location=attrs['location'],
device_type='传感器',
purchase_date=attrs['purchase_date'],
inspection_cycle=attrs['inspection_cycle'],
status=attrs['status'],
purchase_price=attrs['purchase_price'],
measure_param=random.choice(SENSOR_PARAMS),
precision_level=random.choice(PRECISION_LEVELS),
measure_range=random.choice(MEASURE_RANGES))
else:
yield Camera(device_id=attrs['device_id'], device_name=attrs['device_name'], location=attrs['location'], device_type='摄像头',
purchase_date=attrs['purchase_date'], inspection_cycle=attrs['inspection_cycle'], status=attrs['status'], purchase_price=attrs['purchase_price'],
resolution=random.choice(RESOLUTIONS), storage_capacity=random.choice(STORAGE_CAPACITIES), night_vision=random.choice(NIGHT_VISION))
if __name__ == '__main__':
# 测试:生成5个设备并打印
for device in generate_devices(5):
print(device)
print('-' * 50)
'''device_iterator.py'''
from device_generator import generate_devices
class DeviceIterator:
def __init__(self, devices,device_filter_type=None):
self.devices = devices
self.device_filter_type = device_filter_type
self.index = 0
def __iter__(self):
return self
def __next__(self):
while self.index < len(self.devices):
self.device = self.devices[self.index]
self.index += 1
if self.device_filter_type is None or self.device.device_type == self.device_filter_type:
return self.device
else:
continue
raise StopIteration
if __name__ == "__main__":
# 生成设备列表
device = list(generate_devices(3))
print("所有设备:")
for device in DeviceIterator(device, device_filter_type="传感器"):
print(device)
'''inspection_manager.py'''
import device_model as dm
from operation_logger import log_operation
import copy as cp
class InspectionManager:
def __init__(self):
self.devices = []
self.records = []
self.device_types = set()
@log_operation
def add_device(self, device):
"""添加设备"""
self.devices.append(device)
self.device_types.add(device.device_type)
@log_operation
def remove_device(self, device_id):
"""根据设备ID移除设备"""
self.devices = [d for d in self.devices if d.device_id != device_id]
def get_devices(self, device_type=None):
"""获取设备列表,可按类型过滤"""
if device_type:
return [d for d in self.devices if d.device_type == device_type]
return self.devices
def filter_by_status(self,status):
"""按状态过滤设备"""
return [d for d in self.devices if d.status == status]
@log_operation
def perform_inspection(self,record_device_id,status,inspector,inspection_date,fault_desc):
"""执行巡检操作,更新设备状态并记录"""
for d in self.devices:
if d.device_id == record_device_id:
d.status = status
record = dm.InspectionRecord(record_device_id=record_device_id,
inspector=inspector,
inspection_date=inspection_date,
status=status,
is_resolved=('正常' == status),
fault_desc=fault_desc)
self.records.append(record)
return record
def get_statistics(self):
"""获取设备统计信息"""
all_devices = self.get_devices()
status = {
'设备总数':len(all_devices),
'正常':len([d for d in all_devices if d.status == '正常']),
'异常':len([d for d in all_devices if d.status == '异常']),
'已报废':len([d for d in all_devices if d.status == '已报废'])
}
for device_type in self.device_types:
status[device_type] = len([d for d in all_devices if d.device_type == device_type])
return status
def export_devices(self):
"""导出设备信息到文件"""
return cp.deepcopy(self.devices)
if __name__ == "__main__":
manager = InspectionManager()
device2 = dm.Device(device_id="D002",
device_name="摄像头B",
location="车间2",
device_type="摄像头",
purchase_date="2023-02-01",
inspection_cycle=60,
status="异常",
purchase_price=2000)
manager.add_device(device2)
print("所有设备:")
for d in manager.get_devices():
print(d)
print('-' * 50)
print("异常设备:")
for d in manager.filter_by_status('异常'):
print(d)
print('-' * 50)
'''operation_logger.py'''
import datetime as dt
def log_operation(func):
def wrapper(*args, **kwargs):
# 1. 打印日志
print(f"[{dt.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 执行操作:{func.__name__}")
# 2. 执行原函数
result = func(*args, **kwargs)
# 3. 返回原函数的结果
return result
return wrapper
'''config.py'''
import utils
# 公司/工厂的固定信息:公司名称、工厂地址、联系电话、巡检时间范围
COMPANY_INFO = (
"公司名称:科技有限公司",
"工厂地址:湖南省长沙市开福区",
"联系电话:0731-000000",
"巡检时间范围:08:00 - 18:00"
)
def print_config():
"""格式化打印公司/工厂信息"""
utils.print_section("公司/工厂信息:")
for info in COMPANY_INFO:
print(info)
utils.print_section("")
if __name__ == '__main__':
print_config()
'''utils.py'''
import matplotlib.pyplot as plt
def print_section(title):
print("\n" + "=" * 40)
print(title)
print("=" * 40 + "\n")
def plot_status_pie(stats):
plt.rcParams['font.sans-serif'] = ['simhei']
status_stats = {k: v for k, v in stats.items() if k in ['正常', '异常', '已报废']}
plt.pie(status_stats.values(), labels=status_stats.keys(), autopct='%1.1f%%')
plt.title('设备状态分布')
plt.show()
def plot_type_bar(stats):
plt.rcParams['font.sans-serif'] = ['simhei']
type_stats = {k: v for k, v in stats.items() if k not in ['设备总数', '正常', '异常', '已报废']}
plt.figure()
plt.bar(type_stats.keys(), type_stats.values())
plt.title('设备类型数量')
plt.xlabel('类型')
plt.ylabel('数\n量\n',rotation='horizontal')
plt.show()
'''main.py'''
import alert_counter as ac
from config import print_config
import device_generator as dg
import device_iterator as di
import device_model as dm
import inspection_manager as im
import operation_logger as ol
import utils as us
import random
if __name__ == "__main__":
print_config()
manager = im.InspectionManager()
for device in dg.generate_devices(5):
manager.add_device(device)
print(device)
print('-' * 50)
us.print_section("所有设备:")
for d in di.DeviceIterator(manager.get_devices()):
print(d)
print('-' * 50)
us.print_section("传感器设备:")
for d in di.DeviceIterator(manager.get_devices(), device_filter_type="传感器"):
print(d)
print('-' * 50)
us.print_section("异常设备:")
for m in manager.filter_by_status('异常'):
print(m)
print('-' * 50)
us.print_section("执行巡检操作:")
for i in manager.get_devices()[:3]:
status = random.choice(['正常', '异常', '已报废'])
manager.perform_inspection(record_device_id=i.device_id, inspector='张三', inspection_date='2024-01-01',status=status, fault_desc='无')
us.print_section("巡检记录:")
for record in manager.records:
print(record)
print('-' * 50)
us.print_section("设备统计信息:")
manager_stats = manager.get_statistics()
for key, value in manager_stats.items():
print(f"{key}: {value}")
us.plot_status_pie(manager_stats)
us.plot_type_bar(manager_stats)
更多推荐


所有评论(0)