Add BMM_Theory: bmm_theory/branches/base.py
This commit is contained in:
63
BMM/BMM_Theory/bmm_theory/branches/base.py
Normal file
63
BMM/BMM_Theory/bmm_theory/branches/base.py
Normal file
@@ -0,0 +1,63 @@
|
|||||||
|
"""分支基类: 统一的三分支接口.
|
||||||
|
|
||||||
|
每个分支实现三个方法:
|
||||||
|
capable(case) -> (是否可进入, 逐条条件判定结果)
|
||||||
|
plan(case) -> 理论最优实现方案 (ImplPlan)
|
||||||
|
evaluate(case, plan) -> 硬件时延评估 (HardwareTiming + 可行性)
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from dataclasses import dataclass, field
|
||||||
|
|
||||||
|
from ..hardware import NpuSpec, ASCEND950PR
|
||||||
|
from ..models import BmmCase, ImplPlan, HardwareTiming
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class ConditionCheck:
|
||||||
|
"""一条进入条件的判定结果."""
|
||||||
|
name: str
|
||||||
|
passed: bool
|
||||||
|
detail: str = ""
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class BranchResult:
|
||||||
|
capable: bool
|
||||||
|
checks: list = field(default_factory=list) # list[ConditionCheck]
|
||||||
|
plan: ImplPlan | None = None
|
||||||
|
timing: HardwareTiming | None = None
|
||||||
|
note: str = ""
|
||||||
|
|
||||||
|
def failed_conditions(self) -> str:
|
||||||
|
return "; ".join(c.name for c in self.checks if not c.passed)
|
||||||
|
|
||||||
|
|
||||||
|
class Branch:
|
||||||
|
"""分支基类."""
|
||||||
|
|
||||||
|
name: str = "Base"
|
||||||
|
|
||||||
|
def __init__(self, spec: NpuSpec = ASCEND950PR):
|
||||||
|
self.spec = spec
|
||||||
|
|
||||||
|
# ---- 子类实现 ----
|
||||||
|
def check_conditions(self, case: BmmCase) -> list:
|
||||||
|
raise NotImplementedError
|
||||||
|
|
||||||
|
def make_plan(self, case: BmmCase) -> ImplPlan:
|
||||||
|
raise NotImplementedError
|
||||||
|
|
||||||
|
def evaluate(self, case: BmmCase, plan: ImplPlan) -> HardwareTiming:
|
||||||
|
raise NotImplementedError
|
||||||
|
|
||||||
|
# ---- 公共流程 ----
|
||||||
|
def analyze(self, case: BmmCase) -> BranchResult:
|
||||||
|
checks = self.check_conditions(case)
|
||||||
|
capable = all(c.passed for c in checks)
|
||||||
|
res = BranchResult(capable=capable, checks=checks)
|
||||||
|
if capable:
|
||||||
|
res.plan = self.make_plan(case)
|
||||||
|
res.timing = self.evaluate(case, res.plan)
|
||||||
|
return res
|
||||||
Reference in New Issue
Block a user