Coverage for backend/idaes_service/solver/methods/adapter_methods.py: 89%

45 statements  

« prev     ^ index     » next       coverage.py v7.10.7, created at 2025-11-06 23:27 +0000

1from typing import Any, Callable 

2from abc import ABC, abstractmethod 

3from idaes.core.base.property_base import PhysicalParameterBlock 

4from idaes.core.base.reaction_base import ReactionParameterBlock 

5from ..flowsheet_manager_type import FlowsheetManager 

6from common.models.idaes import PropertyPackageId 

7 

8 

9 

10 

11class AdapterBase(ABC): 

12 """ 

13 Base class for adapter methods 

14 """ 

15 

16 @abstractmethod 

17 def run(self, value, flowsheet_manager: FlowsheetManager) -> Any: 

18 pass 

19 

20 

21class PropertyPackage(AdapterBase): 

22 """Get a property package by ID""" 

23 

24 def run(self, id: PropertyPackageId, flowsheet_manager: FlowsheetManager) -> PhysicalParameterBlock: 

25 return flowsheet_manager.property_packages.get(id) 

26 

27class ReactionPackage(AdapterBase): 

28 """Get a reaction package""" 

29 def run(self, id: str, flowsheet_manager: FlowsheetManager) -> ReactionParameterBlock: 

30 return flowsheet_manager.model.fs.reaction_params 

31 

32class ReactorPropertyPackage(AdapterBase): 

33 def run(self,value,flowsheet_manager) -> PhysicalParameterBlock: 

34 return flowsheet_manager.model.fs.peng_robinson 

35 

36 

37class Value(AdapterBase): 

38 """Do not modify the value passed""" 

39 

40 def run(self, value: Any,flowsheet_manager) -> Any: 

41 return value 

42 

43 

44class Constant(AdapterBase): 

45 """Always return the argument passed""" 

46 def __init__(self,constant) -> None: 

47 self._constant = constant 

48 

49 def run(self, value, flowsheet_manager) -> Callable: 

50 return self._constant 

51 

52 

53class Dictionary(AdapterBase): 

54 """Validate a dictionary against a schema and adapt the values""" 

55 def __init__(self,schema) -> None: 

56 self._schema = schema 

57 

58 def run(self, dictionary: dict, flowsheet_manager) -> Callable[[dict[str, Callable]], dict]: 

59 schema = self._schema 

60 

61 result = {} 

62 for item, adapter in schema.items(): 

63 # TODO: do we need to handle optional items better? 

64 result[item] = adapter.run(dictionary.get(item, None),flowsheet_manager) 

65 return result 

66 

67 

68class PowerPropertyPackage(AdapterBase): 

69 def run(self,value,flowsheet_manager) -> PhysicalParameterBlock: 

70 return flowsheet_manager.model.fs.power_pp # See __init__ method of Flowsheet_Manager.py  

71 

72class acPropertyPackage(AdapterBase): 

73 def run(self,value,flowsheet_manager) -> PhysicalParameterBlock: 

74 return flowsheet_manager.model.fs.ac_pp # See __init__ method of Flowsheet_Manager.py  

75 

76class TransformerPropertyPackage(AdapterBase): 

77 def run(self,value,flowsheet_manager) -> PhysicalParameterBlock: 

78 return flowsheet_manager.model.fs.tr_pp # See __init__ method of Flowsheet_Manager.py  

79 

80