import json
|
import configparser
|
import os
|
import sys
|
import subprocess
|
import logging
|
import threading
|
import argparse
|
import time
|
import datetime
|
import signal
|
import functools
|
from kafka import KafkaConsumer
|
from datetime import datetime, timedelta
|
import schedule
|
|
# 创建日志记录器
|
logger = logging.getLogger(__name__)
|
|
# 设置日志级别
|
logger.setLevel(logging.INFO)
|
|
# 创建文件处理器，日志写入到文件
|
file_handler = logging.FileHandler('./parkwatch.log')
|
file_handler.setLevel(logging.INFO) # 文件输出的日志级别
|
|
# 创建控制台处理器，日志输出到控制台
|
console_handler = logging.StreamHandler()
|
console_handler.setLevel(logging.INFO) # 控制台输出的日志级别
|
|
# 定义日志格式
|
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
|
file_handler.setFormatter(formatter)
|
console_handler.setFormatter(formatter)
|
|
# 将处理器添加到日志记录器
|
logger.addHandler(file_handler)
|
logger.addHandler(console_handler)
|
|
# 记录日志
|
logger.info("parkwatch Program started")
|
|
process_data = {}
|
|
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']['access_topic']
|
|
consumer = KafkaConsumer(
|
topic,
|
bootstrap_servers=kafka_server,
|
value_deserializer=lambda m: json.loads(m.decode('utf-8'))
|
#auto_offset_reset='earliest'
|
)
|
logger.info(f"subscribe Kafka topic : {topic} success!")
|
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)
|
file_name, file_extension = os.path.splitext(os.path.basename(config['TRANSFER']['outfile']))
|
channel_size = int(config['TRANSFER']['channel_size'])
|
default_skip_interval = int(config['TRANSFER']['default_skip_interval'])
|
default_gate_interval = int(config['TRANSFER']['default_gate_interval'])
|
source_type = config['TRANSFER']['source_type']
|
video_resize_ratio = round(float(config['TRANSFER']['video_resize_ratio']), 1)
|
kfk_server = config['KAFKA']['server']
|
public_topic = config['KAFKA']['public_topic']
|
need_independent = config.getboolean('TRANSFER', 'need_independent')
|
independent_action = [item.strip() for item in config['TRANSFER']['independent_action'].split(',')]
|
independent_channel_size = int(config['TRANSFER']['independent_channel_size'])
|
independent_channels = [] # 用于存储需要单独处理的通道
|
# 读取允许的 actions
|
allowed_actions = set(config['TRANSFER']['actions'].split(','))
|
|
# 解析通道信息
|
channels = message.get("channels", [])
|
logger.info(f"get Kafka message, contains {len(channels)} channels")
|
|
# 读取配置模板
|
config_template = {
|
"version": "1.0",
|
"show_screen": False,
|
"show_board": False,
|
"kafka_server": kfk_server,
|
"kafka_topic": public_topic,
|
"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_person_fall_down_config": message.get("ba_person_fall_down_config", {}),
|
"ba_wrong_direction_config": message.get("ba_wrong_direction_config", {}),
|
"ba_barrier_gate_config": message.get("ba_barrier_gate_config", {}),
|
"ba_fire_smoke_config": message.get("ba_fire_smoke_config", {}),
|
"ba_junk_stack_config": message.get("ba_junk_stack_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": channel.get("source_type", source_type),
|
"skip_interval": channel.get("skip_interval", default_skip_interval),
|
"resize_ratio": channel.get("resize_ratio", video_resize_ratio),
|
"pix_width": channel.get("pix_width", 1280),
|
"pix_height": channel.get("pix_height", 720),
|
"actions": []
|
}
|
|
# 处理 actions，仅保留 STOP 和 JAM
|
for act in channel.get("actions", []):
|
act["action"] = act.get("action", "").upper()
|
if act["action"] not in allowed_actions:
|
continue # 过滤掉不在配置文件中的 action
|
if act["action"] == independent_action[0] and need_independent:
|
processed_channel["skip_interval"] = default_gate_interval
|
# 将该通道添加到 independent_channels 中，而不是 config_template["channels"]
|
independent_channels.append(processed_channel)
|
processed_channel["actions"].append(act)
|
# 不将该通道加入到 config_template["channels"]
|
continue
|
processed_channel["actions"].append(act)
|
|
# 仅当 actions 不为空时才加入 channels
|
if processed_channel["actions"]:
|
config_template["channels"].append(processed_channel)
|
|
# 写入 JSON 文件
|
file_path = os.path.join(output_dir, f"{file_name}_{file_index}{file_extension}")
|
write_config_file(config_template, file_path)
|
file_index += 1
|
|
# 如果 independent_channels 不为空，将其写入一个新的文件
|
if independent_channels:
|
# 过滤掉 independent_channels 中的 actions
|
for channel in independent_channels:
|
channel["actions"] = [act for act in channel["actions"] if act["action"] in independent_action]
|
|
# 更新 config_template，设置最大通道数为 independent_channel_size
|
config_template["channels"] = independent_channels
|
config_template["max_input_per_detection"] = independent_channel_size
|
|
# 写入文件
|
file_path = os.path.join(output_dir, f"{file_name}_gate{file_extension}")
|
write_config_file(config_template, file_path)
|
|
return file_index
|
|
def write_config_file(config_template, file_path):
|
try:
|
with open(file_path, 'w') as f:
|
json.dump(config_template, f, indent=4)
|
logger.info(f"created file: {file_path}")
|
except IOError as e:
|
logger.error(f"write file: {file_path} faild: {e}")
|
|
|
def get_process_pids(process_name):
|
"""获取进程名称匹配的所有PID"""
|
try:
|
output = subprocess.check_output(["pgrep", "-x", process_name], text=True)
|
return [int(pid) for pid in output.strip().split("\n") if pid]
|
except subprocess.CalledProcessError:
|
return []
|
|
def get_process_info(config):
|
"""
|
获取进程信息，从配置文件中的process_file加载数据。
|
:param config: 配置字典，包含PROCESS配置
|
:return: 进程数据字典
|
"""
|
process_data = {}
|
# 从配置中获取process_file路径
|
process_file = config['PROCESSES']['process_file']
|
if not process_file:
|
logger.error("Configuration missing 'process_file' in PROCESSES section.")
|
return process_data
|
try:
|
# 打开文件并加载JSON数据
|
with open(process_file, 'r') as file:
|
process_data = json.load(file)
|
logger.info(f"Successfully loaded process data from {process_file}")
|
except FileNotFoundError:
|
logger.error(f"Process file '{process_file}' not found.")
|
except json.JSONDecodeError:
|
logger.error(f"Error decoding JSON from the process file '{process_file}'.")
|
except Exception as e:
|
logger.error(f"Unexpected error when reading process file '{process_file}': {e}")
|
|
return process_data
|
|
def manage_processes(process_data):
|
"""定期检查进程状态并在进程停止时重新启动"""
|
if not process_data:
|
logging.error("process_data is empty, not starting the monitor thread")
|
return
|
# 停止进程
|
process_list = process_data.get("process_list", [])
|
for process in process_list:
|
pids = get_process_pids(process.get("process_name"))
|
for pid in pids:
|
try:
|
if os.path.exists(f"/proc/{pid}"):
|
os.kill(pid, 9)
|
logger.info(f"stop success {process} (PID: {pid})")
|
else:
|
logger.warning(f"Process {process} (PID: {pid}) does not exist")
|
except ProcessLookupError:
|
logger.warning(f"process {process} (PID: {pid}) is not exists")
|
except Exception as e:
|
logger.error(f"stop process {process} faild: {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"start process : {command_}")
|
except FileNotFoundError as e:
|
logger.error(f"terminel programe not find: {e}")
|
except Exception as e:
|
logger.error(f"start process faild: {e}")
|
|
def stop_process(pids):
|
"""停止指定PID的进程"""
|
for pid in pids:
|
try:
|
os.kill(pid, signal.SIGTERM) # Try graceful stop
|
logging.info(f"Sent SIGTERM to process with PID: {pid}")
|
time.sleep(1) # Wait for process to terminate
|
if os.path.exists(f"/proc/{pid}"):
|
os.kill(pid, signal.SIGKILL) # Force kill if still running
|
logging.info(f"Process {pid} killed with SIGKILL")
|
except ProcessLookupError:
|
logging.warning(f"Process with PID {pid} not found, might have exited already.")
|
except PermissionError:
|
logging.error(f"Permission denied while stopping process {pid}. Try running as root.")
|
except Exception as e:
|
logging.error(f"Error stopping process with PID {pid}: {e}")
|
|
|
def monitor_processes(process_data):
|
"""定期检查进程状态并在进程停止时重新启动"""
|
if not process_data:
|
logging.error("process_data is null, not starting the monitor thread")
|
return
|
|
process_list = process_data.get("process_list", [])
|
interval_seconds = process_data.get("interval_monitor_seconds", 10)
|
stop_start_hour = process_data.get("stop_start_hour", 0)
|
stop_end_hour = process_data.get("stop_end_hour", 5)
|
|
if not isinstance(interval_seconds, (int, float)) or interval_seconds <= 0:
|
logging.warning(f"the interval_seconds is not effective {interval_seconds}, defaulting to 10 seconds")
|
interval_seconds = 10 # 兜底策略
|
|
while True:
|
try:
|
current_time = datetime.now()
|
current_hour = current_time.hour
|
for process in process_list:
|
process_name = process.get("process_name")
|
if stop_start_hour <= current_hour < stop_end_hour and process_name!="Gate":
|
pids = get_process_pids(process_name)
|
if pids:
|
stop_process(pids)
|
else:
|
pids = get_process_pids(process_name)
|
if not pids:
|
logging.warning(f"Process {process_name} is not running, starting...")
|
start_process(process)
|
except Exception as e:
|
logging.error(f"Monitoring processes encountered an exception: {e}")
|
|
time.sleep(interval_seconds) # 按配置时间间隔检查
|
|
|
def stop_all_processes(process_data):
|
"""Stop all processes listed in the process_data."""
|
process_list = process_data.get("process_list", [])
|
for process in process_list:
|
process_name = process.get("process_name")
|
if process_name:
|
pids = get_process_pids(process_name)
|
for pid in pids:
|
try:
|
os.kill(pid, signal.SIGTERM) # Try graceful stop
|
logger.info(f"Sent SIGTERM to process: {process_name} (PID: {pid})")
|
time.sleep(1) # Wait for process to terminate
|
if os.path.exists(f"/proc/{pid}"):
|
os.kill(pid, signal.SIGKILL) # Force kill if still running
|
logger.info(f"Sent SIGKILL to process: {process_name} (PID: {pid})")
|
except ProcessLookupError:
|
logger.warning(f"Process {process_name} (PID: {pid}) not found.")
|
except Exception as e:
|
logger.error(f"Error stopping process {process_name} (PID: {pid}): {e}")
|
|
def restart_processes(process_data):
|
"""Restart all processes listed in the process_data."""
|
process_list = process_data.get("process_list", [])
|
for process in process_list:
|
start_process(process) # Restart the process using existing start_process function
|
|
def exit_handler(signum, frame, process_data):
|
"""Exit handler to stop all processes on script exit."""
|
logger.info("Exiting program, stopping all processes...")
|
stop_all_processes(process_data) # Stop all processes
|
sys.exit(0)
|
|
def clean_files(log_dir,log_expiry_time,file_dir,file_expiry_time):
|
"""清理过期的日志文件和录制文件"""
|
logger.info(f"[{datetime.now()}] Starting cleanup process...")
|
|
# 清理日志文件
|
for filename in os.listdir(log_dir):
|
file_path = os.path.join(log_dir, filename)
|
if os.path.isfile(file_path):
|
file_mtime = os.path.getmtime(file_path) # 获取文件的修改时间
|
if file_mtime < log_expiry_time: # 如果文件超过指定的清理时间
|
logger.info(f"Deleting log file: {file_path}")
|
os.remove(file_path)
|
|
# 清理录制文件
|
for filename in os.listdir(file_dir):
|
file_path = os.path.join(file_dir, filename)
|
if os.path.isfile(file_path):
|
file_mtime = os.path.getmtime(file_path) # 获取文件的修改时间
|
if file_mtime < file_expiry_time: # 如果文件超过指定的清理时间
|
logger.info(f"Deleting file: {file_path}")
|
os.remove(file_path)
|
|
logger.info(f"[{datetime.now()}] Cleanup process completed.")
|
|
def schedule_cleanup(config):
|
"""计划清理任务"""
|
try:
|
# 读取配置
|
cleanup_time = config['clean'].get('cleanup_time')
|
log_dir = config['clean'].get('log_dir')
|
log_expiry_days = int(config['clean'].get('log_expiry_days', 7)) # Default to 7 days if not set
|
file_dir = config['clean'].get('file_dir')
|
file_expiry_days = int(config['clean'].get('file_expiry_days', 7)) # Default to 7 days if not set
|
|
if not cleanup_time or not log_dir or not file_dir:
|
raise ValueError("Missing required configuration for cleanup: cleanup_time, log_dir, or file_dir.")
|
|
# 计算过期时间
|
log_expiry_time = time.time() - (log_expiry_days * 24 * 60 * 60)
|
file_expiry_time = time.time() - (file_expiry_days * 24 * 60 * 60)
|
|
# 设置清理任务
|
schedule_time = datetime.strptime(cleanup_time, '%H:%M:%S').time()
|
|
# 安排每天在指定时间清理文件
|
schedule.every().day.at(cleanup_time).do(clean_files, log_dir=log_dir, log_expiry_time=log_expiry_time,
|
file_dir=file_dir, file_expiry_time=file_expiry_time)
|
|
# 获取当前时间
|
now = datetime.now()
|
target_time = now.replace(hour=schedule_time.hour, minute=schedule_time.minute, second=schedule_time.second, microsecond=0)
|
|
# 如果目标时间已经过去，计划任务将推迟到明天
|
if now > target_time:
|
target_time += timedelta(days=1)
|
|
wait_time = (target_time - now).total_seconds()
|
logger.info(f"Waiting {wait_time} seconds until the next cleanup task.")
|
time.sleep(wait_time)
|
|
# 进入调度循环
|
while True:
|
schedule.run_pending()
|
time.sleep(60*60) # 等待下一次任务
|
|
except ValueError as ve:
|
logger.error(f"Configuration error: {ve}")
|
except KeyError as ke:
|
logger.error(f"Missing configuration key: {ke}")
|
except Exception as e:
|
logger.error(f"Unexpected error occurred in schedule_cleanup: {e}")
|
|
def main():
|
"""主函数"""
|
# 设置命令行参数解析
|
parser = argparse.ArgumentParser(description="Process some integers.")
|
parser.add_argument('--config', type=str, help="Path to the config file", default='msg_config.ini')
|
args = parser.parse_args()
|
|
config_path = args.config # 从命令行获取配置文件路径
|
config = read_config(config_path)
|
process_data = get_process_info(config)
|
signal.signal(signal.SIGINT, functools.partial(exit_handler, process_data=process_data))
|
signal.signal(signal.SIGTERM, functools.partial(exit_handler, process_data=process_data))
|
|
# 启动进程监控线程
|
monitor_thread = threading.Thread(target=monitor_processes, args=(process_data,), daemon=True)
|
monitor_thread.start()
|
|
# 启动清理任务调度线程
|
cleanup_thread = threading.Thread(target=schedule_cleanup, args=(config,), daemon=True)
|
cleanup_thread.start()
|
|
consumer = subscribe_kafka(config)
|
if not consumer:
|
return
|
|
for message in consumer:
|
try:
|
process_kafka_message(message.value, config, 1)
|
manage_processes(process_data)
|
except Exception as e:
|
logger.error(f"deal the Kafka messages exception: {e}")
|
|
if __name__ == "__main__":
|
main()
|
