简介这是一套基于 Python Django 框架开发的流量计远程抄表管理系统完整源码包适合计算机、数学、电子信息等专业的课程设计、期末大作业或毕业设计参考。系统围绕远程抄表场景构建后端以 Python 文件为主辅以 JavaScript、HTML 实现前端交互与页面展示CSS 负责样式布局JSON/TXT 等文件用于配置与说明整体结构清晰便于二次开发。压缩包共含 2000 个文件大小约 37.96MB内置项目说明文档下载后可直接运行帮助读者理解 Django 项目从模型设计到视图、模板的完整实现路径。目前已有 203 人学习浏览适合具备一定 Python 基础、希望快速搭建同类管理系统的开发者作为实战蓝本也可作为论文写作与答辩前的功能演示样例。1. 流量计远程抄表管理系统其实是一条数据链路而不是一个页面“流量计远程抄表管理系统”这个标题里真正费功夫的不是 Django 管理页面而是把流量计的数稳定地拿到数据库里。流量计现场的通讯方式大多是 Modbus RTU 或 Modbus TCP前者走 RS-485 串口后者走网线Django 负责的是设备建模、历史数据存储、管理后台和查询接口。串口读取、协议解析、定时调度这些环节在 Django 里分别落到 pymodbus、Celery 这些成熟组件上。这套方案和大型 IoT 平台相比少了很多中间件但在几十台到几百台设备的规模内非常实用。下面是按建模、采集、调度、部署四段展开的落地路径适合正在做设备管理系统、又不想从裸 socket 开始写的 Python 工程师参考。2. 设备建模与抄表记录表先决定归档结构再写采集器远程抄表系统最常见的返工原因是把站点、设备和每次抄表数据塞进一张表。现场有多个站点每个站点下有多台流量计每台流量计的串口、波特率、从站地址、寄存器地址又不一样。如果设备信息、通讯参数、读数混在一张宽表里前期跑通容易加仪表型号和按时间范围查曲线时就会很痛苦。合理的建模分三层Site 表示物理位置Device 表示仪表及其通讯参数MeterReading 表示一次采集快照。2.1 Site、Device、MeterReading 三层模型与字段设计设备模型与普通的业务对象有一个明显区别它必须保存“怎么把数读出来”的通讯参数而不是只保存名称和型号。下面这段 models.py 把 RTU 和 TCP 两种模式的连接参数放在同一张表里用 mode 字段区分采集任务取到一条记录后就能直接决定用哪种客户端。# meter/models.py from django.db import models class Site(models.Model): name models.CharField(max_length64, uniqueTrue) location models.CharField(max_length128, blankTrue) created_at models.DateTimeField(auto_now_addTrue) def __str__(self): return self.name class Device(models.Model): MODE_CHOICES ((rtu, Modbus RTU), (tcp, Modbus TCP)) site models.ForeignKey(Site, on_deletemodels.CASCADE, related_namedevices) name models.CharField(max_length64) slave_id models.PositiveSmallIntegerField(default1, verbose_name从站地址) mode models.CharField(max_length8, choicesMODE_CHOICES, defaultrtu) # Modbus RTU 参数 serial_port models.CharField(max_length16, blankTrue, help_text如 /dev/ttyUSB0) baudrate models.PositiveIntegerField(default9600) # Modbus TCP 参数 host models.GenericIPAddressField(nullTrue, blankTrue, verbose_name设备 IP) port models.PositiveIntegerField(default502) # 寄存器映射 register_flow models.PositiveIntegerField(default0, help_text瞬时流量寄存器地址) register_total models.PositiveIntegerField(default2, help_text累积量寄存器地址) class Meta: ordering [site, name] def __str__(self): return f{self.site.name} - {self.name}把通讯参数直接挂在 Device 上有一个直接好处采集任务只需要拿到一个 device_id就能取出连接方式、波特率、从站号、寄存器地址。slave_id 在 Modbus RTU 里叫从站地址在 Modbus TCP 协议里叫单元标识取值范围都是 1 到 247如果现场有多台表挂同一条 RS-485 总线这个字段就是区分它们的唯一依据。serial_port 和 host 按 mode 二选一填写不能两个都留空后面写采集校验时会用到这个约束。2.2 抄表记录采用快照行而不是一指标一行采集数据每分钟写入一次MeterReading 表会持续膨胀。常见设计是一次采集生成一行记录每个指标是表里的一个字段。这样查询某台设备某几天的瞬时流量曲线时只需要 filter 一个设备 ID 和时间区间如果拆成一指标一行还要先做行转列查询代码会明显变复杂。# meter/models.py 续 class MeterReading(models.Model): device models.ForeignKey(Device, on_deletemodels.CASCADE, related_namereadings) read_at models.DateTimeField(db_indexTrue) instantaneous_flow models.FloatField(nullTrue, blankTrue, verbose_name瞬时流量) total_flow models.FloatField(nullTrue, blankTrue, verbose_name累积流量) raw_data models.JSONField(defaultdict, blankTrue) class Meta: db_table meter_reading ordering [-read_at] indexes [ models.Index(fields[device, read_at]), ]瞬时流量和累积流量是两个最常见的指标单独建列。压力、温度、信号强度这些可能按仪表型号增减的数据放进 raw_data JSONField避免每次换一款表就执行一次迁移。这个取舍值得做的时候先想清楚设计方式优点代价一指标一行新增指标不用改表结构查询曲线要聚合代码量增加固定列快照一次采集一行按时间查很快指标变更需要数据库迁移常用列 JSON 混合常用字段支持索引扩展灵活JSON 字段在 MySQL 里条件查询性能有限上表最后一种方案在仪表型号多、协议字段差异大的工厂里最常见。如果系统刚起步只有一台测试表建议先做固定快照列跑两三个月数据量在十几万行时也完全没问题。2.3 Django admin 与 DRF 序列化器的双入口这个系统有两类操作者运维人员要在后台维护设备信息、看是否有异常读数前端大屏或外部系统要通过 HTTP 接口按设备和时间段拉数据。Django admin 可以用最少的代码提供第一个入口DRF 提供第二个入口。# meter/admin.py admin.register(Device) class DeviceAdmin(admin.ModelAdmin): list_display [name, site, mode, slave_id, serial_port, host] list_filter [site, mode] search_fields [name, host] admin.register(MeterReading) class MeterReadingAdmin(admin.ModelAdmin): list_display [device, read_at, instantaneous_flow, total_flow] date_hierarchy read_at list_filter [device__site]date_hierarchy 会在 admin 列表页上方生成一个按日期钻取的控件对抄表系统特别有用。设备数量上来以后找某一天的记录不用翻几十页直接按日期点进去。admin 界面的美化也不一定要装第三方主题先把 list_display、list_filter、search_fields 这三个属性用好可用性已经比默认状态高一大截。接口侧用 ModelSerializer 加上只读来源字段返回给前端的读数里带上设备名避免前端再发一次请求查设备表# meter/serializers.py from rest_framework import serializers from .models import Device, MeterReading class DeviceSerializer(serializers.ModelSerializer): class Meta: model Device fields __all__ class MeterReadingSerializer(serializers.ModelSerializer): device_name serializers.CharField(sourcedevice.name, read_onlyTrue) class Meta: model MeterReading fields [id, device, device_name, read_at, instantaneous_flow, total_flow]视图集用 ReadOnlyModelViewSet 就够了抄表记录对外只读写入口只能由采集任务完成。要加时间范围过滤viewsets 里打开 DjangoFilterBackend 后请求参数直接写?device3read_at__gte2024-01-01即可。3. 用 pymodbus 读流量计Modbus RTU 与 TCP 的参数和坑仪表通讯是整个系统里最贴近硬件的一层。流量计的寄存器表在手册里写得很清楚但真去读的时候会遇到字节序、从站地址偏移、串口占用等问题。把这一层封装成独立的 modbus_service.py不直接和 Django 视图、模型耦合后续换仪表型号时只改这里。3.1 为什么选 pymodbus 而不是自己拼报文Modbus 协议虽然简单但 CRC 校验、功能码封装、异常码处理这些细节手工实现一遍并不划算。pymodbus 把常用功能码封装成 read_holding_registers、read_input_registers、write_register 等方法同步客户端和异步客户端都有本地采集用同步版就足够。安装依赖只需要两条命令离线环境也可以提前下载 whl 包后本地安装pip install pymodbus pyserialpymodbus 的版本差异主要影响导入路径2.x 里客户端在 pymodbus.client.sync3.x 在 pymodbus.client。下面的代码用 3.x 的导入方式如果你手里的源码包锁定的是 2.x把from pymodbus.client import ModbusSerialClient改成from pymodbus.client.sync import ModbusSerialClient即可。3.2 读取瞬时流量与累积流量的最小实现采集函数的核心逻辑只有三步按 mode 创建客户端按寄存器地址读取两个 float 寄存器把寄存器列表解析成浮点数。难的是第三步的字节序处理。# meter/modbus_service.py import struct from pymodbus.client import ModbusSerialClient, ModbusTcpClient def _decode_float(registers): 将两个 16 位寄存器解析为 IEEE754 单精度浮点数。 2H 表示两个无符号短整型按大端对齐之后整体按 float 解析。 raw struct.pack(2H, registers[0], registers[1]) return struct.unpack(f, raw)[0] def read_device(device): if device.mode rtu: client ModbusSerialClient( portdevice.serial_port, baudratedevice.baudrate, timeout3, parityN, stopbits1, bytesize8, ) else: client ModbusTcpClient(hostdevice.host, portdevice.port, timeout3) if not client.connect(): raise ConnectionError(f连接流量计失败: {device.name}) try: flow client.read_holding_registers( addressdevice.register_flow, count2, slavedevice.slave_id ) total client.read_holding_registers( addressdevice.register_total, count2, slavedevice.slave_id ) if flow.isError() or total.isError(): raise IOError(f寄存器读取错误: {device.name}) return { instantaneous_flow: _decode_float(flow.registers), total_flow: _decode_float(total.registers), raw_data: { flow_registers: flow.registers, total_registers: total.registers, }, } finally: client.close()address 参数是从协议偏移量 0 开始的地址不是手册上写的 40001 这类 PLC 地址。很多流量计手册写的是保持寄存器 40001转换成 pymodbus 的 address 时要减去 40001也就是填 0。slave 参数对应设备从站地址RS-485 总线上每台设备地址必须唯一否则报文会被多台设备同时响应直接造成 CRC 错误。finally 里的 close 不是可选项串口资源是独占的不关闭会导致下次连接直接超时。3.3 寄存器里可能是浮点、BCD 码或整数流量计的累积量有时用 BCD 编码存储每个字节表示两位十进制数。_decode_float 只能应对 IEEE754 浮点格式遇到 BCD 编码时先按字节解包再组合出数值def _decode_bcd(raw_bytes): value 0 for byte in raw_bytes: value value * 100 (byte 4) * 10 (byte 0x0F) return value这个函数按字解码高四位和低四位最后得到的数值要乘以仪表手册里给出的倍率比如 0.001。不同仪表的数据格式差异可以用一张表固定下来方便排查现场数对不上数据编码寄存器个数解析方式典型用途IEEE754 float2 个寄存器struct 解包区分大小端瞬时流量、压力BCD 码4 个寄存器按字节高低四位拆累积流量、累计时间int161 个寄存器直接转有符号整数温度、百分比时间戳6 个寄存器年/月/日/时/分/秒仪表自记录日志项目说明或 README 里一般会给寄存器映射表但字节序不一定写清楚。调试时用flow.registers把原始寄存器值打印出来对照表头判断是不是大小端反了如果读出 1.8e-38 这种接近 0 的数大概率是两端反序。4. Celery 定时抄表周期任务、失败重试与漏采补数采集逻辑不能写在 Django 视图函数里这是远程抄表系统最容易绕路的地方。一次 HTTP 请求的生命周期只有几秒而读一台串口仪表可能要阻塞三秒要是现场有几十台表前端点一次刷新会连带把所有表都读一遍。更合理的做法是把采集做成独立定时任务由 Celery beat 按固定周期下发worker 进程去执行。4.1 为什么用 Celery 而不是 django-crontabdjango-crontab 也可以执行周期任务它的实现原理就是系统 crontab每次到点新起一个进程跑任务。设备少时看不出区别一旦采集任务里要记录日志、重试失败项、按设备分队列执行crontab 不好管理。Celery 把调度和执行拆成 beat 和 worker 两个进程任务结果和重试状态都集中管理后续要加实时 WebSocket 推送也有基础。4.2 Celery beat 最小配置与采集任务实现先安装依赖再在项目里创建 celery 应用和任务文件。有几个搜索“python django 搭建 web 项目”的人会忽略的细节settings 里配置 CELERY_BEAT_SCHEDULE 时schedule 的时区必须和业务时区一致否则定时任务会按 UTC 时间触发。# meter_project/celery.py import os from celery import Celery os.environ.setdefault(DJANGO_SETTINGS_MODULE, meter_project.settings) app Celery(meter_project) app.config_from_object(django.conf:settings, namespaceCELERY) app.autodiscover_tasks()# meter_project/settings.py 片段 CELERY_BROKER_URL redis://127.0.0.1:6379/1 CELERY_RESULT_BACKEND redis://127.0.0.1:6379/2 CELERY_TIMEZONE Asia/Shanghai CELERY_TASK_SERIALIZER json from celery.schedules import crontab CELERY_BEAT_SCHEDULE { collect_all_meters_once_a_minute: { task: meter.tasks.collect_all_meters, schedule: 60.0, options: {queue: meter_collect}, }, }任务文件里做两件事collect_all_meters 负责把设备 ID 列表遍历出来逐台派发独立任务collect_single_device 负责实际读取和落库。每台设备一个任务的好处是某一台仪表无响应时只影响它自己的重试不会拖累整批采集。# meter/tasks.py from celery import shared_task from .models import Device, MeterReading from .modbus_service import read_device shared_task(bindTrue, max_retries3, default_retry_delay30) def collect_single_device(self, device_id): device Device.objects.get(iddevice_id) try: data read_device(device) MeterReading.objects.create( devicedevice, instantaneous_flowdata[instantaneous_flow], total_flowdata[total_flow], raw_datadata.get(raw_data, {}), ) return {device_id: device_id, status: ok} except Exception as exc: raise self.retry(excexc) shared_task def collect_all_meters(): for device_id in Device.objects.values_list(id, flatTrue): collect_single_device.delay(device_id)bindTrue 让任务对象持有 self这样失败时能调用 self.retry。max_retries3 和 default_retry_delay30 表示失败后等 30 秒再重试最多三次。这里要注意retry 抛出的不是原始异常而是任务重试信号不要用 try 包住整个 retry。单台设备如果连续三次都失败会在 celery 日志里标记为 failed同时 MeterReading 里产生一个读取空档后续靠补采逻辑处理。启动命令分开跑 beat 和 worker便于单独重启celery -A meter_project worker -l info -Q meter_collect celery -A meter_project beat -l info-Q meter_collect指定队列确保周期采集任务不会和系统其他任务混抢 worker。如果部署环境内存紧张worker 和 beat 可以合并成一条命令生产环境还是分开稳妥。Celery 的核心参数和执行含义在改动前先确认参数常用值含义max_retries3某台设备连续失败的最大重试次数default_retry_delay30 秒首次重试的等待时间countdown30 秒调用 retry 时临时指定的延迟CELERY_BROKER_URLredis://127.0.0.1:6379/1存放任务消息的队列CELERY_RESULT_BACKENDredis://127.0.0.1:6379/2存放任务执行结果4.3 漏采数据补采用时间间隔扫描空洞任务重试解决的是瞬时故障仪表断电一天的场景需要另一种补采机制。常见的做法是扫历史数据找空洞取某台设备最近 N 条记录按 read_at 排序后两两比较间隔超过设定阈值就标记为缺失区间再由一个独立任务按区间重新读取仪表内部记录。流量计本身大多有存储功能能从仪表内部按时间查询数据这样补采才有实际意义。# meter/tasks.py 续 from celery import shared_task from .models import MeterReading GAP_THRESHOLD_SECONDS 90 shared_task def scan_reading_gaps(device_id, sample_count200): readings list( MeterReading.objects.filter(device_iddevice_id) .order_by(-read_at)[:sample_count] ) readings.reverse() gaps [] for prev, cur in zip(readings, readings[1:]): interval (cur.read_at - prev.read_at).total_seconds() if interval GAP_THRESHOLD_SECONDS: gaps.append({start: prev.read_at, end: cur.read_at}) return gaps这个任务只扫描不写库返回的 gaps 交给前端确认或另一个后台任务处理避免自动补采时写入大量重复数据。5. 宝塔部署 Django 抄表服务Nginx、uWSGI 与 Celery 进程协同源码包拿到手后本地跑通只是第一步放到服务器上才是真正考验配置能力的地方。很多部署问题不是 Django 代码的问题而是静态文件、socket 权限和进程守护这三件事没对齐。5.1 初始化项目与收集静态文件在宝塔面板的网站目录下执行虚拟环境创建和迁移。如果 requirements 里包含 pymodbus确认版本是否 3.xMySQL 场景下 mysqlclient 安装依赖系统 libmysqlclient-dev装不上时改用 PyMySQL 并在项目__init__.py里 pymysql.install_as_MySQLdb()。cd /www/wwwroot/meter_project python3 -m venv venv source venv/bin/activate pip install -r requirements.txt python manage.py migrate python manage.py collectstatic --noinputcollectstatic 这步别省。admin 和 DRF 的静态资源都要靠它集中到 STATIC_ROOTNginx 再直接指向这个目录。5.2 uWSGI 与 Nginx 的关键配置uWSGI 配置尽量保持最小用 socket 文件而不是端口和 Nginx 通讯[uwsgi] chdir /www/wwwroot/meter_project module meter_project.wsgi:application master true processes 2 threads 2 socket /tmp/meter.sock chmod-socket 664 vacuum true die-on-term trueNginx 站点配置里两个 location 必须正确/static/ 指向 collectstatic 输出目录/ 走 uwsgi_pass 转发。uwsgi_read_timeout 不要设太短等待结果超过 30 秒返回的查询会直接断开。server { listen 80; server_name meter.example.com; client_max_body_size 10m; location /static/ { alias /www/wwwroot/meter_project/static/; } location / { include uwsgi_params; uwsgi_pass unix:/tmp/meter.sock; uwsgi_read_timeout 30; } }5.3 部署后按这个顺序查问题管理员后台能打开但没样式先看 Nginx 的 /static/ 是不是指向了 collectstatic 的输出目录502 Bad Gateway 时检查 /tmp/meter.sock 是否存在、uWSGI 进程是否存活Celery beat 不下发任务时确认 redis-server 在跑、broker 地址没写错、worker 和 beat 使用的是同一套代码。串口设备采集超时先用lsof /dev/ttyUSB0查占用再检查波特率、数据位、停止位参数RS-485 接线极性反了会表现为连得上但读回来的数全是异常值。现象优先排查项处理手段管理后台无 CSS静态文件 alias 路径错误重新 collectstatic核对 Nginx alias502 Bad Gatewayuwsgi 未启动或 socket 无权限重启 uwsgi检查 chmod-socketbeat 不下发任务Redis 未启动或队列不一致单独启动 beatworker 加 -Q 指定队列串口读取超时串口被占用或 A/B 接反lsof 查占用核对 RS-485 极性读回数值异常小字节序解析错误打印原始寄存器按仪表手册调整大小端等这套排错流程跑过一遍系统的稳定性基本就由仪表本身决定而不是 Web 服务部分。本文还有配套的精品资源点击获取