chentianrui
369 lines
12 KiB
Python
369 lines
12 KiB
Python
from abc import ABC, abstractmethod
|
|
from typing import Dict, List, Any, Optional, Set, Tuple
|
|
|
|
|
|
class CalculationStrategy(ABC):
|
|
"""
|
|
计算策略接口
|
|
|
|
定义了计算过程中可以自定义的各种方法。
|
|
每种软件类型可以提供自己的实现，以修改计算过程中的任何规则。
|
|
"""
|
|
|
|
@abstractmethod
|
|
def calculate_fee_base(
|
|
self,
|
|
fee_base: str,
|
|
cost_table: Dict[str, Any],
|
|
project_node: Dict[str, Any],
|
|
context: Any,
|
|
in_calculation: Optional[Set[str]] = None,
|
|
) -> float:
|
|
"""
|
|
计算取费基数
|
|
|
|
Args:
|
|
fee_base: 取费基数表达式
|
|
cost_table: 取费表
|
|
project_node: 项目节点
|
|
context: 计算上下文
|
|
in_calculation: 正在计算中的费用代码集合，用于检测循环依赖
|
|
|
|
Returns:
|
|
float: 计算结果
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def calculate_internal_fee(
|
|
self,
|
|
fee_code: str,
|
|
cost_table: Dict[str, Any],
|
|
project_node: Dict[str, Any],
|
|
context: Any,
|
|
in_calculation: Optional[Set[str]] = None,
|
|
) -> float:
|
|
"""
|
|
计算表内费用
|
|
|
|
Args:
|
|
fee_code: 费用代码
|
|
cost_table: 取费表
|
|
project_node: 项目节点
|
|
context: 计算上下文
|
|
in_calculation: 正在计算中的费用代码集合，用于检测循环依赖
|
|
|
|
Returns:
|
|
float: 计算结果
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def calculate_external_variable(self, var_name: str, context: Any) -> float:
|
|
"""
|
|
计算表外变量
|
|
|
|
Args:
|
|
var_name: 变量名
|
|
context: 计算上下文
|
|
|
|
Returns:
|
|
float: 计算结果
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def find_fee_item_by_code(self, cost_table: Dict[str, Any], code: str) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
在取费表中查找指定代码的费用项
|
|
|
|
Args:
|
|
cost_table: 取费表
|
|
code: 费用代码
|
|
|
|
Returns:
|
|
Optional[Dict[str, Any]]: 找到的费用项，如果未找到则返回None
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def find_fee_base_nodes(self, node: Dict[str, Any], result: Optional[List] = None) -> List[Dict[str, Any]]:
|
|
"""
|
|
递归查找取费表中包含"取费基数"的节点或有子节点的费用项
|
|
|
|
Args:
|
|
node: 取费表节点
|
|
result: 结果列表
|
|
|
|
Returns:
|
|
List[Dict[str, Any]]: 包含"取费基数"的节点或有子节点的费用项列表
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def calculate_all_fees(
|
|
self,
|
|
project_node: Dict[str, Any],
|
|
cost_table: Dict[str, Any],
|
|
json_file_path: Optional[str] = None,
|
|
engineering_type: Optional[str] = None,
|
|
) -> Dict[str, float]:
|
|
"""
|
|
计算所有费用
|
|
|
|
Args:
|
|
project_node: 项目节点
|
|
cost_table: 取费表
|
|
json_file_path: JSON文件路径
|
|
engineering_type: 工程类型
|
|
|
|
Returns:
|
|
Dict[str, float]: 计算结果，键为费用名称，值为计算结果
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def calculate_rcj_count(
|
|
self,
|
|
rcj_nodes: List[Tuple[Dict[str, Any], str]],
|
|
project_children: List[Dict[str, Any]],
|
|
json_file_path: Optional[str] = None,
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
计算人材机节点的数量，考虑父级消耗量
|
|
|
|
Args:
|
|
rcj_nodes: 人材机节点列表，每个元素是(节点, 父级ID)元组
|
|
project_children: 项目划分级别下的所有工程量节点
|
|
json_file_path: JSON文件路径，用于获取工程信息
|
|
|
|
Returns:
|
|
List[Dict[str, Any]]: 计算数量后的人材机节点列表
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def cat_rcj_count(self, rcj_nodes: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|
"""
|
|
汇总人材机节点的数量
|
|
|
|
Args:
|
|
rcj_nodes: 人材机节点列表
|
|
|
|
Returns:
|
|
List[Dict[str, Any]]: 汇总后的人材机节点列表
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def calc_rcj_fee(self, rcj_nodes: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|
"""
|
|
计算人材机节点的费用
|
|
|
|
Args:
|
|
rcj_nodes: 人材机节点列表
|
|
|
|
Returns:
|
|
List[Dict[str, Any]]: 计算费用后的人材机节点列表
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def format_rcj_output(self, rcj_nodes: List[Dict[str, Any]], node_type: str) -> List[Dict[str, Any]]:
|
|
"""
|
|
格式化人材机节点输出
|
|
|
|
Args:
|
|
rcj_nodes: 人材机节点列表
|
|
node_type: 节点类型
|
|
|
|
Returns:
|
|
List[Dict[str, Any]]: 格式化后的人材机节点列表
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def calculate_rcj_fees(
|
|
self,
|
|
json_file_path: str,
|
|
project_name: str,
|
|
engineering_type: str,
|
|
project_guid: str = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
计算人材机合价
|
|
|
|
Args:
|
|
json_file_path: JSON文件路径
|
|
project_name: 项目名称
|
|
engineering_type: 工程类型
|
|
project_guid: 项目GUID，用于区分同名项目
|
|
|
|
Returns:
|
|
Dict[str, Any]: 计算结果
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def post_process_quantity_fees(self, output_file: str, project_name: str) -> None:
|
|
"""
|
|
对工程量取费表进行后处理
|
|
|
|
Args:
|
|
output_file: 输出文件路径
|
|
project_name: 项目名称
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def preprocess_project_nodes(self, project_nodes: List[Dict[str, Any]]) -> None:
|
|
"""
|
|
对工程量节点进行预处理
|
|
|
|
Args:
|
|
project_nodes: 工程量节点列表
|
|
"""
|
|
pass
|
|
|
|
|
|
class DefaultCalculationStrategy(CalculationStrategy):
|
|
"""
|
|
默认计算策略实现
|
|
|
|
实现了所有计算方法的默认行为，与原有代码逻辑相同。
|
|
其他计算策略可以继承此类，只重写需要修改的方法。
|
|
"""
|
|
|
|
def __init__(self):
|
|
# 导入必要的模块
|
|
from equipment_calculation.expressioncalculator import ExpressionCalculator
|
|
from equipment_calculation.bcl_utils import calculator
|
|
from equipment_calculation.item_acquisition import get_quantity_nodes, get_classified_resource_nodes
|
|
|
|
# 缓存已计算过的费用
|
|
self.calculated_fees = {}
|
|
|
|
def calculate_fee_base(
|
|
self,
|
|
fee_base: str,
|
|
cost_table: Dict[str, Any],
|
|
project_node: Dict[str, Any],
|
|
context: Any,
|
|
in_calculation: Optional[Set[str]] = None,
|
|
) -> float:
|
|
"""计算取费基数"""
|
|
from equipment_calculation.quantity_fee_calculator import calculate_fee_base as original_calculate_fee_base
|
|
|
|
# 传递自身作为计算策略
|
|
return original_calculate_fee_base(fee_base, cost_table, project_node, context, in_calculation, self)
|
|
|
|
def calculate_internal_fee(
|
|
self,
|
|
fee_code: str,
|
|
cost_table: Dict[str, Any],
|
|
project_node: Dict[str, Any],
|
|
context: Any,
|
|
in_calculation: Optional[Set[str]] = None,
|
|
) -> float:
|
|
"""计算表内费用"""
|
|
from equipment_calculation.quantity_fee_calculator import (
|
|
calculate_internal_fee as original_calculate_internal_fee,
|
|
)
|
|
|
|
# 传递自身作为计算策略
|
|
return original_calculate_internal_fee(fee_code, cost_table, project_node, context, in_calculation, self)
|
|
|
|
def calculate_external_variable(self, var_name: str, context: Any) -> float:
|
|
"""计算表外变量"""
|
|
# 直接使用 calculator.calculate，而不是调用 quantity_fee_calculator.py 中的函数
|
|
from equipment_calculation.bcl_utils import calculator
|
|
|
|
result = calculator.calculate(var_name, context)
|
|
return float(result) if result is not None else 0.0
|
|
|
|
def find_fee_item_by_code(self, cost_table: Dict[str, Any], code: str) -> Optional[Dict[str, Any]]:
|
|
"""在取费表中查找指定代码的费用项"""
|
|
from equipment_calculation.quantity_fee_calculator import (
|
|
find_fee_item_by_code as original_find_fee_item_by_code,
|
|
)
|
|
|
|
return original_find_fee_item_by_code(cost_table, code)
|
|
|
|
def find_fee_base_nodes(self, node: Dict[str, Any], result: Optional[List] = None) -> List[Dict[str, Any]]:
|
|
"""递归查找取费表中包含"取费基数"的节点或有子节点的费用项"""
|
|
from equipment_calculation.quantity_fee_calculator import find_fee_base_nodes as original_find_fee_base_nodes
|
|
|
|
return original_find_fee_base_nodes(node, result)
|
|
|
|
def calculate_all_fees(
|
|
self,
|
|
project_node: Dict[str, Any],
|
|
cost_table: Dict[str, Any],
|
|
json_file_path: Optional[str] = None,
|
|
engineering_type: Optional[str] = None,
|
|
) -> Dict[str, float]:
|
|
"""计算所有费用"""
|
|
from equipment_calculation.quantity_fee_calculator import calculate_all_fees as original_calculate_all_fees
|
|
|
|
# 传递自身作为计算策略
|
|
return original_calculate_all_fees(project_node, cost_table, json_file_path, engineering_type, self)
|
|
|
|
def calculate_rcj_count(
|
|
self,
|
|
rcj_nodes: List[Tuple[Dict[str, Any], str]],
|
|
project_children: List[Dict[str, Any]],
|
|
json_file_path: Optional[str] = None,
|
|
) -> List[Dict[str, Any]]:
|
|
"""计算人材机节点的数量"""
|
|
from equipment_calculation.resource_fee_calculator import calc_rcj_count as original_calc_rcj_count
|
|
|
|
return original_calc_rcj_count(rcj_nodes, project_children, json_file_path)
|
|
|
|
def cat_rcj_count(self, rcj_nodes: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|
"""汇总人材机节点的数量"""
|
|
from equipment_calculation.resource_fee_calculator import cat_rcj_count as original_cat_rcj_count
|
|
|
|
return original_cat_rcj_count(rcj_nodes)
|
|
|
|
def calc_rcj_fee(self, rcj_nodes: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|
"""计算人材机节点的费用"""
|
|
from equipment_calculation.resource_fee_calculator import calc_rcj_fee as original_calc_rcj_fee
|
|
|
|
return original_calc_rcj_fee(rcj_nodes)
|
|
|
|
def format_rcj_output(self, rcj_nodes: List[Dict[str, Any]], node_type: str) -> List[Dict[str, Any]]:
|
|
"""格式化人材机节点输出"""
|
|
from equipment_calculation.resource_fee_calculator import format_rcj_output as original_format_rcj_output
|
|
|
|
return original_format_rcj_output(rcj_nodes, node_type)
|
|
|
|
# 修复 calculate_rcj_fees 未定义的错误
|
|
def calculate_rcj_fees(self, json_file_path, project_name, project_guid=None):
|
|
"""
|
|
计算项目划分节点下所有人材机节点的合价
|
|
|
|
Args:
|
|
json_file_path: JSON文件路径
|
|
project_name: 项目名称
|
|
project_guid: 项目GUID，用于区分同名项目
|
|
|
|
Returns:
|
|
Dict[str, Any]: 计算结果
|
|
"""
|
|
# 导入原始函数
|
|
from equipment_calculation.resource_fee_calculator import calculate_rcj_fees as original_calculate_rcj_fees
|
|
|
|
# 调用原始函数
|
|
return original_calculate_rcj_fees(json_file_path, project_name, project_guid)
|
|
|
|
def post_process_quantity_fees(self, output_file: str, project_name: str) -> None:
|
|
"""对工程量取费表进行后处理"""
|
|
# 默认实现不做任何操作
|
|
pass
|
|
|
|
def preprocess_project_nodes(self, project_nodes: List[Dict[str, Any]]) -> None:
|
|
"""
|
|
默认实现不做任何预处理
|
|
"""
|
|
pass
|