123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106 |
- """
- 自定义指标模板
- 用户可以基于此模板创建自己的指标
- """
- from typing import Dict, Any
- import numpy as np
- from modules.lib.score import Score
- from modules.lib.metric_registry import BaseMetric
- METRIC_CATEGORY = "custom"
- class CustomMetricExample(BaseMetric):
- """自定义指标示例 - 计算平均速度"""
-
- def __init__(self, data: Any):
- """
- 初始化指标
-
- Args:
- data: 输入数据
- """
- super().__init__(data)
-
-
- def calculate(self) -> Dict[str, Any]:
- """
- 计算指标
-
- Returns:
- 计算结果字典
- """
-
- result = {
- "value": 0.0,
- "score": 100,
- "details": {}
- }
-
-
- try:
- if hasattr(self.data, 'velocities') and self.data.velocities:
- velocities = self.data.velocities
- if isinstance(velocities, dict) and 'vx' in velocities and 'vy' in velocities:
-
- vx = np.array(velocities['vx'])
- vy = np.array(velocities['vy'])
- speeds = np.sqrt(vx**2 + vy**2)
-
-
- avg_speed = np.mean(speeds)
- result['value'] = float(avg_speed)
-
-
- if avg_speed < 10:
- result['score'] = 60
- elif avg_speed > 50:
- result['score'] = 70
- else:
- result['score'] = 100
-
-
- result['details'] = {
- "max_speed": float(np.max(speeds)),
- "min_speed": float(np.min(speeds)),
- "std_speed": float(np.std(speeds))
- }
- except Exception as e:
-
- result['value'] = 0.0
- result['score'] = 0
- result['details'] = {"error": str(e)}
-
- return result
-
- def report_statistic(self) -> Dict[str, Any]:
- """
- 报告统计结果
- 可以在这里自定义结果格式
- """
- result = self.calculate()
-
-
-
-
- return result
- class AnotherCustomMetric(BaseMetric):
- """另一个自定义指标示例 - 计算加速度变化率"""
-
- def __init__(self, data: Any):
- super().__init__(data)
-
- def calculate(self) -> Dict[str, Any]:
-
- return {"value": 0.0, "score": 100, "details": {}}
|