Coverage for backend/idaes_factory/queryset_lookup.py: 83%

1from __future__ import annotations 

2from typing import TYPE_CHECKING 

3 

4from core.auxiliary.enums.unitOpGraphics import ConType 

5 

6if TYPE_CHECKING: 

7 from flowsheetInternals.unitops.models.SimulationObject import SimulationObject 

8 from flowsheetInternals.unitops.models.Port import Port 

9 from flowsheetInternals.propertyPackages.models.SimulationObjectPropertyPackages import SimulationObjectPropertyPackages 

10 from core.auxiliary.models.PropertySet import PropertySet 

11 from core.auxiliary.models.PropertyInfo import PropertyInfo 

12 from core.auxiliary.models.PropertyValue import PropertyValue 

13 from core.auxiliary.models.IndexedItem import IndexedItem 

14 

15 

16""" 

17This file contains various functions that are used to 

18look up related objects without having to make additional 

19database queries. It effectively loops through related objects 

20in python to get or filter desired objects, rather than 

21using the database to do this. 

22 

23To get the most out of this, all related objects should be 

24prefetched in an initial query. 

25 

26Note: for large datasets (ie. objects with many related 

27objects), this may become slow, and it may be more efficient 

28to use the database to filter objects. 

29""" 

30 

31 

32def get_simulation_object(simulation_objects, obj_id: int) -> SimulationObject: 

33 # get a simulation object by id, or raise an exception if not found 

34 simulation_object = next( 

35 (obj for obj in simulation_objects 

36 if obj.pk == obj_id 

37 ), None 

38 ) 

39 if simulation_object is None: 39 ↛ 40line 39 didn't jump to line 40 because the condition on line 39 was never true

40 raise ValueError(f"Simulation object with id {obj_id} not found") 

41 return simulation_object 

42 

43 

44def filter_simulation_objects(simulation_objects, include: set[str] = set()) -> list[SimulationObject]: 

45 return list(filter( 

46 lambda obj: obj.objectType in include, 

47 simulation_objects 

48 )) 

49 

50 

51def exclude_simulation_objects(simulation_objects, exclude: set[str]) -> list[SimulationObject]: 

52 return list(filter( 

53 lambda obj: obj.objectType not in exclude, 

54 simulation_objects 

55 )) 

56 

57 

58def get_port(simulation_object, key, index=0) -> Port: 

59 # get a port by key, or raise an exception if not found 

60 ports = simulation_object.ports.all() 

61 port = next( 

62 (port for port in ports 

63 if ( 

64 port.key == key 

65 and port.index == index 

66 ) 

67 ), None 

68 ) 

69 if port is None: 69 ↛ 70line 69 didn't jump to line 70 because the condition on line 69 was never true

70 raise ValueError(f"Port `{key}` index={index} not found in simulation object {simulation_object.componentName}") 

71 return port 

72 

73 

74def get_connected_port(simulation_object, direction) -> Port | None: 

75 connected_ports = simulation_object.connectedPorts.all() 

76 return next( 

77 (port for port in connected_ports 

78 if port.direction == direction 

79 ), None 

80 ) 

81 

82def get_active_ports_for_direction(simulation_object: SimulationObject, direction: ConType) -> list[Port]: 

83 """ 

84 Gets ports for the given unit operation with non-null streams for the specified direction/connection type. 

85 """ 

86 ports = simulation_object.ports.all() 

87 

88 return list(filter( 

89 lambda port: port.direction == direction and port.stream_id is not None, 

90 ports) 

91 ) 

92 

93 

94def get_all_ports(simulation_object, keys) -> list[Port]: 

95 return list(filter( 

96 lambda port: port.key in keys, 

97 simulation_object.ports.all() 

98 )) 

99 

100 

101def get_property(property_set, key) -> PropertyInfo: 

102 # get a property by key, or raise an exception if not found 

103 contained_properties = property_set.ContainedProperties.all() 

104 prop = next( 

105 (prop for prop in contained_properties 

106 if prop.key == key 

107 ), None 

108 ) 

109 if prop is None: 109 ↛ 110line 109 didn't jump to line 110 because the condition on line 109 was never true

110 raise ValueError(f"Property `{key}` not found in property set") 

111 return prop 

112 

113 

114def get_value_object(property_info, indexes: list | None = None) -> PropertyValue | None: 

115 # get the value object of the property at the specified indexes, 

116 # or the first value if no indexes are specified 

117 

118 

119 property_values = property_info.values.all() 

120 # TODO: support indexes 

121 property_value = next( 

122 (property_value for property_value in property_values 

123 if indexes is None 

124 ), None 

125 ) 

126 return property_value 

127 

128 

129def get_index(property_value: PropertyValue, index_set: str) -> IndexedItem: 

130 indexed_item = next( 

131 (indexed_item for indexed_item in property_value.indexedItems.all() 

132 if indexed_item.type == index_set 

133 ), None 

134 ) 

135 if indexed_item is None: 135 ↛ 136line 135 didn't jump to line 136 because the condition on line 135 was never true

136 raise ValueError(f"Indexed item not found for property value {property_value.id}") 

137 return indexed_item 

138 

139 

140def get_property_package(simulation_object, key) -> SimulationObjectPropertyPackages: 

141 # get a property package by key, or raise an exception if not found 

142 property_packages = simulation_object.propertyPackages.all() 

143 property_package = next( 

144 (prop for prop in property_packages 

145 if prop.name == key 

146 ), None 

147 ) 

148 if property_package is None: 148 ↛ 149line 148 didn't jump to line 149 because the condition on line 148 was never true

149 raise ValueError(f"Property package `{key}` not found in simulation object {simulation_object.componentName}") 

150 return property_package 

151