chronaxis.fast.totals

源码基线：8904c2f744951857659abd40aaa3535d57faf53c

1"""按独立文件模板的声明顺序汇总同一业务指标。只在请求 --total 时使用。"""23from dataclasses import dataclass4import json5import math6from pathlib import Path789def _semantics(spec):10    """比较业务绑定、计算指令、时间范围及解码选项。单文件路径和展示格式可不同。"""11    settings = {key: value for key, value in spec.items()12                if key not in ('path', 'dbc', 'table_format', '_database', 'inputs')}13    if 'inputs' in spec:14        settings['input_options'] = [{k:v for k,v in item.items() if k != 'path'} for item in spec['inputs']]15    return Path(spec['dbc']).resolve(), json.dumps(settings, sort_keys=True, allow_nan=False)161718@dataclass(frozen=True)19class FastTotals:20    """在开始读文件前冻结可求和的指标，以及这些指标应当来自哪些任务。"""2122    members: tuple23    metrics: tuple2425    @classmethod26    def from_specs(cls, specs, *, template):27        """specs 为按模板输入顺序排列的 job_id→fast 声明。返回只读求和计划。2829        只接受同模板、可核实相同绑定和累计口径的不同文件。失败发生在任务分派前。30        """31        if template is not True:32            raise ValueError('--total 仅支持同一分析模板展开的独立文件')33        if not specs:34            raise ValueError('--total 没有预期文件')35        first = next(iter(specs.values()))36        expected = _semantics(first)37        metrics = tuple((item['name'], item['unit']) for item in first['metrics'])38        if not metrics or len(metrics) != len({name for name, _ in metrics}):39            raise ValueError('批量总计需要互不重复的计算项')40        files = set()41        for job, spec in specs.items():42            if _semantics(spec) != expected:43                raise ValueError(f'{job} 与首文件的 DBC、绑定、公式、方向或时间口径不同')44            path = Path(spec['path']).resolve()45            if path in files:46                raise ValueError(f'{job} 重复选择了同一个输入文件')47            files.add(path)48        return cls(tuple(specs), metrics)4950    def finish(self, outcomes):51        """按计算项和方向桶分别求和。失败/取消/未运行任务须显式写为 None。5253        outcomes 的正常值是 worker 返回的未舍入行。成员或行不符合计划属于协议错误。54        数值缺失导致该项总计不可用。有限加数之和溢出时抛出 OverflowError。55        """56        if set(outcomes) != set(self.members):57            raise ValueError('批量结果成员与启动前冻结的文件清单不一致')58        schema = dict(self.metrics)59        sums = dict.fromkeys(schema, 0.0)60        unavailable = {}61        for job in self.members:62            rows = outcomes[job]63            if rows is None:64                for name in schema:65                    unavailable[name] = 'member_not_completed'66                continue67            seen = set()68            for row in rows:69                name = row['name']70                if name not in schema or name in seen or row['unit'] != schema[name]:71                    raise ValueError(f'{job} 返回了未知、重复或单位不符的计算项 {name}')72                seen.add(name)73                value = row['value']74                if value is None and row['status'] == 'indeterminate':75                    unavailable.setdefault(name, 'member_value_unknown')76                elif row['status'] != 'complete' or type(value) not in (float, int):77                    raise ValueError(f'{job} 的 {name} 数值与状态不符')78                elif not math.isfinite(value) or float(value) != value:79                    raise ValueError(f'{job} 的 {name} 不能无损表示为有限 float64')80                else:81                    sums[name] += value82            if seen != schema.keys():83                raise ValueError(f'{job} 缺少预期计算项')84        result = []85        for name, unit in self.metrics:86            reason = unavailable.get(name)87            value = None88            if reason is None:89                value = sums[name]90                if not math.isfinite(value):91                    raise OverflowError(f'批量总计 {name} 溢出')92            result.append(dict(name=name, unit=unit, value=value,93                               status='complete' if reason is None else 'indeterminate', reason=reason))94        return result95