Django+MySQL智慧交通公交系统:从ORM建模到实时位置模拟实战
简介这是一份基于PythonDjangoMySQL的智慧交通公交系统完整项目包适合Web开发学习者、毕业设计开发者以及对城市智能出行感兴趣的技术人员。项目围绕公交线路、车辆、站点与时刻表等核心数据完整实现用户管理、线路管理、车辆管理、时刻表、地图展示、数据分析与API接口等模块覆盖从数据库设计、后端业务逻辑到前端交互的整套流程可作为快速掌握DjangoMySQL联动开发的实战范例。资源共1510个文件其中HTML/CSS/JS构成前端页面Python/PYC实现后端逻辑SQL提供数据库初始化与表结构PNG/JPG等图片为界面素材另含文档、BPMN流程模型与概念数据模型等设计文件压缩包约114.15MB分层目录便于按模块查阅代码与设计文档配合可快速理解业务建模与二次开发。目前已有2105人学习适合需要完整项目参考和动手实践的技术人群。1. 智慧交通公交系统先搞清楚 zip 里装的到底是什么很多开发者的网盘里都躺着这样一个pythondjangomysql智慧交通公交系统.zip名字听着高大上但解压之后你会发现它没有多玄妙的算法本质就是一套公交车信息管理平台用户能查线路、查站牌、查某辆车现在跑到哪个路口管理员能维护线路、站点、班次和车辆。Django 负责业务逻辑和后台管理MySQL 负责把线路、站点、车辆位置这些结构化数据存好。它适合两类人一类是做课程设计或毕业设计需要一套能讲清楚前后端如何连接、数据库如何设计的完整项目另一类是想把某个城市的公交数据从 Excel 搬到线上、做一个能演示的调度大屏的开发者。这篇文章从数据模型一路写到避坑按我自己落地的方式把这个系统复现一遍。2. 把公交系统拆成数据模型Django ORM 与 MySQL 的表设计2.1 选型理由Django ORM 为什么适合公交业务快速迭代公交系统的核心实体是线路、站点、车辆、班次它们之间是典型的强关系模型一条线路经过多个站点一辆车在某个时间点归属一条线路一个站点被多条线路共享。这种场景用关系型数据库处理最自然而 Django 自带的 ORM 和迁移机制能把这个过程压缩到几行命令。常见做法是选 Django 加 MySQL而不是 Flask 加 MongoDB。原因不复杂Django 自带 Admin 后台班级或小团队做运营维护时不用额外写管理界面ORM 的迁移文件能记录每一次表结构变更出问题可以回滚MySQL 在 Linux 服务器上部署方便索引和事务机制成熟。如果你不是要处理 GPS 轨迹的千万级吞吐这套组合足够应付每小时几万次查询。我一般会建议新手把mysqlclient作为驱动而不是pymysql因为前者是 C 语言编译的原生驱动连接速度和稳定性更好。Windows 上装不了mysqlclient时再退回到pymysql但需要加上一行pymysql.install_as_MySQLdb()才能被 Django 识别。2.2 核心模型车站、线路、班次、实时位置的四张表怎么建模实际项目中不止四张表但核心定位是六张线路、站点、线路站点关系、班次、车辆、实时位置。线路和站点之间是多对多关系但因为需要保存“第几站”和“上行下行”这类属性所以不能直接用一个简单的ManyToManyField必须通过中间表来链接。下面是我会用的 Django 模型代码放在bus_app/models.py里from django.db import models class Station(models.Model): 公交站点存经纬度方便后续做地图展示 name models.CharField(max_length64, db_indexTrue, verbose_name站点名称) lng models.DecimalField(max_digits9, decimal_places6, verbose_name经度) lat models.DecimalField(max_digits9, decimal_places6, verbose_name纬度) def __str__(self): return self.name class Meta: db_table bus_station class Route(models.Model): 公交线路例如 一路、二路包含首末班时间 name models.CharField(max_length32, uniqueTrue, verbose_name线路名称) code models.CharField(max_length16, uniqueTrue, verbose_name线路编号) start_station models.ForeignKey(Station, on_deletemodels.PROTECT, related_nameroutes_as_start, verbose_name始发站) end_station models.ForeignKey(Station, on_deletemodels.PROTECT, related_nameroutes_as_end, verbose_name终点站) start_time models.TimeField(verbose_name首班时间) end_time models.TimeField(verbose_name末班时间) def __str__(self): return f{self.code} {self.name} class Meta: db_table bus_route class RouteStation(models.Model): 线路站点关系表用 order_no 控制站点顺序 route models.ForeignKey(Route, on_deletemodels.CASCADE, related_nameroute_stations, verbose_name线路) station models.ForeignKey(Station, on_deletemodels.CASCADE, related_nameroute_station_links, verbose_name站点) order_no models.PositiveIntegerField(default0, verbose_name站点顺序) updown models.SmallIntegerField(default0, choices[(0, 上行), (1, 下行)], verbose_name上下行) distance_to_next models.FloatField(default0, verbose_name到下一站距离(km)) class Meta: db_table bus_route_station ordering [route, updown, order_no] unique_together [(route, station, updown, order_no)]我在Route的外键上用了on_deletemodels.PROTECT这是故意为之。如果误删了一条线路系统会拒绝删除避免把站点和班次一起带没。对于RouteStation中间表则用CASCADE因为中间记录本身没有独立价值线路删了中间关系自然应该删掉。2.3 用 Django Migrations 在 MySQL 里建表两条必用的命令模型写完后需要先确认settings.py里的数据库连接。我的配置通常是DATABASES { default: { ENGINE: django.db.backends.mysql, NAME: smart_bus, USER: bus_admin, PASSWORD: your_password, HOST: 127.0.0.1, PORT: 3306, OPTIONS: { charset: utf8mb4, init_command: SET sql_modeSTRICT_TRANS_TABLES, }, } }然后执行生成迁移和同步数据库两条命令python manage.py makemigrations bus_app python manage.py migratemakemigrations只会生成迁移文件不会碰数据库migrate才是真正在 MySQL 里建表。执行完后可以用python manage.py shell进入 Python 解释器通过from bus_app.models import Route快速验证表能不能操作。需要留意的是MySQL 8.0 以上版本对字符集的要求是utf8mb4如果建库时用了utf8存中文没问题但存某些生僻字或 Emoji 会报错。3. 查询引擎与接口让用户查线路、查站牌、查实时车在哪3.1 核心查询逻辑线路查站点、站点查线路的 ORM 写法公交系统最常见的两个查询场景一个是给线路 ID按顺序返回所有站点一个是给站点 ID返回所有经过该站点的线路。用 ORM 写起来并不复杂但如果用错查询时机会掉进 N1 的坑。查线路站点时我通常会先取出线路再通过route_stations过滤并按order_no排序from bus_app.models import Route def get_route_stations(route_id, updown0): route Route.objects.prefetch_related( route_stations__station ).get(idroute_id) stations [ rs.station.name for rs in route.route_stations.all() if rs.updown updown ] return stations这里的关键是prefetch_related(route_stations__station)。如果只写route.route_stations.stationDjango 会在查询每个站点时额外发送一条 SQL30 个站点就是 30 条查询。加上prefetch_related后所有站点一次聚合成 IN 查询接口响应时间从几百毫秒降到几十毫秒。反查站点线路也是一样的思路用到Station的反向外键from django.db.models import Prefetch from bus_app.models import Station, RouteStation def get_routes_by_station(station_id): qs Route.objects.prefetch_related( Prefetch( route_stations, querysetRouteStation.objects.filter(station_idstation_id), to_attrhit_stations ) ).all() return [route.name for route in qs if route.hit_stations]Prefetch可以自定义子查询条件to_attr把过滤结果放进临时属性避免所有站点都被加载。3.2 用 Django REST Framework 暴露公交查询接口光有 ORM 不够前端要拿数据还得有 HTTP 接口。常规做法是引入djangorestframework把上面的查询逻辑放进 ViewSet 中。先定义一个简单的序列化器from rest_framework import serializers from bus_app.models import Station class StationSerializer(serializers.ModelSerializer): class Meta: model Station fields [id, name, lng, lat]再定义 ViewSetfrom rest_framework.decorators import action from rest_framework.response import Response from rest_framework.viewsets import ReadOnlyModelViewSet from bus_app.models import Route from bus_app.serializers import StationSerializer from bus_app.utils import get_route_stations, get_routes_by_station class RouteViewSet(ReadOnlyModelViewSet): queryset Route.objects.all() serializer_class StationSerializer # 仅示例 action(detailTrue, methods[get]) def stations(self, request, pkNone): GET /api/routes/{id}/stations/?updown0 updown request.query_params.get(updown, 0) data get_route_stations(pk, updownint(updown)) return Response({route_id: pk, stations: data})路由配置放在urls.pyfrom rest_framework.routers import DefaultRouter from bus_app.views import RouteViewSet router DefaultRouter() router.register(rroutes, RouteViewSet) urlpatterns router.urlsReadOnlyModelViewSet只给只读操作不需要写接口时尽量不要让系统暴露 POST、PUT 能力。action生成的是自定义子路径detailTrue表示这个动作是针对单个资源的。3.3 实时位置模拟与推送WebSocket 还是轮询实时位置是“智慧交通”最亮眼的点但也是最容易过度设计的点。小项目里前端用轮询就够了不用一上来就上 WebSocket。最简单可靠的轮询方案是额外提供一个接口比如GET /api/vehicles/current/返回所有正在运行车辆的位置集合。后端这个接口的典型实现from django.db.models import Prefetch from rest_framework.decorators import api_view from rest_framework.response import Response from bus_app.models import RealtimeLocation, Vehicle api_view([GET]) def current_vehicles(request): locations RealtimeLocation.objects.select_related(vehicle, schedule) \ .filter(captured_at__gte2024-01-01 00:00:00) # 实际应取最近时间 data [{ vehicle_id: loc.vehicle_id, plate_no: loc.vehicle.plate_no, lng: str(loc.lng), lat: str(loc.lat), speed: loc.speed, direction: loc.direction, captured_at: loc.captured_at.strftime(%H:%M:%S) } for loc in locations[:100]] return Response(data)实际项目中会把“最近时间”定义为datetime.now() - timedelta(minutes5)避免返回几百条过期位置。前端每隔 5 秒调一次这个接口地图上的车辆图标就能动起来。只有等用户量超过几百人时再考虑用 Django Channels 和 Redis 做 WebSocket 推送。基本规律是项目能演示、并发不高时轮询是最稳的还是后悔药最少的方案。4. 数据从哪来模拟实时公交位置与运营数据的三种落地方式4.1 用随机游走模型模拟车辆位置很多下载来的项目里实时位置是一堆死数据演示起来毫无生机。我一般会在开发环境里写一个模拟脚本让车辆沿着线路动起来。核心思路不是简单地给经纬度加随机数那会让车开到马路外去。正确做法是沿着线路站点之间的连线插值。假设一辆车正在从 A 站驶向 B 站已知两站的经纬度和车辆当前速度可以这样推进import math from random import uniform from bus_app.models import RealtimeLocation, Vehicle def move_along_route(vehicle_id, speed_kmh30): loc RealtimeLocation.objects.filter(vehicle_idvehicle_id).first() if not loc: return # 前端展示用高德/百度地图时经纬度需要是 GCJ-02 坐标 # 这里直接假设使用模拟坐标 now_lng float(loc.lng) now_lat float(loc.lat) target_lng float(loc.next_station_lng) target_lat float(loc.next_station_lat) dist_left math.sqrt((target_lng - now_lng) ** 2 (target_lat - now_lat) ** 2) step speed_kmh * 5 / 3600 # 5秒移动的距离单位km近似用十进制度替换 if dist_left step: # 到达下一站切换到下一段 loc.next_station_order (loc.next_station_order 1) % loc.route_station_count else: ratio step / dist_left now_lng (target_lng - now_lng) * ratio now_lat (target_lat - now_lat) * ratio loc.lng now_lng loc.lat now_lat loc.speed uniform(15, 45) loc.captured_at datetime.now() loc.save(update_fields[lng, lat, speed, captured_at])update_fields是 Django 里一个容易被忽略的好习惯它只更新指定列不会让整行数据因为一次位置移动而重新写一遍可以减少锁竞争和无效写日志。4.2 定时任务把位置写入 MySQLCelery 还是 manage.py 命令模拟数据写进数据库不能靠人手动点击需要有定时器持续调用。对于中小项目我倾向不引入 Celery而是写一个 Django management command再用系统 cron 调度。命令文件放在bus_app/management/commands/run_simulation.pyfrom django.core.management.base import BaseCommand from django.utils import timezone from bus_app.services import simulate_all_vehicles class Command(BaseCommand): help 模拟所有在线车辆的位置更新 def add_arguments(self, parser): parser.add_argument(--interval, typeint, default5, help每次更新的间隔秒数) def handle(self, *args, **options): interval options[interval] while True: simulate_all_vehicles() time.sleep(interval)需要说明的是这个命令不建议用while True常驻在普通的管理命令里它更适合在开发环境调试。生产环境建议把这个循环抽成独立进程或者用celery beat配置定时任务。如果项目想控制服务生命周期可以改写成--loop参数实现单次运行由外部 cron 每 5 秒调用一次。4.3 让系统能导入真实数据CSV 导入脚本模拟数据只能用来演示真实业务需要批量导入。通常公交公司能导出的数据是 Excel转成 CSV 后按列映射到我们的模型。我会在management/commands/import_stations.py里写一个可复用的脚本import csv from django.core.management.base import BaseCommand from bus_app.models import Station class Command(BaseCommand): help 从 CSV 导入站点数据表头: name,lng,lat def add_arguments(self, parser): parser.add_argument(csv_file, typestr, helpCSV 文件路径) def handle(self, *args, **options): with open(options[csv_file], r, encodingutf-8-sig) as f: reader csv.DictReader(f) station_objs [ Station(namerow[name], lngrow[lng], latrow[lat]) for row in reader ] Station.objects.bulk_create(station_objs, ignore_conflictsTrue) self.stdout.write(self.style.SUCCESS(f成功导入 {len(station_objs)} 条站点))utf-8-sig编码是为了兼容从 Windows 导出的带 BOM 头的 CSV否则第一列字段会被\ufeff污染。ignore_conflictsTrue也一样重要它保证重复导入时不会因为唯一约束中断最多跳过重复行。5. 避坑与常见问题DjangoMySQL 公交系统最容易翻车的 5 个地方5.1 现象启动项目一直报 Cant connect to MySQL server刚下载的项目在自己的电脑上跑不起来最常见原因是 MySQL 服务没启动或者项目里的数据库账号密码与本地环境不一致。有的项目为了让压缩包开箱即用会在配置里写死一个root/123456但你的 MySQL 可能是root加随机密码。另外HOST 填localhost和127.0.0.1在某些系统上行为不同localhost可能走 Unix socket127.0.0.1走 TCP如果 MySQL 没开 TCP 或者防火墙拦截前者通后者不通。解决步骤是先确认服务在运行再用命令行客户端测试同样的连接串。如果确认服务没问题就在settings.py里把 HOST 改成127.0.0.1并检查是否给当前用户做了远程授权。注意grant all privileges on smart_bus.* to bus_admin127.0.0.1这类授权命令必须带上 IP不要只授权localhost。5.2 现象写入中文时提示 Incorrect string value这个坑在 MySQL 5.7 时代尤其常见因为默认字符集是latin1。如果你建库时没有显式指定CREATE DATABASE smart_bus DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;那么即使 Django 的OPTIONS里写了charset: utf8mb4表结构和连接的字符集不一致写入中文仍然报错。解决方式是先建库再建表。如果已经建错了用ALTER DATABASE smart_bus CHARACTER SET utf8mb4;改库再ALTER TABLE CONVERT TO CHARACTER SET utf8mb4;改表。更关键的是检查settings.py中OPTIONS是否真的写上了charsetDjango 对 MySQL 的默认连接字符集是utf8它和utf8mb4并不完全等价。5.3 现象实时位置比本地时间晚了 8 个小时位置时间在界面上显示成 UTC 时间每到中午看过去还是凌晨。原因是settings.py里保留了默认的TIME_ZONE UTC而USE_TZ True。Django 默认把所有DateTimeField按 UTC 存入 MySQL展示时才转成本地时间。如果项目做的是国内公交系统我建议直接TIME_ZONE Asia/Shanghai USE_TZ TrueUSE_TZ True时Django ORM 写入前会把本地时间转成 UTC存进去的其实是 UTC读取时会转回 Asia/Shanghai。如果不想碰时区转换这团乱麻可以设置USE_TZ False让系统按 MySQL 所在时区纯本地时间处理。两种方案都能用但务必统一不能一半靠自动转换一半手动datetime.now()。5.4 现象查询接口越调越慢最后直接超时公交系统的接口通常只查几十条数据不至于超时除非踩了 N1 查询。比如前面get_route_stations如果不用prefetch_related每次返回 30 个站点就会产生 31 条 SQL 语句。调试方法是在 Django shell 里打开查询日志from django.db import connection connection.queries观察queries数量如果一次列表接口产生几十条 SQL就要检查 ORM 查询链中每一层外键。还有一个容易忽略的点是给常用字段加数据库索引比如RouteStation的(route, updown, order_no)组合索引Django 的unique_together会自动创建不用额外加。但如果经常按站点反查线路RouteStation.station字段也需要db_indexTrue否则查询全表扫描。5.5 现象多车并发更新位置时出现 deadlock模拟脚本开启多个进程后偶尔会看到Deadlock found when trying to get lock。原因是多个事务同时更新同一张表的不同行或者同时执行update_or_create时相互等待。我的解决方式是收紧事务范围让每个位置更新尽量只碰自己的唯一记录。try: with transaction.atomic(): RealtimeLocation.objects.update_or_create( vehicle_idvehicle_id, defaults{ lng: lng, lat: lat, speed: speed, captured_at: now, } ) except OperationalError: # 重试一次 pass这里我故意把vehicle_id作为唯一查询条件并且不用“查询再保存”两步而是直接update_or_create事务持有锁的时间尽量短。如果还频繁死锁需要确认数据库隔离级别是否为REPEATABLE READ可以临时改成READ COMMITTED但这一步通常没必要先检查代码里有没有多余的查询写在事务里面。6. 进阶从课程设计到可用的公交信息平台还有三步要走项目跑通之后你可能会发现它还是本地演示版。我的建议是先做三件事缓存高频查询、接入地图渲染、把 Admin 变成运营后台。第一缓存。线路和站点数据基本不变完全不值得每次请求都查 MySQL。用 Django 的缓存框架配合 Redis把get_route_stations的结果缓存 60 秒最简单做法是from django.core.cache import cache def get_route_stations_cached(route_id, updown0): key froute_stations_{route_id}_{updown} data cache.get(key) if data is None: data get_route_stations(route_id, updown) cache.set(key, data, 60) return data实时位置不能缓存太久否则车辆看起来不动但可以把计算结果缓 3 秒足够支撑大屏演示。第二地图渲染。前端用 Leaflet 加 OpenStreetMap 或国内厂商的地图把captured_at最新的车辆经纬度画成点每隔 5 秒重新请求一次就是一个能看的“智慧交通大屏”。第三Admin 后台。Django Admin 默认只管理你注册的模型但公交运营需要按线路查看站点顺序而不是一张平坦的列表。你可以重写admin.py中的TabularInline把RouteStation内联到RouteAdmin下面运营人员一眼看出某条线路的站序。我最后想提醒的是这类 zip 项目最大的问题不是代码而是没有初始数据。解压后跑通只是第一步真正让系统有价值的是把一条真实线路、十辆车、几十个站点填进去让接口返回的数据可信。我以前也以为下载来的项目比自己做省事后来才发现改别人的表结构比重新写还难受。所以如果你时间允许还是建议把模型按自己业务重写一遍。整个流程走下来你对 Django ORM、MySQL 设计和接口分层都会上一个台阶。希望帮到你。本文还有配套的精品资源点击获取