cv_unet_image-colorization保姆级教程：日志监控与推理耗时统计埋点-编程实验室

cv_unet_image-colorization保姆级教程：日志监控与推理耗时统计埋点

1. 项目简介与监控需求

cv_unet_image-colorization是一个基于UNet架构的深度学习图像上色工具，能够将黑白照片自动转换为生动的彩色图像。在实际使用过程中，了解模型的运行状态和性能表现至关重要。

为什么需要监控和统计？当用户上传图片进行上色时，你可能会遇到这些问题：

不知道每次上色处理需要多长时间
无法确定系统在高负载下的表现如何
出现性能问题时难以定位瓶颈
缺少数据来优化用户体验

通过添加日志监控和推理耗时统计，你可以：

实时掌握系统运行状态
识别性能瓶颈并进行优化
为用户提供更准确的处理时间预估
建立系统健康度的量化指标

2. 环境准备与基础配置

在开始添加监控功能前，确保你的环境包含以下依赖：

# 基础依赖 pip install modelscope opencv-python torch streamlit Pillow numpy # 新增的监控相关依赖 pip install loguru # 更友好的日志库 pip install psutil # 系统资源监控

建议的目录结构：

your_project/ ├── app.py # 主应用文件 ├── utils/ │ ├── logging_setup.py # 日志配置 │ └── monitor.py # 监控工具函数 ├── logs/ │ ├── performance.log # 性能日志 │ └── system.log # 系统日志 └── stats/ # 统计数据存储

3. 日志系统搭建实战

3.1 基础日志配置

首先创建日志工具文件utils/logging_setup.py：

from loguru import logger import sys import os from datetime import datetime def setup_logging(): # 创建日志目录 log_dir = "logs" os.makedirs(log_dir, exist_ok=True) # 移除默认配置 logger.remove() # 添加控制台输出 logger.add( sys.stdout, format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>", level="INFO" ) # 添加性能日志文件 logger.add( os.path.join(log_dir, "performance_{time:YYYY-MM-DD}.log"), format="{time:YYYY-MM-DD HH:mm:ss} | {level} | {message}", level="INFO", rotation="1 day", # 每天轮换 retention="7 days", # 保留7天 filter=lambda record: "performance" in record["extra"] ) # 添加系统日志文件 logger.add( os.path.join(log_dir, "system_{time:YYYY-MM-DD}.log"), format="{time:YYYY-MM-DD HH:mm:ss} | {level} | {name} | {function} | {message}", level="DEBUG", rotation="1 day", retention="30 days" ) return logger # 创建日志实例 log = setup_logging()

3.2 在主应用中集成日志

修改你的主应用文件，添加日志功能：

import streamlit as st from utils.logging_setup import log from utils.monitor import PerformanceMonitor import time class ImageColorizationApp: def __init__(self): self.monitor = PerformanceMonitor() log.info("应用初始化完成") def process_image(self, image): """处理图像的主要函数，添加了监控埋点""" # 开始监控 self.monitor.start_inference() try: log.info("开始处理图像", extra={"performance": True}) log.debug(f"图像尺寸: {image.size}", extra={"performance": True}) # 记录初始内存使用 memory_before = self.monitor.get_memory_usage() # 这里是你的图像处理逻辑 start_time = time.time() result = self.colorize_image(image) processing_time = time.time() - start_time # 记录性能数据 memory_after = self.monitor.get_memory_usage() self.monitor.record_inference(processing_time, memory_after - memory_before) log.info(f"图像处理完成，耗时: {processing_time:.2f}秒", extra={"performance": True}) return result except Exception as e: log.error(f"图像处理失败: {str(e)}") self.monitor.record_error() raise def colorize_image(self, image): """具体的图像上色逻辑""" # 你的UNet模型推理代码在这里 # ... return colored_image

4. 推理耗时统计实现

4.1 性能监控工具类

创建utils/monitor.py来专门处理性能监控：

import time import psutil import json from datetime import datetime from pathlib import Path class PerformanceMonitor: def __init__(self): self.stats = { "total_inferences": 0, "successful_inferences": 0, "failed_inferences": 0, "total_processing_time": 0, "memory_usage": [], "inference_times": [] } self.current_start_time = None self.stats_file = Path("stats/performance_stats.json") self.stats_file.parent.mkdir(exist_ok=True) def start_inference(self): """开始推理计时""" self.current_start_time = time.time() process = psutil.Process() self.current_memory_start = process.memory_info().rss / 1024 / 1024 # MB def record_inference(self, processing_time, memory_delta): """记录推理完成""" self.stats["total_inferences"] += 1 self.stats["successful_inferences"] += 1 self.stats["total_processing_time"] += processing_time self.stats["inference_times"].append(processing_time) self.stats["memory_usage"].append(memory_delta) # 实时保存统计信息 self.save_stats() def record_error(self): """记录推理错误""" self.stats["total_inferences"] += 1 self.stats["failed_inferences"] += 1 self.save_stats() def get_memory_usage(self): """获取当前内存使用情况""" process = psutil.Process() return process.memory_info().rss / 1024 / 1024 # MB def get_performance_summary(self): """获取性能摘要""" if not self.stats["inference_times"]: return {"average_time": 0, "min_time": 0, "max_time": 0} times = self.stats["inference_times"] return { "average_time": sum(times) / len(times), "min_time": min(times), "max_time": max(times), "total_inferences": self.stats["total_inferences"], "success_rate": (self.stats["successful_inferences"] / self.stats["total_inferences"] * 100 if self.stats["total_inferences"] > 0 else 0) } def save_stats(self): """保存统计信息到文件""" with open(self.stats_file, 'w') as f: json.dump({ **self.stats, "last_updated": datetime.now().isoformat() }, f, indent=2) def load_stats(self): """从文件加载统计信息""" if self.stats_file.exists(): with open(self.stats_file, 'r') as f: self.stats = json.load(f)

4.2 在Streamlit界面中展示统计数据

在你的主应用中添加性能监控面板：

import streamlit as st import pandas as pd import plotly.express as px def show_performance_dashboard(monitor): """显示性能监控仪表板""" st.sidebar.header(" 性能监控") summary = monitor.get_performance_summary() st.sidebar.metric("平均处理时间", f"{summary['average_time']:.2f}秒") st.sidebar.metric("总处理次数", summary['total_inferences']) st.sidebar.metric("成功率", f"{summary['success_rate']:.1f}%") # 显示详细统计的扩展区域 with st.sidebar.expander("详细统计"): if monitor.stats["inference_times"]: # 创建时间分布图表 df = pd.DataFrame({ '处理时间(秒)': monitor.stats["inference_times"], '次数': range(len(monitor.stats["inference_times"])) }) fig = px.histogram(df, x='处理时间(秒)', title='处理时间分布', nbins=20) st.plotly_chart(fig, use_container_width=True) # 显示原始数据 st.write("最近10次处理时间:") st.dataframe(df.tail(10))

5. 完整集成示例

下面是完整的应用集成示例：

import streamlit as st from PIL import Image import numpy as np from utils.logging_setup import log from utils.monitor import PerformanceMonitor import time # 初始化应用 class ColorizationApp: def __init__(self): self.monitor = PerformanceMonitor() self.model = self.load_model() log.info("图像上色应用初始化完成") def load_model(self): """加载UNet模型""" # 你的模型加载代码 log.info("正在加载UNet模型...") # model = your_model_loading_code() log.info("模型加载完成") return "model_placeholder" def process_image(self, image): """处理图像并记录性能""" self.monitor.start_inference() try: log.info(f"开始处理图像，尺寸: {image.size}", extra={"performance": True}) # 模拟处理时间 processing_time = np.random.uniform(1.0, 3.0) # 替换为实际处理 time.sleep(processing_time) # 记录成功处理 self.monitor.record_inference(processing_time, 0) log.info(f"图像处理完成，耗时: {processing_time:.2f}秒", extra={"performance": True}) # 返回处理后的图像（这里用原始图像代替） return image except Exception as e: log.error(f"处理失败: {str(e)}") self.monitor.record_error() st.error(f"处理失败: {str(e)}") return None # 创建应用实例 app = ColorizationApp() # Streamlit界面 st.title(" AI图像上色工具") st.write("上传黑白照片，AI自动为其上色") uploaded_file = st.file_uploader("选择黑白图片", type=["jpg", "jpeg", "png"]) if uploaded_file is not None: image = Image.open(uploaded_file) st.image(image, caption="原始图像", use_column_width=True) if st.button(" 开始上色"): with st.spinner("AI正在为图像上色..."): result = app.process_image(image) if result: st.success("上色完成！") col1, col2 = st.columns(2) with col1: st.image(image, caption="原始图像", use_column_width=True) with col2: st.image(result, caption="上色结果", use_column_width=True) # 显示性能信息 summary = app.monitor.get_performance_summary() st.info(f"本次处理耗时: {summary['average_time']:.2f}秒") # 显示性能仪表板 show_performance_dashboard(app.monitor)

6. 高级监控功能

6.1 实时性能监控

添加实时性能监控功能：

def setup_realtime_monitoring(): """设置实时性能监控""" import threading import time def monitor_loop(): while True: # 监控系统资源 cpu_percent = psutil.cpu_percent() memory_info = psutil.virtual_memory() disk_usage = psutil.disk_usage('/') log.info( f"系统资源 - CPU: {cpu_percent}%, " f"内存: {memory_info.percent}%, " f"磁盘: {disk_usage.percent}%", extra={"performance": True} ) time.sleep(60) # 每分钟记录一次 # 启动监控线程 monitor_thread = threading.Thread(target=monitor_loop, daemon=True) monitor_thread.start()

6.2 异常检测与告警

添加简单的异常检测：

class AnomalyDetector: def __init__(self, window_size=10, threshold=2.0): self.window_size = window_size self.threshold = threshold self.recent_times = [] def check_anomaly(self, processing_time): """检测异常处理时间""" self.recent_times.append(processing_time) if len(self.recent_times) > self.window_size: self.recent_times.pop(0) if len(self.recent_times) >= 5: # 至少有5个数据点 mean = sum(self.recent_times) / len(self.recent_times) std_dev = (sum((x - mean) ** 2 for x in self.recent_times) / len(self.recent_times)) ** 0.5 if std_dev > 0 and abs(processing_time - mean) > self.threshold * std_dev: log.warning( f"检测到异常处理时间: {processing_time:.2f}秒, " f"平均: {mean:.2f}秒, 标准差: {std_dev:.2f}", extra={"performance": True} ) return True return False