import json
|
import configparser
|
import os
|
import subprocess
|
import logging
|
from kafka import KafkaConsumer
|
|
# 设置日志记录
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
logger = logging.getLogger(__name__)
|
|
def read_config(config_path):
|
"""读取配置文件"""
|
config = configparser.ConfigParser()
|
config.read(config_path)
|
return config
|
|
def subscribe_kafka(config):
|
"""订阅Kafka topic"""
|
kafka_server = config['KAFKA']['server']
|
topic = config['TRANSFER']['topic_1']
|
|
consumer = KafkaConsumer(
|
topic,
|
bootstrap_servers=kafka_server,
|
value_deserializer=lambda m: json.loads(m.decode('utf-8'))
|
#auto_offset_reset='earliest'
|
)
|
logger.info(f"成功订阅 Kafka 主题: {topic}")
|
return consumer
|
|
def process_kafka_message(message, config, file_index):
|
"""解析Kafka消息并按channel_size拆分写入JSON"""
|
output_dir = os.path.dirname(config['TRANSFER']['outfile'])
|
os.makedirs(output_dir, exist_ok=True)
|
channel_size = int(config['TRANSFER']['channel_size'])
|
|
# 读取允许的 actions
|
allowed_actions = set(config['TRANSFER']['actions'].split(','))
|
|
# 解析通道信息
|
channels = message.get("channels", [])
|
logger.info(f"收到 Kafka 消息，包含 {len(channels)} 个通道")
|
|
# 读取配置模板
|
config_template = {
|
"version": "1.0",
|
"show_screen": False,
|
"show_board": False,
|
"kafka_server": "172.18.0.247:9092",
|
"kafka_topic": "vp_ba_result",
|
"max_input_per_detection": channel_size,
|
"ba_jam_config": message.get("ba_jam_config", {}),
|
"ba_stop_config": message.get("ba_stop_config", {}),
|
"ba_person_gathering_config": message.get("ba_person_gathering_config", {}),
|
"ba_wrong_direction_config": message.get("ba_wrong_direction_config", {}),
|
"channels": []
|
}
|
|
# 拆分通道，每channel_size个写入一个新文件
|
for i in range(0, len(channels), channel_size):
|
config_template["channels"] = []
|
for channel in channels[i:i + channel_size]:
|
processed_channel = {
|
"id": str(channel["id"]),
|
"url": channel["url"],
|
"source_type": "ffmpeg_rtsp",
|
"resize_ratio": 0.5,
|
"pix_width": channel.get("pix_width", 1280),
|
"pix_height": channel.get("pix_height", 720),
|
"actions": []
|
}
|
|
# 处理 actions，仅保留 STOP 和 JAM
|
for act in channel.get("actions", []):
|
action_name = act.get("action", "").upper() # 统一转换为大写
|
if action_name not in allowed_actions:
|
continue # 过滤掉不在配置文件中的 action
|
|
regions = act.get("regions", [])
|
formatted_regions = []
|
|
if len(regions) == 0:
|
processed_channel["actions"].append({
|
"action": action_name
|
})
|
else:
|
# 确保 regions 是一个数组，并且其中的元素符合格式
|
action_region = []
|
for point in regions:
|
if isinstance(point, list) and len(point) > 0:
|
action_region.append([int(point[0]),int(point[1])])
|
formatted_regions.append(action_region)
|
processed_channel["actions"].append({
|
"action": action_name,
|
"regions": formatted_regions
|
})
|
|
# 仅当 actions 不为空时才加入 channels
|
if processed_channel["actions"]:
|
config_template["channels"].append(processed_channel)
|
|
# 写入 JSON 文件
|
file_path = os.path.join(output_dir, f"config_{file_index}.json")
|
try:
|
with open(file_path, 'w') as f:
|
json.dump(config_template, f, indent=4)
|
logger.info(f"已生成配置文件: {file_path}")
|
except IOError as e:
|
logger.error(f"写入 {file_path} 失败: {e}")
|
|
file_index += 1
|
|
return file_index
|
|
def get_process_pids(process_name):
|
"""获取进程名称匹配的所有PID"""
|
try:
|
output = subprocess.check_output(["pgrep", "-f", process_name], text=True)
|
return [int(pid) for pid in output.strip().split("\n") if pid]
|
except subprocess.CalledProcessError:
|
return []
|
|
def manage_processes(config):
|
"""管理进程：停止旧进程，启动新进程"""
|
process_file = config['PROCESSES']['process_file']
|
try:
|
with open(process_file, 'r') as f:
|
process_data = json.load(f)
|
except Exception as e:
|
logger.error(f"读取进程文件失败: {e}")
|
return
|
|
# 停止进程
|
process_names = process_data.get("process_names", [])
|
for process in process_names:
|
pids = get_process_pids(process)
|
for pid in pids:
|
try:
|
os.kill(pid, 9)
|
logger.info(f"成功停止进程 {process} (PID: {pid})")
|
except ProcessLookupError:
|
logger.warning(f"进程 {process} (PID: {pid}) 不存在")
|
except Exception as e:
|
logger.error(f"停止进程 {process} 失败: {e}")
|
|
# 启动进程
|
for process in process_data.get("process_list", []):
|
start_process(process)
|
|
def start_process(proc):
|
"""启动单个进程"""
|
program_path = proc.get('program_path')
|
command_ = proc.get('commands')
|
|
try:
|
if program_path == "gnome-terminal":
|
subprocess.Popen([program_path, '--', 'bash', '-c', command_])
|
elif program_path == "xterm":
|
subprocess.Popen([program_path, '-e', 'bash -c "' + command_ + '"'])
|
elif program_path == "konsole":
|
subprocess.Popen([program_path, '--hold', '-e', 'bash -c "' + command_ + '"'])
|
else:
|
subprocess.Popen([program_path, '-c', command_])
|
|
logger.info(f"启动进程: {command_}")
|
except FileNotFoundError as e:
|
logger.error(f"终端程序未找到: {e}")
|
except Exception as e:
|
logger.error(f"启动进程失败: {e}")
|
|
def main():
|
"""主函数"""
|
config_path = 'msg_config.ini'
|
config = read_config(config_path)
|
|
consumer = subscribe_kafka(config)
|
if not consumer:
|
return
|
|
file_index = 0
|
for message in consumer:
|
try:
|
file_index = process_kafka_message(message.value, config, file_index)
|
manage_processes(config)
|
except Exception as e:
|
logger.error(f"处理 Kafka 消息时发生错误: {e}")
|
|
if __name__ == "__main__":
|
main()
|
|
|
|
|
'''
|
现有msg_config.ini文件,编写一份python程序,实现以下功能：
|
1. 读取msg_config.ini文件,msg_config.ini文件内容如下:
|
[KAFKA]
|
server = 192.168.0.79:9092
|
topics = vp_ba_jam, vp_ba_stop, vp_ba_crossline, vp_tracking
|
|
[OUTPUT]
|
file = kafka_messages.txt
|
|
[TRANSFER]
|
topic_1 = cameralist
|
channel_size = 12
|
outfile = textjson/config.json
|
|
[PROCESSES]
|
process_file = textjson/process_list.json
|
|
[SEND_MSG]
|
run_send = true
|
send_topics = cameralist
|
send_interval = 300
|
json_file = textjson/cameralist.json
|
|
2. 订阅kafka并接受topic为cameralist消息,并解析该消息，消息格式示例如：
|
{"version":"1.0","channels":[{"id":9,"url":"rtsp://admin:Shmixcqwe%21%40%23@172.18.0.241:554/Streaming/Channels/101","actions":[{"action":"STOP","regions":[["560","120"],["706","116"],["1188","698"],["114","700"]],"frame_interval":10,"save_image":true,"save_video":true,"max_distance":3,"notify_interval":30},{"action":"JAM","regions":[["520","92"],["712","80"],["1264","488"],["338","700"]],"frame_interval":10,"save_image":true,"save_video":true,"min_stops":2,"max_distance":3,"notify_interval":30},{"action":"SMOCKING","frame_interval":10,"save_image":true,"save_video":true},{"action":"FALL","frame_interval":10,"save_image":true,"save_video":true,"notify_interval":30},{"action":"JUNK_STACK","frame_interval":10,"save_image":true,"save_video":true},{"action":"CROWD_GATHERING","frame_interval":10,"save_image":true,"save_video":true},{"action":"FIRE","frame_interval":10,"save_image":true,"save_video":true}],"pix_width":1280,"pix_height":720},{"id":10,"url":"rtsp://admin:Shmixcqwe%21%40%23@172.18.0.241:554/Streaming/Channels/202","actions":[{"action":"STOP","regions":[["538","80"],["692","68"],["1268","476"],["364","700"]],"frame_interval":10,"save_image":true,"save_video":true,"max_distance":3,"notify_interval":30},{"action":"JAM","regions":[["560","100"],["692","108"],["880","698"],["40","676"]],"frame_interval":10,"save_image":true,"save_video":true,"min_stops":2,"max_distance":3,"notify_interval":30},{"action":"SMOCKING","frame_interval":10,"save_image":true,"save_video":true},{"action":"FALL","frame_interval":10,"save_image":true,"save_video":true,"notify_interval":30},{"action":"JUNK_STACK","frame_interval":10,"save_image":true,"save_video":true},{"action":"CROWD_GATHERING","frame_interval":10,"save_image":true,"save_video":true},{"action":"FIRE","frame_interval":10,"save_image":true,"save_video":true}],"pix_width":1280,"pix_height":720}]，
|
"ba_jam_config": {
|
"check_interval_frames": 20,
|
"check_max_distance": 8,
|
"check_min_hit_frames": 50,
|
"check_min_stops": 8,
|
"check_notify_interval": 10,
|
"need_record_image": true,
|
"need_record_video": true
|
},
|
"ba_stop_config": {
|
"check_interval_frames": 20,
|
"check_max_distance": 5,
|
"check_min_hit_frames": 50,
|
"need_record_image": true,
|
"need_record_video": true
|
},
|
"ba_person_gathering_config": {
|
"check_interval_frames": 1,
|
"check_min_hit_frames": 1,
|
"check_notify_interval": 10,
|
"min_gathering_person": 3,
|
"need_record_image": true,
|
"need_record_video": true,
|
"person_gathering_distance_factor": 4.0
|
},
|
"ba_wrong_direction_config": {
|
"check_interval_frames": 5,
|
"check_min_hit_frames": 25,
|
"check_notify_interval": 10,
|
"half_width": 320,
|
"need_record_image": true,
|
"need_record_video": true
|
}
|
}
|
|
3. 解析上述消息内容,生成以下json格式的内容并覆盖写入outfile对应值的文件路径,每channel_size个channel生成一份新文件,命名为:textjson/config_{n}.json:
|
{
|
"version": "1.0",
|
"show_screen": false,
|
"show_board": false,
|
"kafka_server": "172.18.0.247:9092",
|
"kafka_topic": "vp_ba_result",
|
"max_input_per_detection": 4,
|
"ba_jam_config": {
|
"check_interval_frames": 20,
|
"check_max_distance": 8,
|
"check_min_hit_frames": 50,
|
"check_min_stops": 8,
|
"check_notify_interval": 10,
|
"need_record_image": true,
|
"need_record_video": true
|
},
|
"ba_stop_config": {
|
"check_interval_frames": 20,
|
"check_max_distance": 5,
|
"check_min_hit_frames": 50,
|
"need_record_image": true,
|
"need_record_video": true
|
},
|
"ba_person_gathering_config": {
|
"check_interval_frames": 1,
|
"check_min_hit_frames": 1,
|
"check_notify_interval": 10,
|
"min_gathering_person": 3,
|
"need_record_image": true,
|
"need_record_video": true,
|
"person_gathering_distance_factor": 4.0
|
},
|
"ba_wrong_direction_config": {
|
"check_interval_frames": 5,
|
"check_min_hit_frames": 25,
|
"check_notify_interval": 10,
|
"half_width": 320,
|
"need_record_image": true,
|
"need_record_video": true
|
},
|
"channels": [
|
{
|
"id": "9",
|
"url": "rtsp://admin:Shmixcqwe%21%40%23@172.18.0.241:554/Streaming/Channels/101",
|
"source_type": "ffmpeg_rtsp",
|
"resize_ratio": 0.5,
|
"pix_width": 1280,
|
"pix_height": 720,
|
"actions": [
|
{
|
"action": "STOP",
|
"regions": [
|
[[560,120],[706,116],1188,698],[114,700]]
|
]
|
},
|
{
|
"action": "JAM",
|
"regions": [
|
[[520,92],[712, 80,[1264,488],[338,700]]
|
]
|
}
|
]
|
},
|
{
|
"id": "10",
|
"url": "rtsp://admin:Shmixcqwe%21%40%23@172.18.0.241:554/Streaming/Channels/201",
|
"source_type": "ffmpeg_rtsp",
|
"resize_ratio": 0.5,
|
"pix_width": 1280,
|
"pix_height": 720,
|
"actions": [
|
{
|
"action": "STOP",
|
"regions": [
|
[[560,120],[706,116],1188,698],[114,700]]
|
]
|
},
|
{
|
"action": "JAM",
|
"regions": [
|
[[520,92],[712, 80,[1264,488],[338,700]]
|
]
|
}
|
]
|
}
|
]
|
}
|
其中,max_input_per_detection的值为channel_size,actions下的元素仅保留action和regions,将regions写在一行之内,如没有regions,则将regions设置为默认值[]
|
|
4. 文件写入完毕之后,读取process_file指定的textjson/process_list.json进程清单文件,文件格式如:
|
{
|
"process_names": ["top", "mixc2"],
|
"process_list": [
|
{"program_path": "gnome-terminal", "commands": "top -d 1"},
|
{"program_path": "gnome-terminal", "commands": "top -p 1234"},
|
{"program_path": "gnome-terminal", "commands": "top -u wr"},
|
{"program_path": "gnome-terminal", "commands": "top -H"}
|
]
|
}
|
5. 读取process_names,停止进程名开头为"top"、“mixc2”的所有进程;
|
|
6. 停止上述进程之后,读取process_list进程清单,并调起命令行终端启动process_list中的所有程序,直到所有程序启动完毕;
|
|
7. 确保该程序在linux下能执行。
|
'''
|
