# coding:utf-8 # !/usr/bin/env python3 __author__ = "Wenjian Long" __version__ = "0.0.1" __email__ = "Wenjian.Long@lnxall.com" import threading import paho.mqtt.client as mqtt #from queue import Queue #import serial import time import json import base64 import struct import types import syslog import os import sys from time import sleep #from onvif import ONVIFCamera #from onvif.exceptions import ONVIFError MQTT_BROKER = "localhost" MQTT_PORT = 1883 MQTT_USER = None MQTT_PASSWD = None MQTT_QoS = 2 MQTT_CONN_TIMEOUT = 10 MQTT_PUB_TIMEOUT = 6000 GW_SN = "218806000001" IPC_CONFIG_NAME = "/app/config/ip_camera_cfg.json" NODE_CONFIG_NAME = "/app/node/nodes_cfg.json" FAC_INI_FILE = "/app/config/fac.ini" ipc_objects = [] ipc_config = [] node_config = [] mqtt_client = None data_list = [] mqtt_msg_list = [] pgrep_ffmpeg_fl = "pgrep -fl ffmpeg" pgrep_ffmpeg_afl = "pgrep -afl ffmpeg" class IPCamera: def __init__(self, host, port, user, password, nvrAddr=None, rtsp_url=None, wsdl_dir=None): self.host = host self.port = port self.user = user self.password = password self.rtsp_url = rtsp_url self.wsdl_dir = '/usr/wsdl' self.cam = None self.media = None self.media_profile = None self.ptz = None self.invalid = 0 self.onvif_error_info = 0 self.streamCnt = 1 self.ipc_InitEnv(host, port, user, password) if self.invalid == 0 and nvrAddr != None: self.event_init() def event_listen(self, req): syslog.syslog("===============================event_listen===================================") def ipc_InitEnv(self, host, port, user, password): syslog.syslog("==============(%s %d %s %s)" %(host, port, user, password)) self.streamCnt = 1 self.ptz_error_info = 0 self.ptz = 0 self.invalid = 0 syslog.syslog("python ipcamera ONVIFCamera init ok...") def event_init(self): syslog.syslog("------event_init------") def ipc_GetStreamUri(self): syslog.syslog("ipc_GetStreamUri...") url = self.rtsp_url syslog.syslog("url:%s" %url) obj_result = {} obj_result['rtspUrl'] = url obj_result['streamCnt'] = self.streamCnt obj_result['result'] = 0 data_json = json.dumps(obj_result) return data_json def perform_move(self, request, timeout=1): # Start continuous move syslog.syslog("perform_move...") def move_ContinuousMove(self, x, y, z=0, timeout=1): syslog.syslog("move_ContinuousMove...") def perform_RelativeMove(self, request, timeout): # Start continuous move syslog.syslog("perform_RelativeMove...") def move_RelativeMove(self, x, y, z=0, timeout=1): syslog.syslog('move_RelativeMove...') def ptz_GetPresets(self): syslog.syslog('ptz_GetPresets...') def ptz_SetPreset(self, name): syslog.syslog('ptz_SetPreset...') def ptz_GotoPreset(self, token): syslog.syslog('ptz_GotoPreset...') def ptz_RemovePreset(self, token): syslog.syslog('ptz_RemovePreset...') def ptz_GotoHomePosition(self): syslog.syslog('ptz_GotoHomePosition...') def ptz_SetHomePosition(self): syslog.syslog('ptz_SetHomePosition...') def perform_AbsoluteMove(self, request, timeout): # Start continuous move syslog.syslog('perform_AbsoluteMove...') def move_AbsoluteMove(self, x, y, z=0, timeout=1): syslog.syslog("move_AbsoluteMove...") def ipc_CheckPtz(self): if self.ptz_error_info != 0 or self.ptz == 0: return 1 return 0 def protocol_encode(self, json_str): err_code=0 #parse obj=json.loads(json_str) obj_result={} #encode identifier = obj['identifier'] syslog.syslog("python ipcamera protocol_encode in...") if identifier == 'GetStreamUri': json_str = self.ipc_GetStreamUri() syslog.syslog("json_str:%s" %json_str) return json_str elif identifier == 'GetPresets': err_code = self.ipc_CheckPtz() if err_code != 0: obj_result['error_info'] = self.ptz_error_info obj_result['result'] = err_code json_str=json.dumps(obj_result) else: json_str = self.ptz_GetPresets() syslog.syslog("json_str:%s" %json_str) return json_str elif identifier == 'SetPreset': err_code = self.ipc_CheckPtz() if err_code != 0: obj_result['error_info'] = self.ptz_error_info else: preset_token = self.ptz_SetPreset(obj['preset_name']) if preset_token == None: err_code = 1 else: obj_result['preset_token'] = preset_token obj_result['result'] = err_code json_str=json.dumps(obj_result) syslog.syslog("json_str:%s" %json_str) return json_str elif identifier == 'GotoPreset': err_code = self.ipc_CheckPtz() if err_code != 0: obj_result['error_info'] = self.ptz_error_info else: preset_token = self.ptz_GotoPreset(obj['preset_token']) obj_result['result'] = err_code json_str=json.dumps(obj_result) syslog.syslog("json_str:%s" %json_str) return json_str elif identifier == 'RemovePreset': err_code = self.ipc_CheckPtz() if err_code != 0: obj_result['error_info'] = self.ptz_error_info else: preset_token = self.ptz_RemovePreset(obj['preset_token']) obj_result['result'] = err_code json_str=json.dumps(obj_result) syslog.syslog("json_str:%s" %json_str) return json_str elif identifier == 'GotoHomePosition': err_code = self.ipc_CheckPtz() if err_code != 0: obj_result['error_info'] = self.ptz_error_info else: preset_token = self.ptz_GotoHomePosition() obj_result['result'] = err_code json_str=json.dumps(obj_result) syslog.syslog("json_str:%s" %json_str) return json_str elif identifier == 'SetHomePosition': err_code = self.ipc_CheckPtz() if err_code != 0: obj_result['error_info'] = self.ptz_error_info else: preset_token = self.ptz_SetHomePosition() obj_result['result'] = err_code json_str=json.dumps(obj_result) syslog.syslog("json_str:%s" %json_str) return json_str elif identifier == 'ContinuousMove': err_code = self.ipc_CheckPtz() if err_code != 0: obj_result['error_info'] = self.ptz_error_info else: self.move_ContinuousMove(obj['x'], obj['y']) obj_result['result'] = err_code json_str=json.dumps(obj_result) syslog.syslog("json_str:%s" %json_str) return json_str else: syslog.syslog("ERROR can't find cmd") syslog.syslog("python ipcamera protocol_encode out...") obj_result['result'] = 1 json_str=json.dumps(obj_result) return json_str class MyEncoder(json.JSONEncoder): def default(self, obj): """ 只要检查到了是bytes类型的数据就把它转为str类型 :param obj: :return: """ if isinstance(obj, bytes): return str(obj, encoding='utf-8') return json.JSONEncoder.default(self, obj) class IPCConfig: def __init__(self, file_name, list_name, mode): self.file_name = file_name self.list_name = list_name self.node_cnt = 0 self.node_json = {} self.file_object = open(file_name, mode) self.file_data = self.file_object.read() syslog.syslog(self.file_data) if self.file_data is not None and len(self.file_data) > 0: self.node_json = json.loads(self.file_data) self.node_cnt = len(self.node_json[self.list_name]) syslog.syslog("self.node_cnt:%d" % self.node_cnt) def get_node_by_sn(self, sn): for node in self.node_json[self.list_name]: if node["sn"] == sn: return node def get_node_by_index(self, index): syslog.syslog("--sn %s" % self.node_json[self.list_name][index]["sn"]) return self.node_json[self.list_name][index] def delete_node_by_index(self, index): del self.node_json[self.list_name][index] self.node_cnt = len(self.node_json[self.list_name]) self.file_data = json.dumps(self.node_json) tmp_object = open(self.file_name, "w") #self.file_object.seek(0, 0) tmp_object.write(self.file_data) tmp_object.close() def add_update_node(self, node): syslog.syslog("sn:%s" %(node["sn"])) find = 0 if len(self.node_json) > 0: for index in range(len(self.node_json[self.list_name])): syslog.syslog("index:%d" %(index)) syslog.syslog("sn:%s %s" %(node["sn"],self.node_json[self.list_name][index]["sn"])) if self.node_json[self.list_name][index]["sn"] == node["sn"]: find = 1 syslog.syslog("self.file_data:%s" %self.file_data) del self.node_json[self.list_name][index] self.node_json[self.list_name].append(node) self.file_data = json.dumps(self.node_json) syslog.syslog("self.file_data:%s" %self.file_data) tmp_object = open(self.file_name, "w") #self.file_object.seek(0, 0) tmp_object.write(self.file_data) tmp_object.close() syslog.syslog("add ok...") if find == 0: if len(self.node_json) < 1: self.node_json[self.list_name] = [] self.node_json[self.list_name].append(node) self.node_cnt = len(self.node_json[self.list_name]) self.file_data = json.dumps(self.node_json) tmp_object = open(self.file_name, "w") #self.file_object.seek(0, 0) tmp_object.write(self.file_data) tmp_object.close() def update_node_config(self, node): syslog.syslog("sn:%s" %(node["sn"])) for index in range(len(self.node_json[self.list_name])): syslog.syslog("index:%d" %(index)) syslog.syslog("sn:%s %s" %(node["sn"],self.node_json[self.list_name][index]["sn"])) if self.node_json[self.list_name][index]["sn"] == node["sn"]: syslog.syslog("self.file_data:%s" %self.file_data) del self.node_json[self.list_name][index] self.node_json[self.list_name].append(node) self.file_data = json.dumps(self.node_json) syslog.syslog("self.file_data:%s" %self.file_data) tmp_object = open(self.file_name, "w") #self.file_object.seek(0, 0) tmp_object.write(self.file_data) tmp_object.close() class MQTTClient: def __init__(self, host, port, qos, timeout=60, user=None, passwd=None): self.host = host self.port = port self.qos = qos self.timeout = timeout #self.ser = ser self.mqtt_client = mqtt.Client() if user is not None and passwd is not None: self.mqtt_client.username_pw_set(user, passwd) self.mqtt_client.on_message = self.on_message self.mqtt_client.on_disconnect = self.on_disconnect self.mqtt_client.on_connect = self.on_connect self.mqtt_client.connect(self.host, self.port, self.timeout) def on_message(self, client, userdata, msg): payload = msg.payload.decode().strip().replace('\\n', '').replace('\\t', '') mqtt_recv_msg(msg.topic.decode(), payload) def on_disconnect(self, client, userdata, rc): syslog.syslog("disconnect with result code: " + str(rc)) self.mqtt_client.reconnect() def on_connect(self, client, userdata, flags, rc): syslog.syslog("connect with result code: " + str(rc)) def subscribe(self, topic): self.mqtt_client.subscribe(topic, self.qos) syslog.syslog("subscribe %s" % topic) def unsubscribe(self, topic): self.mqtt_client.unsubscribe(topic) syslog.syslog('unsubscribe %s' % topic) def publish(self, topic, blob): self.mqtt_client.publish(topic, blob) def loop_start(self): if self.mqtt_client is not None: self.mqtt_client.loop_forever() def init_ipcamera(ipc_object, ipc_json): #初始化IPC实例 if ipc_object != None: ipc_object["ipc_client"].ipc_InitEnv(host=ipc_json["ip_addr"], port=ipc_json["port"], user=ipc_json["user"], password=ipc_json["password"]) ipc_object["user"] = ipc_json["user"] ipc_object["password"] = ipc_json["password"] ipc_object["status_time"] = 0 if ipc_object["ipc_client"].invalid == 0: raw_rtspUrl = ipc_object["ipc_client"].protocol_encode("{\"identifier\":\"GetStreamUri\"}") ipc_object["rtspUrl"],ipc_object["wan_rtspUrl"] = generate_rtsp_url(raw_rtspUrl, ipc_object["ip_addr"], ipc_object["user"], ipc_object["password"]) ipc_object["online"] = True syslog.syslog("IPCamera ReInit rtsp %s Connect OK..." %ipc_object["rtspUrl"]) else: syslog.syslog("IPCamera Connect invalid:%d..." %ipc_object["ipc_client"].invalid) return ipc["ipc_client"].invalid else: ipc = {} if "nvrAddr" in ipc_json: ipc["ipc_client"] = IPCamera(host=ipc_json["ip_addr"], port=ipc_json["port"], user=ipc_json["user"], password=ipc_json["password"], nvrAddr=ipc_json["nvrAddr"]) elif "rtsp_url" in ipc_json: ipc["ipc_client"] = IPCamera(host=ipc_json["ip_addr"], port=ipc_json["port"], user=ipc_json["user"], password=ipc_json["password"], nvrAddr=None, rtsp_url=ipc_json["rtsp_url"]) else: ipc["ipc_client"] = IPCamera(host=ipc_json["ip_addr"], port=ipc_json["port"], user=ipc_json["user"], password=ipc_json["password"]) syslog.syslog("invalid:%d" %(ipc["ipc_client"].invalid)) if ipc["ipc_client"] != None: ipc["sn"] = ipc_json["sn"] ipc["ip_addr"] = ipc_json["ip_addr"] ipc["port"] = ipc_json["port"] ipc["user"] = ipc_json["user"] ipc["password"] = ipc_json["password"] ipc["online"] = False ipc["status_time"] = 0 if "nvrUser" in ipc_json: ipc["nvrUser"] = ipc_json["nvrUser"] ipc["nvrPassword"] = ipc_json["nvrPassword"] ipc["nvrAddr"] = ipc_json["nvrAddr"] ipc["nvrChannel"] = ipc_json["nvrChannel"] if ipc["ipc_client"].invalid == 0: raw_rtspUrl = ipc["ipc_client"].protocol_encode("{\"identifier\":\"GetStreamUri\"}") ipc["rtspUrl"],ipc["wan_rtspUrl"] = generate_rtsp_url(raw_rtspUrl, ipc["ip_addr"], ipc["user"], ipc["password"]) ipc["online"] = True syslog.syslog("IPCamera rtsp %s Connect OK..." %ipc["rtspUrl"]) else: syslog.syslog("IPCamera Connect invalid:%d..." %ipc["ipc_client"].invalid) ipc_objects.append(ipc) return ipc["ipc_client"].invalid else: return 1 #数据上报 def data_report(data, type, result_data): send_data = {} send_data["sn"] = data["sn"] send_data["identifier"] = data["identifier"] send_data["time"] = int(time.time()) send_data["mi"] = data["mi"] send_data["port"] = "TCP_CLIENT" send_data["tag_node"] = result_data send_data["data_type"] = type send_json = json.dumps(send_data, cls=MyEncoder, indent=4, ensure_ascii=False) if type == 0: publish_topic = "ipc/" + GW_SN + "/TCP_CLIENT/device/" + data["sn"] + "/data/property/" + data["identifier"] elif type == 1: publish_topic = "ipc/" + GW_SN + "/TCP_CLIENT/device/" + data["sn"] + "/data/event/" + data["identifier"] elif type == 2: publish_topic = "ipc/" + GW_SN + "/TCP_CLIENT/device/" + data["sn"] + "/data/service/" + data["identifier"] topic_payload = {} topic_payload["topic"] = publish_topic topic_payload["payload"] = send_json data_list.append(topic_payload) #数据上报 def status_report(ipc_obj, now_time): ipc_status = {} ipc_status["status"] = [] status = {} status["sn"] = ipc_obj["sn"] status["online"] = ipc_obj["online"] status["last_rcv"] = now_time status["login_time"] = 0 ipc_status["status"].append(status) send_json = json.dumps(ipc_status, cls=MyEncoder, indent=4, ensure_ascii=False) publish_topic = "ipc/" + GW_SN + "/TCP_CLIENT/Evt_NodesStatus" topic_payload = {} topic_payload["topic"] = publish_topic topic_payload["payload"] = send_json data_list.append(topic_payload) #检查IPC是否在线 def ipcamera_online_check(ip_addr): ping_str = "ping {0} -c 3".format(ip_addr) find_str = "bytes from {0}".format(ip_addr) f = os.popen(ping_str, "r") ping_status = f.read() if ping_status.find(find_str) > 0: syslog.syslog("ping %s result 1" %(ip_addr)) return 1 else: syslog.syslog("ping %s result 0" %(ip_addr)) return 0 def MQTT_listen(mqtt_client): mqtt_client.loop_start() #IPC事件处理消息处理 def event_msg_ctrl(ip_addr, event_info): data = {} syslog.syslog("==ip_addr %s event_info:%s" %(ip_addr,event_info)) for index in range(len(ipc_objects)): if ip_addr == ipc_objects[index]["ip_addr"]: data["mi"] = 0 data["sn"] = ipc_objects[index]["sn"] data["identifier"] = "IpcAlarm" result_data = json.dumps(event_info) data_report(data, 1, result_data) #mqtt消息处理 def ipc_msg_ctrl(mqtt_client): status_time_5m = 0 while True: find = 0 now_time = int(time.time()) #syslog.syslog("==node_cnt %d ipc_cnt:%d" %(node_config.node_cnt,len(ipc_objects))) for msg_index in range(len(mqtt_msg_list)): topic = mqtt_msg_list[msg_index]["topic"] payload = mqtt_msg_list[msg_index]["payload"] syslog.syslog("==len(mqtt_msg_list):%d" %(len(mqtt_msg_list))) syslog.syslog("ipc_msg_ctrl topic:%s" %topic) syslog.syslog("ipc_msg_ctrl payload:%s" %payload) try: tmp_data = json.loads(payload) data = json.loads(tmp_data["data_b64"]) result = 0 except Exception,err: syslog.syslog("error data_b64 json.loads:%s" %tmp_data["data_b64"]) mqtt_msg_list.pop(msg_index) break; for index in range(len(ipc_objects)): if data["sn"] == ipc_objects[index]["sn"]: find = 1 syslog.syslog("find connected sn:%s" %(ipc_objects[index]["sn"])) #IPC初始化 if data["identifier"] == "InitIPCamera": result_data = {} #更新用户名密码 if ipc_objects[index]["ipc_client"].invalid == 1 or ipc_objects[index]["online"] == False: #初始化 if ipc_objects[index]["ipc_client"].onvif_error_info.find("Host is unreachable") > 0 and ipcamera_online_check(ipc_objects[index]["ip_addr"]) == 1: result = init_ipcamera(ipc_objects[index], ipc_objects[index]) else: result = 1 if result == 1: result_data['error_info'] = ipc_objects[index]["ipc_client"].onvif_error_info else: #初始化成功，直接返回成功 result = 0 result_data["result"] = result result_data = json.dumps(result_data) elif ipc_objects[index]["ipc_client"].invalid == 1: result_data = {} result_data["result"] = 1 result_data['error_info'] = ipc_objects[index]["ipc_client"].onvif_error_info result_data = json.dumps(result_data) #开始推流 elif data["identifier"] == "PushVideoStreaming": #检查是否正在推流 push = get_push_video_status(ipc_objects[index]["ip_addr"]) #没有推流，再开户推流 if push == 0: push_video_streaming(ipc_objects[index]["rtspUrl"], data["serverUrl"]) sleep(1) push = get_push_video_status(ipc_objects[index]["ip_addr"]) if push == 0: result = 1 else: result = 0 ipc_objects[index]["push_time"] = now_time syslog.syslog("push_time startTime %d..." %ipc_objects[index]["push_time"]) #组回应包 result_data = {} result_data["result"] = result result_data = json.dumps(result_data) #停止推流 elif data["identifier"] == "StopVideoStreaming": #检查是否正在推流 ffmpeg_pid = get_push_video_status(ipc_objects[index]["ip_addr"]) ipc_objects[index]["push_time"] = 0 #停止推流 if ffmpeg_pid != 0: stop_video_streaming(ffmpeg_pid) #组回应包 result_data = {} result_data["result"] = 0 result_data = json.dumps(result_data) #获取rtspUrl elif data["identifier"] == "GetStreamUri": result_data = {} result_data["rtspUrl"] = ipc_objects[index]["wan_rtspUrl"] result_data["result"] = 0 result_data = json.dumps(result_data) #开始推NVR回放流 elif data["identifier"] == "PushNvrReplayStreaming": #生成端口映射 if get_port_mapping(554, ipc_objects[index]["nvrAddr"]) == 0: generate_port_mapping(555, ipc_objects[index]["nvrAddr"], 554) #组NVR rtspUrl #rtsp://admin:a12345678@10.3.1.46:554/Streaming/tracks/101?starttime=20200721t103000z&endtime=20200721t104000z #检查是否正在推流 push = get_push_video_status(ipc_objects[index]["nvrAddr"]) nvrRtspUrl = "rtsp://{0}:{1}@{2}:554/Streaming/tracks/10{3}?starttime={4}z&endtime={5}z".format(ipc_objects[index]["nvrUser"],ipc_objects[index]["nvrPassword"],ipc_objects[index]["nvrAddr"],ipc_objects[index]["nvrChannel"],data["startTime"].replace(" ", "t"),data["endTime"].replace(" ", "t")) #没有推流，再开户推流 if push == 0: push_video_streaming(nvrRtspUrl, data["serverUrl"]) #组回应包 result_data = {} result_data["result"] = 0 result_data = json.dumps(result_data) #停止推NVR回放流 elif data["identifier"] == "StopNvrReplayStreaming": #检查是否正在推流 ffmpeg_pid = get_push_video_status(ipc_objects[index]["nvrAddr"]) #停止推流 if ffmpeg_pid != 0: stop_video_streaming(ffmpeg_pid) #组回应包 result_data = {} result_data["result"] = 0 result_data = json.dumps(result_data) #执行相应摄像机控制操作 else: result_data = ipc_objects[index]["ipc_client"].protocol_encode(tmp_data["data_b64"]) #上位机消息应答 data_report(data, 0, result_data) data_report(data, 2, result_data) break mqtt_msg_list.pop(msg_index) break; if find == 0: sleep(1) for index in range(len(ipc_objects)): if "push_time" in ipc_objects[index] and ipc_objects[index]["push_time"] > 0 and now_time - ipc_objects[index]["push_time"] >= 300: #检查是否正在推流 ffmpeg_pid = get_push_video_status(ipc_objects[index]["ip_addr"]) #停止推流 if ffmpeg_pid != 0: syslog.syslog("StopVideoStreaming Auto startTime %d..." %ipc_objects[index]["push_time"]) stop_video_streaming(ffmpeg_pid) ipc_objects[index]["push_time"] = 0 #5分钟检测一下设备状态 if now_time - status_time_5m > 300 and ipc_objects[index]["online"] == True: status_time_5m = now_time if ipcamera_online_check(ipc_objects[index]["ip_addr"]) == 0: ipc_objects[index]["online"] = False status_report(ipc_objects[index], now_time) #1小时更新一下设备状态 if now_time - ipc_objects[index]["status_time"] > 3600 or ipc_objects[index]["status_time"] == 0: ipc_objects[index]["status_time"] = now_time status_report(ipc_objects[index], now_time) #接收MQTT消息，加入到mqtt消息队列 def mqtt_recv_msg(topic, payload): #syslog.syslog("topic:%s payload:%s" %(topic, payload)) mqtt_msg = {} mqtt_msg["topic"] = topic mqtt_msg["payload"] = payload mqtt_msg_list.append(mqtt_msg) return #生成端口映射 def generate_port_mapping(out_port, ip_addr, port): mapping = "uci add firewall redirect;uci set firewall.@redirect[-1].src=wan;uci set firewall.@redirect[-1].src_dport={0};uci set firewall.@redirect[-1].dest_ip={1};uci set firewall.@redirect[-1].dest=lan;uci set firewall.@redirect[-1].dest_port={2};uci set firewall.@redirect[-1].proto=tcp;uci commit firewall;/etc/init.d/firewall restart;".format(out_port, ip_addr, port) f = os.popen(mapping, "r") d = f.read() f.close() if d == None: syslog.syslog("generate_port_mapping 0") return 0 else: syslog.syslog("generate_port_mapping 1") return 1 #获取当前端口是否已映射 def get_port_mapping(port, ip_addr): mapping = "grep -A 5 -B 5 \"src_dport '{0}'\" /etc/config/firewall | grep \"dest_ip '{1}'\"".format(port, ip_addr) f = os.popen(mapping, "r") d = f.read() f.close() if d == None: syslog.syslog("get_port_mapping 0") return 0 else: syslog.syslog("get_port_mapping 1") return 1 #生成rtsp url def generate_rtsp_url(raw_rtspUrl, ip_addr, user, password): #raw_rtspUrl = "{\"streamCnt\": 3, \"rtspUrl\": \"rtsp://172.16.254.102:554/cam/realmonitor?channel=1&subtype=0&unicast=true&proto=Onvif\", \"result\": 0}" data = json.loads(raw_rtspUrl) if data["rtspUrl"] == None: return None,None #获取wan ip address f = os.popen(r"ubus call network.interface.wan status | grep \"address\" | grep -oE '[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}';", "r") wan_ip = f.read() f.close() syslog.syslog("raw_rtspUrl:%s wan_ip:%s" %(data["rtspUrl"],wan_ip)) #rtsp = user + ":" + password + "@" + wan_ip[0:-1] rtsp = user + ":" + password + "@" + ip_addr #rtspUrl = data["rtspUrl"].replace(ip_addr,rtsp) subtype_tmp = "subtype={0}".format(data["streamCnt"]-1) Hik_subtype = "/10{0}?".format(data["streamCnt"]) #rtspUrl = rtspUrl.replace("subtype=0", subtype_tmp) #rtspUrl = rtspUrl.replace("/101?", Hik_subtype) rtspUrl = data["rtspUrl"] #生成端口映射 if get_port_mapping(554, ip_addr) == 0: generate_port_mapping(554, ip_addr, 554) syslog.syslog("rtspUrl:%s" %rtspUrl) return rtspUrl,rtspUrl.replace(ip_addr, wan_ip[0:-1]) #换取当前推流状态,返回pid def get_push_video_pid(ip_addr, cmd): ffmpeg_status = cmd f = os.popen(ffmpeg_status, "r") ffmpeg_start = f.read() #try: sh_index = ffmpeg_start.find(cmd) if sh_index >= 0 and sh_index < 30: syslog.syslog("strip cmd:%s" %(cmd)) ffmpeg_start = ffmpeg_start[sh_index+len(cmd)+1:] ffmpeg_start.strip() ip_index = ffmpeg_start.find(ip_addr) syslog.syslog("cmd:%s ip_index:%d %s" %(cmd, ip_index, ffmpeg_start)) pid_index = -1 if ip_index > 100: tmp_string = ffmpeg_start[ip_index-100:] tmp_index = tmp_string.find("logger -t ffmpeg") + len("logger -t ffmpeg") ffmpeg_start = tmp_string[tmp_index+1:] syslog.syslog("ffmpeg_start:%s" %(ffmpeg_start)) pid_index = ffmpeg_start.find("ffmpeg -rtsp_transport") if ffmpeg_start.find(ip_addr) < 0 or pid_index < 0: syslog.syslog("get_push_video_status 0") return 0 else: syslog.syslog("pid_index:%d" %pid_index) ffmpeg_pid = int(ffmpeg_start[0:(pid_index-1)]) syslog.syslog("get_push_video_status 1 ffmpeg_pid:%d" %ffmpeg_pid) return ffmpeg_pid #except TypeError,err: #syslog.syslog("err %s " %err) #换取当前推流状态,返回pid def get_push_video_status(ip_addr): ffmpeg_pid = get_push_video_pid(ip_addr, pgrep_ffmpeg_afl) if ffmpeg_pid == 0: ffmpeg_pid = get_push_video_pid(ip_addr, pgrep_ffmpeg_fl) return ffmpeg_pid #停止推流 def stop_video_streaming(ffmpeg_pid): stop_video = "kill -9 {0}".format(ffmpeg_pid) syslog.syslog("push:%s" %stop_video) os.system(stop_video) return 1 #推流 def push_video_streaming(rtsp_url, server_url): push_video = "ffmpeg -rtsp_transport tcp -nostdin -i \"{0}\" -vcodec copy -acodec copy -f flv \"{1}\" 2>&1 | logger -t 'ffmpeg' & ".format(rtsp_url, server_url) syslog.syslog("push_video_streaming:%s" %push_video) os.system(push_video) return 0 def get_sn(): global GW_SN file_object = open(FAC_INI_FILE, "r+") file_data = file_object.read() sn_index = file_data.find("sn") if sn_index > 0: sn_tmp = file_data[sn_index:] sn_list = sn_tmp.split('=') GW_SN = sn_list[-1].strip() syslog.syslog("GW_SN:%s" %GW_SN) def main_loop(): syslog.syslog("*******************start ipcamera python*******************") global ipc_config,node_config #加载IPC配置文件 get_sn() #加载节点配置文件 node_config = IPCConfig(NODE_CONFIG_NAME, "nodes_cfg", "r") #ipc消息处理线程 ipc_msg_thread = threading.Thread(target=ipc_msg_ctrl, args=(None,)) ipc_msg_thread.setDaemon(True) ipc_msg_thread.start() for node_index in range(node_config.node_cnt): node_json = node_config.get_node_by_index(node_index) if node_json["connect_port"] == "TCP_CLIENT" and "ext_data" in node_json: ext_data=json.loads(node_json["ext_data"]) if "user" in ext_data and "password" in ext_data: ipc_json = {} ipc_json["sn"] = node_json["sn"] ipc_json["ip_addr"] = node_json["tcp_ip_addr"] ipc_json["port"] = int(node_json["tcp_port"]) ipc_json["user"] = ext_data["user"] ipc_json["password"] = ext_data["password"] if "rtsp_url" in ext_data: ipc_json["rtsp_url"] = ext_data["rtsp_url"] if "nvrUser" in ext_data: ipc_json["nvrUser"] = ext_data["nvrUser"] ipc_json["nvrPassword"] = ext_data["nvrPassword"] ipc_json["nvrAddr"] = ext_data["nvrAddr"] ipc_json["nvrChannel"] = ext_data["nvrChannel"] ipc_config.append(ipc_json) #初始化IPC实例 init_ipcamera(None, ipc_json) #建立MQTT连接 mqtt_client = MQTTClient(MQTT_BROKER, MQTT_PORT, MQTT_QoS, timeout=MQTT_CONN_TIMEOUT, user=MQTT_USER, passwd=MQTT_PASSWD) topic = "ipc/+/TCP_CLIENT/device/+/data/Set_Rglt_Raw" #topic = "G/" + GW_SN + "/Rsp_ServiceData" #订阅 mqtt_client.subscribe(topic) #MQTT线程 mqtt_thread = threading.Thread(target=MQTT_listen, args=(mqtt_client,)) mqtt_thread.setDaemon(True) mqtt_thread.start() #发布MQTT消息 while True: for index in range(len(data_list)): syslog.syslog("publish topic:%s payload:%s" %(data_list[index]["topic"],data_list[index]["payload"])) mqtt_client.publish(data_list[index]["topic"], data_list[index]["payload"]) data_list.pop(index) break; sleep(1) if __name__ == '__main__': main_loop() syslog.syslog("finished")