4. 状态切换事件模拟器：编写模拟用户/设备状态切换的事件生成器，支持随机与预设模式

在联合调试工程中，真实的状态切换数据往往难以获取或复现。状态切换事件模拟器（State Transition Event Simulator）正是为了解决这一痛点而设计：它能够按照指定的规则批量生成用户或设备的状态变更日志，为后续的 Log 分析、异常检测、告警策略验证提供稳定、可控的测试数据源。

4.1 模拟器架构与核心组件

事件模拟器采用模块化设计，由以下四个核心组件构成：

• 状态机引擎（State Machine Engine）：定义合法的状态集合与允许的转换路径，防止生成非法跳转。
• 事件调度器（Event Scheduler）：控制事件生成的时间分布，支持均匀分布、泊松分布、固定间隔等模式。
• 模式控制器（Mode Controller）：切换随机模式与预设模式，预设模式可加载 JSON/YAML 格式的场景剧本。
• 输出格式化器（Output Formatter）：将事件序列转换为标准 Log 格式（JSON、CSV、Syslog 等）。

4.2 状态机定义与转换规则

以典型的 用户在线状态 为例，定义以下状态集合与合法转换路径：

4.3 随机模式：概率驱动的状态跳转

随机模式下，模拟器根据预设的 转换概率矩阵 决定下一个状态。每个状态到其他合法状态的转换概率之和为 1.0。

示例概率矩阵（以当前状态为行，目标状态为列）：

随机模式生成流程：

1. 初始化所有设备/用户为 OFFLINE 状态。
2. 根据时间调度器（如每 30 秒触发一次）选择一批活跃实体。
3. 对每个实体，根据当前状态查找概率矩阵，使用加权随机算法选择目标状态。
4. 记录事件：时间戳 | 实体ID | 源状态 | 目标状态 | 触发原因=随机转换。
5. 更新实体状态，进入下一轮调度。

4.4 预设模式：剧本驱动的场景复现

预设模式允许用户编写 场景剧本（Scenario Script），精确控制每个实体在特定时间点的状态切换。剧本采用 JSON 格式，结构清晰，易于版本管理。

剧本文件结构示例：

{
  "scenario_name": "用户高峰时段状态波动",
  "entities": [
    {
      "entity_id": "user_001",
      "initial_state": "OFFLINE",
      "events": [
        {"time_offset": 0,   "target_state": "ONLINE",    "reason": "上班签到"},
        {"time_offset": 120, "target_state": "BUSY",      "reason": "晨会"},
        {"time_offset": 300, "target_state": "ONLINE",    "reason": "会议结束"},
        {"time_offset": 600, "target_state": "AWAY",      "reason": "午休"},
        {"time_offset": 720, "target_state": "ONLINE",    "reason": "午休结束"},
        {"time_offset": 900, "target_state": "OFFLINE",   "reason": "下班签退"}
      ]
    },
    {
      "entity_id": "device_gw_01",
      "initial_state": "ONLINE",
      "events": [
        {"time_offset": 0,   "target_state": "ONLINE",    "reason": "设备启动"},
        {"time_offset": 500, "target_state": "OFFLINE",   "reason": "网络断连"},
        {"time_offset": 530, "target_state": "ONLINE",    "reason": "网络恢复"}
      ]
    }
  ]
}

4.5 双模式对比与选型建议

4.6 代码实现：核心引擎骨架

以下 Python 代码实现了模拟器的核心调度逻辑，支持随机与预设两种模式的无缝切换：

import json
import random
import time
from datetime import datetime, timedelta

class StateTransitionSimulator:
    """状态切换事件模拟器"""

    def __init__(self, state_machine, mode='random', seed=None):
        self.state_machine = state_machine      # 状态机定义
        self.mode = mode                        # 'random' | 'preset'
        self.entities = {}                      # entity_id -> current_state
        self.event_log = []
        if seed is not None:
            random.seed(seed)

    def load_preset_scenario(self, scenario_path):
        """加载预设剧本"""
        with open(scenario_path, 'r', encoding='utf-8') as f:
            scenario = json.load(f)
        for entity in scenario['entities']:
            eid = entity['entity_id']
            self.entities[eid] = {
                'state': entity['initial_state'],
                'events': entity['events'],
                'event_index': 0
            }

    def add_entity(self, entity_id, initial_state):
        """添加实体（随机模式使用）"""
        self.entities[entity_id] = {
            'state': initial_state,
            'events': None,
            'event_index': 0
        }

    def step(self, current_time):
        """执行一步模拟，返回该时刻生成的事件列表"""
        events_this_step = []
        for eid, entity in self.entities.items():
            if self.mode == 'preset':
                # 预设模式：按剧本时间偏移触发
                while (entity['event_index'] < len(entity['events']) and
                       entity['events'][entity['event_index']]['time_offset'] <= current_time):
                    ev = entity['events'][entity['event_index']]
                    target = ev['target_state']
                    if self._is_valid_transition(entity['state'], target):
                        events_this_step.append({
                            'timestamp': datetime.now().isoformat(),
                            'entity_id': eid,
                            'from_state': entity['state'],
                            'to_state': target,
                            'reason': ev.get('reason', 'preset')
                        })
                        entity['state'] = target
                    entity['event_index'] += 1
            else:
                # 随机模式：按概率选择下一个状态
                if random.random() < 0.3:  # 30% 概率触发转换
                    target = self._random_next_state(entity['state'])
                    if target and target != entity['state']:
                        events_this_step.append({
                            'timestamp': datetime.now().isoformat(),
                            'entity_id': eid,
                            'from_state': entity['state'],
                            'to_state': target,
                            'reason': 'random_transition'
                        })
                        entity['state'] = target
        return events_this_step

    def _is_valid_transition(self, from_state, to_state):
        """校验状态转换合法性"""
        return to_state in self.state_machine.get(from_state, [])

    def _random_next_state(self, current_state):
        """根据概率矩阵随机选择下一个状态"""
        transitions = self.state_machine.get(current_state, {})
        if not transitions:
            return None
        states = list(transitions.keys())
        probs = list(transitions.values())
        return random.choices(states, weights=probs, k=1)[0]

    def run(self, duration_seconds=3600, interval_seconds=30):
        """运行模拟器，生成指定时长的事件流"""
        for t in range(0, duration_seconds, interval_seconds):
            events = self.step(t)
            self.event_log.extend(events)
        return self.event_log

4.7 输出示例与 Log 格式

模拟器生成的每条事件记录包含以下字段，可直接写入文件或发送到消息队列：

{
  "timestamp": "2025-03-21T14:23:10.456789",
  "entity_id": "user_042",
  "from_state": "ONLINE",
  "to_state": "BUSY",
  "reason": "random_transition",
  "mode": "random",
  "session_id": "sim-20250321-001"
}

CSV 格式输出示例：

timestamp,entity_id,from_state,to_state,reason,mode
2025-03-21T14:23:10.456789,user_042,ONLINE,BUSY,random_transition,random
2025-03-21T14:23:40.123456,device_gw_01,ONLINE,OFFLINE,network_disconnect,preset

4.8 本章小结