系列第8篇 | 从单设备监控到多宠物家庭的完整管理平台建设前言当一个家庭养了3只猫、2只狗还有一只仓鼠时单摄像头方案显然不够用了。你需要的是一个多设备、多宠物、多用户的统一管理平台。本文将从云端架构设计、设备管理、数据聚合、权限控制四个维度手把手教你搭建一个可扩展的多宠物管理平台。一、整体架构设计1.1 系统架构图┌─────────────────────────────────────────────────────┐ │ 客户端层 │ │ ┌──────────┐ ┌──────────┐ ┌──────────────────┐ │ │ │ 移动App │ │ Web管理台 │ │ 智能音箱/手表 │ │ │ └────┬─────┘ └────┬─────┘ └────────┬─────────┘ │ │ └──────────────┼─────────────────┘ │ ├──────────────────────┼───────────────────────────────┤ │ API Gateway (Kong/Nginx) │ ├──────────────────────┼───────────────────────────────┤ │ 服务层 │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌─────────┐ │ │ │设备管理 │ │宠物管理 │ │告警服务 │ │用户服务 │ │ │ │Service │ │Service │ │Service │ │Service │ │ │ └────┬─────┘ └────┬─────┘ └────┬─────┘ └────┬────┘ │ ├───────┼────────────┼────────────┼─────────────┼──────┤ │ 数据层 │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌─────────┐ │ │ │PostgreSQL│ │Redis │ │InfluxDB │ │MinIO/S3 │ │ │ │业务数据 │ │缓存/实时 │ │时序指标 │ │视频/图片│ │ │ └──────────┘ └──────────┘ └──────────┘ └─────────┘ │ ├──────────────────────────────────────────────────────┤ │ 设备层 │ │ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐ │ │ │摄像头1 │ │摄像头2 │ │喂食器 │ │智能猫砂 │ │ │ └────────┘ └────────┘ └────────┘ └────────┘ │ └──────────────────────────────────────────────────────┘1.2 技术选型组件技术选型选型理由API服务Go Gin高并发、低内存适合IoT场景消息队列NATS轻量级支持MQTT桥接延迟低时序数据库InfluxDB 2.x专为IoT指标设计写入性能优异关系数据库PostgreSQL支持JSONB适合灵活的设备元数据对象存储MinIO / 阿里云OSS视频片段和图片存储实时通信WebSocket WebRTC视频流和实时通知二、设备管理服务2.1 设备注册与发现每个设备通过唯一DeviceID注册支持自动发现局域网设备# device_registry.pyimportuuidimportjsonfromdatetimeimportdatetimefromdataclassesimportdataclass,field,asdictfromenumimportEnumfromtypingimportOptionalclassDeviceType(Enum):CAMERAcameraFEEDERfeederLITTER_BOXlitter_boxWATER_FOUNTAINwater_fountainDOOR_FLAPdoor_flapWEARABLEwearableclassDeviceStatus(Enum):ONLINEonlineOFFLINEofflineUPDATINGupdatingERRORerrordataclassclassDeviceInfo:device_id:strdevice_type:DeviceType name:strlocation:str# 如 客厅, 卧室firmware_version:str1.0.0ip_address:Optional[str]Nonemac_address:Optional[str]Nonestatus:DeviceStatusDeviceStatus.OFFLINE capabilities:listfield(default_factorylist)bound_pets:listfield(default_factorylist)last_heartbeat:Optional[datetime]Nonemetadata:dictfield(default_factorydict)classDeviceRegistry:设备注册中心 - 管理所有宠物监控设备def__init__(self,redis_client):self.redisredis_client self.prefixdevice:defregister(self,device:DeviceInfo)-bool:注册新设备device_keyf{self.prefix}{device.device_id}dataasdict(device)# Enum序列化data[device_type]device.device_type.value data[status]device.status.value data[registered_at]datetime.utcnow().isoformat()# 使用Hash存储设备信息self.redis.hset(device_key,mapping{k:json.dumps(v)ifisinstance(v,(dict,list))elsestr(v)fork,vindata.items()})# 设置设备在线状态TTL 60秒心跳超时self.redis.setex(fdevice:online:{device.device_id},60,1)# 添加到设备索引self.redis.sadd(devices:all,device.device_id)self.redis.sadd(fdevices:type:{device.device_type.value},device.device_id)returnTruedefheartbeat(self,device_id:str)-bool:设备心跳上报self.redis.setex(fdevice:online:{device_id},60,1)self.redis.hset(f{self.prefix}{device_id},last_heartbeat,datetime.utcnow().isoformat())returnTruedefget_online_devices(self)-list:获取所有在线设备all_devicesself.redis.smembers(devices:all)online[]fordidinall_devices:ifself.redis.exists(fdevice:online:{did}):infoself.redis.hgetall(f{self.prefix}{did})online.append(info)returnonlinedefbind_pet(self,device_id:str,pet_id:str):将宠物绑定到设备self.redis.sadd(fdevice:pets:{device_id},pet_id)self.redis.sadd(fpet:devices:{pet_id},device_id)defget_device_pets(self,device_id:str)-list:获取设备关联的宠物列表returnlist(self.redis.smembers(fdevice:pets:{device_id}))2.2 心跳检测与离线告警# heartbeat_monitor.pyimportasyncioimportloggingfromdatetimeimportdatetime,timedelta loggerlogging.getLogger(__name__)classHeartbeatMonitor:设备心跳监控 - 检测设备离线并触发告警def__init__(self,redis_client,notification_service):self.redisredis_client self.notifiernotification_service self.check_interval30# 每30秒检查一次self.offline_threshold120# 120秒无心跳视为离线asyncdefstart(self):启动心跳监控循环logger.info(心跳监控服务启动)whileTrue:awaitself._check_all_devices()awaitasyncio.sleep(self.check_interval)asyncdef_check_all_devices(self):devicesself.redis.smembers(devices:all)fordevice_idindevices:device_iddevice_id.decode()ifisinstance(device_id,bytes)elsedevice_id is_onlineself.redis.exists(fdevice:online:{device_id})ifnotis_online:# 检查是否已经标记离线避免重复告警already_offlineself.redis.get(fdevice:offline_flag:{device_id})ifnotalready_offline:awaitself._handle_device_offline(device_id)asyncdef_handle_device_offline(self,device_id:str):处理设备离线事件logger.warning(f设备{device_id}离线)# 标记离线避免重复告警self.redis.setex(fdevice:offline_flag:{device_id},3600,1)# 获取设备信息和关联宠物device_infoself.redis.hgetall(fdevice:{device_id})bound_petsself.redis.smembers(fdevice:pets:{device_id})# 构建告警消息pet_names[]forpet_idinbound_pets:pet_nameself.redis.hget(fpet:{pet_id},name)ifpet_name:pet_names.append(pet_name.decode())device_namedevice_info.get(bname,device_id)message(f⚠️ 设备离线告警\nf设备:{device_name}\nf关联宠物:{, .join(pet_names)or无}\nf时间:{datetime.now().strftime(%Y-%m-%d %H:%M:%S)}\nf请检查设备电源和网络连接)# 获取设备所属用户并发送通知user_iddevice_info.get(buser_id,b).decode()ifuser_id:awaitself.notifier.send(user_id,message,priorityhigh)三、多宠物数据聚合3.1 宠物行为数据模型# pet_data_aggregator.pyfromdataclassesimportdataclassfromdatetimeimportdatetime,date,timedeltafromtypingimportOptionalfromenumimportEnumclassActivityType(Enum):SLEEPINGsleepingEATINGeatingPLAYINGplayingWALKINGwalkingRESTINGrestingGROOMINGgroomingLITTER_USElitter_useDRINKINGdrinkingdataclassclassPetActivity:pet_id:stractivity_type:ActivityType start_time:datetime end_time:Optional[datetime]Noneduration_seconds:int0source_device:strconfidence:float0.0metadata:dictNoneclassPetDataAggregator:多宠物数据聚合服务 - 汇总各设备数据生成宠物健康报告def__init__(self,influx_client,pg_client):self.influxinflux_client self.pgpg_clientdefrecord_activity(self,activity:PetActivity):记录宠物行为到时序数据库point{measurement:pet_activity,tags:{pet_id:activity.pet_id,activity:activity.activity_type.value,device:activity.source_device,},fields:{duration:activity.duration_seconds,confidence:activity.confidence,},time:activity.start_time,}self.influx.write_api.write(bucketpet_monitor,recordpoint)defget_daily_summary(self,pet_id:str,target_date:dateNone)-dict:获取单只宠物每日行为摘要iftarget_dateisNone:target_datedate.today()startdatetime.combine(target_date,datetime.min.time())endstarttimedelta(days1)queryf from(bucket: pet_monitor) | range(start:{start.isoformat()}Z, stop:{end.isoformat()}Z) | filter(fn: (r) r[pet_id] {pet_id}) | filter(fn: (r) r[_field] duration) | group(columns: [activity]) | sum() resultsself.influx.query_api().query(query)summary{}fortableinresults:forrecordintable.records:activityrecord.values.get(activity,unknown)summary[activity]record.get_value()return{pet_id:pet_id,date:target_date.isoformat(),activities:summary,total_active_minutes:sum(summary.values())/60,}defget_multi_pet_comparison(self,household_id:str,days:int7)-dict:多宠物行为对比分析petsself._get_household_pets(household_id)comparison{}forpetinpets:daily_summarie