从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)
Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐