Coverage for backend/django/idaes_factory/queryset_lookup.py: 84%

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 core.auxiliary.models.PropertySet import PropertySet 

10 from core.auxiliary.models.PropertyInfo import PropertyInfo 

11 from core.auxiliary.models.PropertyValue import PropertyValue 

12 from core.auxiliary.models.IndexedItem import IndexedItem 

13 

14 

15""" 

16This file contains various functions that are used to 

17look up related objects without having to make additional 

18database queries. It effectively loops through related objects 

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

20using the database to do this. 

21 

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

23prefetched in an initial query. 

24 

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

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

27to use the database to filter objects. 

28""" 

29 

30 

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

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

33 simulation_object = next( 

34 (obj for obj in simulation_objects 

35 if obj.pk == obj_id 

36 ), None 

37 ) 

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

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

40 return simulation_object 

41 

42 

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

44 return list(filter( 

45 lambda obj: obj.objectType in include, 

46 simulation_objects 

47 )) 

48 

49 

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

51 return list(filter( 

52 lambda obj: obj.objectType not in exclude, 

53 simulation_objects 

54 )) 

55 

56 

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

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

59 ports = simulation_object.ports.all() 

60 port = next( 

61 (port for port in ports 

62 if ( 

63 port.key == key 

64 and port.index == index 

65 ) 

66 ), None 

67 ) 

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

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

70 return port 

71 

72 

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

74 connected_ports = simulation_object.connectedPorts.all() 

75 return next( 

76 (port for port in connected_ports 

77 if port.direction == direction 

78 ), None 

79 ) 

80 

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

82 """ 

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

84 """ 

85 ports = simulation_object.ports.all() 

86 

87 return list(filter( 

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

89 ports) 

90 ) 

91 

92 

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

94 return list(filter( 

95 lambda port: port.key in keys, 

96 simulation_object.ports.all() 

97 )) 

98 

99 

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

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

102 contained_properties = property_set.ContainedProperties.all() 

103 prop = next( 

104 (prop for prop in contained_properties 

105 if prop.key == key 

106 ), None 

107 ) 

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

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

110 return prop 

111 

112 

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

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

115 # or the first value if no indexes are specified 

116 

117 

118 property_values = property_info.values.all() 

119 # TODO: support indexes 

120 property_value = next( 

121 (property_value for property_value in property_values 

122 if indexes is None 

123 ), None 

124 ) 

125 return property_value 

126 

127 

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

129 indexed_item = next( 

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

131 if indexed_item.type == index_set 

132 ), None 

133 ) 

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

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

136 return indexed_item 

137