贺州市网站建设_网站建设公司_虚拟主机_seo优化

智能电脑节能与安全控制系统

一、实际应用场景与痛点

应用场景

现代办公和家庭环境中，电脑使用存在以下场景：

- 临时离开：员工离开工位参加会议、午餐、临时外出

- 长时间空闲：下载文件、渲染视频等不需要人工干预的操作

- 多人共用电脑：实验室、图书馆、会议室等公共场所电脑

- 远程工作：在家办公时频繁离开电脑

- 夜间工作：程序员、设计师等工作到深夜可能忘记关机

痛点分析

1. 能源浪费：

- 空闲电脑持续耗电，显示器、外设保持运行

- 办公室夜间大量电脑未关机，电力消耗巨大

- 服务器、工作站24小时运行，利用率低

2. 安全风险：

- 临时离开时，敏感信息暴露

- 公共场所电脑被他人恶意操作

- 公司数据泄露风险

3. 设备损耗：

- 显示器长时间点亮加速老化

- 外设（打印机、音箱等）持续工作缩短寿命

- 电脑硬件长期高负荷运行

4. 用户体验差：

- 需要手动设置屏幕保护

- 每次离开都要记得锁定电脑

- 外设电源管理繁琐

5. 合规性要求：

- 企业安全政策要求自动锁定

- 节能环保考核指标

- 数据保护法规要求

二、核心逻辑讲解

系统架构

采用"监测-分析-决策-执行"的闭环控制系统：

数据采集层 → 状态分析层 → 智能决策层 → 执行控制层 → 反馈调整层

│ │ │ │ │

用户活动 空闲检测 策略选择 电源控制 效果评估

外设状态 能耗分析 条件判断 屏幕锁定 参数优化

系统负载 风险评估 优先级调度 进程管理 自适应学习

核心算法原理

1. 用户活动检测（多源融合）

- 输入设备监控：鼠标、键盘、触摸板

- 系统活动监控：CPU、内存、磁盘IO

- 进程监控：活跃应用程序

- 网络监控：网络流量、远程连接

2. 空闲状态识别（模糊逻辑）

空闲度 = w1×输入空闲 + w2×CPU空闲 + w3×进程空闲 + w4×网络空闲

其中：w1+w2+w3+w4=1，动态调整权重

3. 智能决策引擎（状态机+强化学习）

状态：正常 → 警告 → 轻度节能 → 深度节能 → 睡眠

触发条件：时间阈值、系统负载、用户习惯

奖励函数：节能效果 + 用户体验 - 恢复成本

4. 分级节能策略

第1级（5分钟）：降低屏幕亮度

第2级（10分钟）：关闭外设电源

第3级（20分钟）：锁定屏幕

第4级（30分钟）：睡眠模式

第5级（60分钟）：休眠模式

优化目标函数

总效益 = α·节能效果 + β·安全保护 + γ·设备寿命 - δ·用户干扰

约束条件：恢复时间 < 3秒，数据不丢失，用户体验优先

三、模块化代码实现

项目结构

smart_power_manager/

│

├── main.py # 主程序入口

├── config.py # 配置文件

├── requirements.txt # 依赖包

├── setup.py # 安装脚本

│

├── core/ # 核心算法模块

│ ├── __init__.py

│ ├── activity_monitor.py # 活动监控

│ ├── power_manager.py # 电源管理

│ ├── security_manager.py # 安全管理

│ ├── decision_engine.py # 决策引擎

│ └── adaptive_learner.py # 自适应学习

│

├── hardware/ # 硬件控制模块

│ ├── __init__.py

│ ├── display_control.py # 显示器控制

│ ├── usb_control.py # USB设备控制

│ ├── peripheral_control.py # 外设控制

│ └── system_power.py # 系统电源控制

│

├── sensors/ # 传感器模块

│ ├── __init__.py

│ ├── input_monitor.py # 输入设备监控

│ ├── system_monitor.py # 系统资源监控

│ ├── network_monitor.py # 网络活动监控

│ └── process_monitor.py # 进程监控

│

├── models/ # 数据模型

│ ├── __init__.py

│ ├── system_state.py

│ ├── power_policy.py

│ └── user_habit.py

│

├── utils/ # 工具函数

│ ├── __init__.py

│ ├── system_utils.py

│ ├── time_utils.py

│ ├── energy_calculator.py

│ └── logging_utils.py

│

├── database/ # 数据存储

│ ├── __init__.py

│ ├── activity_database.py

│ └── config_database.py

│

├── tests/ # 测试文件

│ ├── __init__.py

│ ├── test_activity_monitor.py

│ └── test_power_manager.py

│

├── docs/ # 文档

│ ├── API.md

│ └── USER_GUIDE.md

│

└── web_dashboard/ # Web管理界面

├── app.py

├── templates/

└── static/

核心代码实现

main.py

#!/usr/bin/env python3

"""

智能电脑节能与安全控制系统

基于用户活动的自适应节能与安全管理

"""

import os

import sys

import time

import signal

import logging

import threading

import json

from datetime import datetime, timedelta

from typing import Dict, List, Optional, Tuple

import queue

import atexit

import ctypes

import platform

# 导入自定义模块

from config import SystemConfig

from models.system_state import SystemState

from models.power_policy import PowerPolicy

from models.user_habit import UserHabit

from core.activity_monitor import ActivityMonitor

from core.power_manager import PowerManager

from core.security_manager import SecurityManager

from core.decision_engine import DecisionEngine

from core.adaptive_learner import AdaptiveLearner

from hardware.display_control import DisplayController

from hardware.usb_control import USBController

from hardware.peripheral_control import PeripheralController

from hardware.system_power import SystemPowerController

from sensors.input_monitor import InputMonitor

from sensors.system_monitor import SystemMonitor

from utils.energy_calculator import EnergyCalculator

from utils.logging_utils import setup_logging

from database.activity_database import ActivityDatabase

from database.config_database import ConfigDatabase

# 设置日志

logger = setup_logging()

# 全局变量

running = False

system_threads = []

class SmartPowerManager:

"""智能电源管理器主类"""

def __init__(self, config_path: str = 'config.yaml'):

"""

初始化智能电源管理器

Args:

config_path: 配置文件路径

"""

# 加载配置

self.config = SystemConfig(config_path)

# 初始化系统状态

self.system_state = SystemState()

# 初始化数据库

self.activity_db = ActivityDatabase(self.config.DATABASE_PATH)

self.config_db = ConfigDatabase(self.config.CONFIG_DB_PATH)

# 初始化用户习惯

self.user_habit = UserHabit()

# 初始化电源策略

self.power_policy = PowerPolicy(self.config)

# 初始化硬件控制器

self.display_controller = DisplayController()

self.usb_controller = USBController()

self.peripheral_controller = PeripheralController()

self.system_power = SystemPowerController()

# 初始化传感器

self.input_monitor = InputMonitor()

self.system_monitor = SystemMonitor()

# 初始化核心模块

self.activity_monitor = ActivityMonitor(

self.input_monitor,

self.system_monitor

)

self.power_manager = PowerManager(

self.display_controller,

self.usb_controller,

self.peripheral_controller,

self.system_power

)

self.security_manager = SecurityManager()

self.decision_engine = DecisionEngine(self.config)

self.adaptive_learner = AdaptiveLearner(self.config)

# 初始化工具

self.energy_calculator = EnergyCalculator()

# 消息队列

self.activity_queue = queue.Queue()

self.control_queue = queue.Queue()

# 线程控制

self.running = False

self.monitor_thread = None

self.decision_thread = None

self.control_thread = None

# 定时器

self.idle_timer = None

self.warning_timer = None

# 统计信息

self.stats = {

'start_time': datetime.now(),

'total_energy_saved': 0.0, # 节约的电量（kWh）

'total_time_saved': 0, # 节约的时间（秒）

'lock_count': 0, # 锁定次数

'sleep_count': 0, # 睡眠次数

'warning_shown': 0, # 警告显示次数

'user_overrides': 0, # 用户覆盖次数

'avg_idle_time': 0.0, # 平均空闲时间

'avg_recovery_time': 0.0 # 平均恢复时间

}

# 加载用户习惯

self._load_user_habits()

# 注册退出处理

atexit.register(self.cleanup)

# 设置信号处理

signal.signal(signal.SIGINT, self.signal_handler)

signal.signal(signal.SIGTERM, self.signal_handler)

logger.info("智能电源管理器初始化完成")

def _load_user_habits(self):

"""加载用户习惯"""

try:

habits = self.config_db.get_user_habits(self.config.USER_ID)

if habits:

self.user_habit.load_from_dict(habits)

logger.info("用户习惯加载成功")

# 根据用户习惯调整配置

self._adjust_config_by_habits()

except Exception as e:

logger.warning(f"加载用户习惯失败: {e}")

def _adjust_config_by_habits(self):

"""根据用户习惯调整配置"""

# 根据用户工作时间调整策略

work_hours = self.user_habit.get_work_hours()

if work_hours:

# 工作时间段使用工作模式策略

pass

# 根据用户活动模式调整阈值

activity_pattern = self.user_habit.get_activity_pattern()

if activity_pattern == 'frequent_breaks':

# 频繁离开的用户，使用较短的空闲阈值

self.config.IDLE_THRESHOLD = min(self.config.IDLE_THRESHOLD, 10 * 60) # 10分钟

elif activity_pattern == 'long_sessions':

# 长时间工作的用户，使用较长的空闲阈值

self.config.IDLE_THRESHOLD = max(self.config.IDLE_THRESHOLD, 30 * 60) # 30分钟

def _monitor_activity(self):

"""监控用户活动"""

logger.info("活动监控线程启动")

last_activity_time = time.time()

last_save_time = time.time()

while self.running:

try:

current_time = time.time()

# 1. 监控用户输入活动

input_activity = self.activity_monitor.get_input_activity()

# 2. 监控系统活动

system_activity = self.activity_monitor.get_system_activity()

# 3. 监控网络活动

network_activity = self.activity_monitor.get_network_activity()

# 4. 监控进程活动

process_activity = self.activity_monitor.get_process_activity()

# 5. 计算综合活动度

activity_level = self.activity_monitor.calculate_activity_level(

input_activity,

system_activity,

network_activity,

process_activity

)

# 6. 更新系统状态

self.system_state.update_activity(

input_activity=input_activity,

system_activity=system_activity,

network_activity=network_activity,

process_activity=process_activity,

activity_level=activity_level,

timestamp=current_time

)

# 7. 检查是否有活动

if activity_level > self.config.ACTIVITY_THRESHOLD:

last_activity_time = current_time

# 如果有活动且系统处于节能状态，恢复

if self.system_state.current_power_state != 'normal':

self._handle_user_return()

# 8. 计算空闲时间

idle_time = current_time - last_activity_time

self.system_state.idle_time = idle_time

# 9. 记录活动数据

activity_record = {

'timestamp': datetime.now(),

'input_activity': input_activity,

'system_activity': system_activity,

'network_activity': network_activity,

'process_activity': process_activity,

'activity_level': activity_level,

'idle_time': idle_time,

'power_state': self.system_state.current_power_state

}

self.activity_queue.put(activity_record)

# 10. 定期保存到数据库

if current_time - last_save_time > self.config.DATA_SAVE_INTERVAL:

self._save_activity_data()

last_save_time = current_time

# 控制监控频率

time.sleep(self.config.MONITOR_INTERVAL)

except Exception as e:

logger.error(f"活动监控错误: {e}")

time.sleep(5) # 出错等待5秒

def _save_activity_data(self):

"""保存活动数据到数据库"""

try:

# 从队列获取数据

records = []

while not self.activity_queue.empty():

try:

record = self.activity_queue.get_nowait()

records.append(record)

except queue.Empty:

break

if records:

self.activity_db.save_activity_records(records)

logger.debug(f"保存 {len(records)} 条活动记录")

except Exception as e:

logger.error(f"保存活动数据失败: {e}")

def _make_decisions(self):

"""决策引擎线程"""

logger.info("决策引擎线程启动")

last_decision_time = 0

while self.running:

try:

current_time = time.time()

# 控制决策频率

if current_time - last_decision_time < self.config.DECISION_INTERVAL:

time.sleep(0.1)

continue

# 获取当前状态

current_state = {

'idle_time': self.system_state.idle_time,

'activity_level': self.system_state.activity_level,

'power_state': self.system_state.current_power_state,

'system_load': self.system_state.system_load,

'battery_level': self.system_state.battery_level,

'time_of_day': datetime.now().hour

}

# 获取用户习惯

user_context = self.user_habit.get_current_context()

# 使用决策引擎做出决策

decision = self.decision_engine.make_decision(

current_state,

user_context

)

if decision['action'] != 'none':

# 记录决策

decision_record = {

'timestamp': datetime.now(),

'current_state': current_state,

'decision': decision,

'user_context': user_context

}

# 将决策放入控制队列

self.control_queue.put(decision_record)

logger.info(f"决策: {decision['action']}, 优先级: {decision['priority']}")

last_decision_time = current_time

# 短暂休眠避免CPU占用过高

time.sleep(0.5)

except Exception as e:

logger.error(f"决策引擎错误: {e}")

time.sleep(1)

def _execute_controls(self):

"""执行控制线程"""

logger.info("执行控制线程启动")

while self.running:

try:

if not self.control_queue.empty():

control_record = self.control_queue.get_nowait()

# 执行控制动作

self._execute_control_action(control_record)

else:

time.sleep(0.1) # 队列为空时短暂等待

except Exception as e:

logger.error(f"执行控制错误: {e}")

time.sleep(1)

def _execute_control_action(self, control_record: Dict):

"""执行控制动作"""

decision = control_record['decision']

action = decision['action']

priority = decision['priority']

try:

if action == 'reduce_brightness':

# 降低屏幕亮度

brightness_level = decision.get('parameters', {}).get('level', 50)

self.power_manager.reduce_display_brightness(brightness_level)

# 更新系统状态

self.system_state.current_power_state = 'dimmed'

self.system_state.display_brightness = brightness_level

logger.info(f"降低屏幕亮度到 {brightness_level}%")

elif action == 'turn_off_peripherals':

# 关闭外设电源

self.power_manager.turn_off_peripherals()

# 更新系统状态

self.system_state.peripherals_powered = False

logger.info("关闭外设电源")

elif action == 'lock_screen':

# 锁定屏幕

self.security_manager.lock_screen()

# 更新系统状态

self.system_state.screen_locked = True

self.system_state.current_power_state = 'locked'

# 统计

self.stats['lock_count'] += 1

logger.info("锁定屏幕")

elif action == 'sleep_mode':

# 进入睡眠模式

self.power_manager.enter_sleep_mode()

# 更新系统状态

self.system_state.current_power_state = 'sleeping'

# 统计

self.stats['sleep_count'] += 1

logger.info("进入睡眠模式")

elif action == 'hibernate':

# 进入休眠模式

self.power_manager.enter_hibernate_mode()

# 更新系统状态

self.system_state.current_power_state = 'hibernated'

logger.info("进入休眠模式")

elif action == 'show_warning':

# 显示警告

message = decision.get('parameters', {}).get('message', '系统将进入节能模式')

self._show_warning_message(message)

# 统计

self.stats['warning_shown'] += 1

logger.info(f"显示警告: {message}")

elif action == 'restore_normal':

# 恢复正常状态

self._restore_normal_state()

logger.info("恢复正常状态")

# 记录执行结果

self._record_control_result(control_record, success=True)

except Exception as e:

logger.error(f"执行控制动作失败: {e}")

self._record_control_result(control_record, success=False, error=str(e))

def _show_warning_message(self, message: str):

"""显示警告消息"""

try:

if platform.system() == 'Windows':

# Windows系统显示对话框

ctypes.windll.user32.MessageBoxW(0, message, "节能提示", 0x40)

elif platform.system() == 'Darwin':

# macOS系统显示通知

os.system(f'osascript -e \'display notification "{message}" with title "节能提示"\'')

else:

# Linux系统使用notify-send

os.system(f'notify-send "节能提示" "{message}"')

except Exception as e:

logger.warning(f"显示警告消息失败: {e}")

def _handle_user_return(self):

"""处理用户返回"""

try:

# 保存节能前的状态

previous_state = self.system_state.current_power_state

# 恢复外设电源

if not self.system_state.peripherals_powered:

self.power_manager.turn_on_peripherals()

self.system_state.peripherals_powered = True

# 恢复屏幕亮度

if self.system_state.display_brightness < 100:

self.power_manager.restore_display_brightness()

self.system_state.display_brightness = 100

# 解锁屏幕（如果需要）

if self.system_state.screen_locked:

# 注意：解锁需要用户输入密码，这里只记录状态

self.system_state.screen_locked = False

# 更新电源状态

self.system_state.current_power_state = 'normal'

# 计算恢复时间

recovery_start = time.time()

# 执行恢复操作

if previous_state in ['sleeping', 'hibernated']:

# 从睡眠/休眠恢复，需要额外处理

self.power_manager.wake_from_sleep()

recovery_time = time.time() - recovery_start

self.stats['avg_recovery_time'] = (

0.9 * self.stats['avg_recovery_time'] + 0.1 * recovery_time

)

# 记录用户返回事件

return_event = {

'timestamp': datetime.now(),

'previous_state': previous_state,

'recovery_time': recovery_time,

'activity_level': self.system_state.activity_level

}

self.activity_db.save_user_return_event(return_event)

logger.info(f"用户返回，从 {previous_state} 状态恢复，耗时 {recovery_time:.2f}秒")

except Exception as e:

logger.error(f"处理用户返回失败: {e}")

def _record_control_result(self, control_record: Dict, success: bool,

error: str = ''):

"""记

如果你觉得这个工具好用，欢迎关注我！