3. 设计设备分组与升级策略

任务目标

本案例实现一个设备分组与灰度升级策略模块，支持按设备型号、当前固件版本、地域等条件动态筛选设备并形成分组，同时为每个分组配置灰度升级策略（如分批比例、批次间隔、暂停条件）。最终通过一个命令行入口演示：给定设备列表与分组规则，输出各分组命中的设备清单及灰度批次划分结果。

核心设计

分组规则采用条件表达式描述，格式为 字段:操作符:值，多个条件之间为 AND 关系。支持字段：model（型号）、version（固件版本）、region（地域）。操作符支持 eq（等于）、ne（不等于）、contains（包含，用于版本前缀匹配）。灰度策略包含：总批次 totalBatches、每批间隔分钟 intervalMinutes、暂停条件 pauseOnFailureRate（失败率超过该值则暂停后续批次）。

完整实现

import json
from dataclasses import dataclass, field
from typing import List, Dict, Any

# ---------- 数据模型 ----------
@dataclass
class Device:
    device_id: str
    model: str
    version: str
    region: str

@dataclass
class GroupRule:
    name: str
    conditions: List[str]  # ["model:eq:ESP32-C3", "version:contains:2.1."]
    total_batches: int = 3
    interval_minutes: int = 30
    pause_on_failure_rate: float = 0.1

# ---------- 分组引擎 ----------
def _match_condition(device: Device, condition: str) -> bool:
    field_name, op, expected = condition.split(":", 2)
    actual = getattr(device, field_name)
    if op == "eq":
        return actual == expected
    elif op == "ne":
        return actual != expected
    elif op == "contains":
        return expected in actual
    else:
        raise ValueError(f"Unsupported operator: {op}")

def match_group(device: Device, rule: GroupRule) -> bool:
    return all(_match_condition(device, c) for c in rule.conditions)

# ---------- 灰度批次划分 ----------
def assign_batch(device_indices: List[int], total_batches: int) -> Dict[int, List[int]]:
    """将设备索引按顺序均分到 total_batches 个批次中"""
    batches = {i: [] for i in range(1, total_batches + 1)}
    for idx, device_idx in enumerate(device_indices):
        batch_no = (idx % total_batches) + 1
        batches[batch_no].append(device_idx)
    return batches

# ---------- 主流程 ----------
def evaluate_groups(devices: List[Device], rules: List[GroupRule]) -> Dict[str, Any]:
    result = {}
    for rule in rules:
        matched_indices = [i for i, d in enumerate(devices) if match_group(d, rule)]
        batches = assign_batch(matched_indices, rule.total_batches)
        result[rule.name] = {
            "matched_count": len(matched_indices),
            "batches": {
                str(k): [devices[i].device_id for i in v]
                for k, v in batches.items()
            },
            "strategy": {
                "interval_minutes": rule.interval_minutes,
                "pause_on_failure_rate": rule.pause_on_failure_rate
            }
        }
    return result

# ---------- 演示入口 ----------
if __name__ == "__main__":
    devices = [
        Device("dev-001", "ESP32-C3", "2.1.0", "cn-east"),
        Device("dev-002", "ESP32-C3", "2.1.3", "cn-east"),
        Device("dev-003", "ESP32-C3", "2.0.9", "cn-north"),
        Device("dev-004", "ESP32-S3", "2.1.1", "cn-east"),
        Device("dev-005", "ESP32-C3", "2.1.5", "cn-south"),
        Device("dev-006", "ESP32-S3", "2.0.8", "cn-north"),
    ]

    rules = [
        GroupRule(
            name="C3-2.1灰度",
            conditions=["model:eq:ESP32-C3", "version:contains:2.1."],
            total_batches=2,
            interval_minutes=60,
            pause_on_failure_rate=0.05
        ),
        GroupRule(
            name="S3-全部",
            conditions=["model:eq:ESP32-S3"],
            total_batches=3,
            interval_minutes=15,
            pause_on_failure_rate=0.1
        )
    ]

    result = evaluate_groups(devices, rules)
    print(json.dumps(result, indent=2, ensure_ascii=False))

运行与验证

运行方式：

python device_grouping.py

预期输出（由代码逻辑直接推出）：

{
  "C3-2.1灰度": {
    "matched_count": 3,
    "batches": {
      "1": ["dev-001", "dev-003"],
      "2": ["dev-002", "dev-005"]
    },
    "strategy": {
      "interval_minutes": 60,
      "pause_on_failure_rate": 0.05
    }
  },
  "S3-全部": {
    "matched_count": 2,
    "batches": {
      "1": ["dev-004"],
      "2": ["dev-006"],
      "3": []
    },
    "strategy": {
      "interval_minutes": 15,
      "pause_on_failure_rate": 0.1
    }
  }
}

验证要点：

• 分组“C3-2.1灰度”正确匹配型号为ESP32-C3且版本号包含“2.1.”的3台设备（dev-001、dev-002、dev-005），dev-003因版本为2.0.9被排除。
• 批次划分采用轮询分配：3台设备分2批，第1批2台、第2批1台，与输出一致。
• 分组“S3-全部”匹配2台ESP32-S3设备，分3批时空批次保留在结果中，便于后续调度逻辑感知。
• 修改devices列表或rules条件后重新运行，输出应随输入变化且分组逻辑保持一致。

灰度策略说明

本实现将灰度策略拆分为两部分：批次划分（由assign_batch完成）和策略参数（间隔时间、暂停阈值随分组结果输出）。实际OTA后台在消费该结果时，按批次顺序推送升级任务，每批完成后观察失败率，若超过pause_on_failure_rate则停止后续批次。本案例聚焦分组与批次生成逻辑，不包含实际推送与监控实现。