Coverage for backend/django/flowsheetInternals/unitops/models/Port.py: 100%

1from django.db import models 

2from core.auxiliary.enums import ConType, AnchorSide 

3from typing import TYPE_CHECKING 

4 

5from core.managers import AccessControlManager, AllFlowsheetStatesManager 

6from core.auxiliary.models.PropertyValue import PropertyValue 

7from core.auxiliary.models.IndexedItem import IndexedItem 

8from core.auxiliary.models.FlowsheetHistoryModel import FlowsheetHistoryModel 

9from flowsheetInternals.graphicData.models.graphicObjectModel import GraphicObject 

10from flowsheetInternals.unitops.config import get_object_schema 

11 

12if TYPE_CHECKING: 

13 from core.auxiliary.models.Flowsheet import Flowsheet 

14 from flowsheetInternals.unitops.models.SimulationObject import SimulationObject 

15 from flowsheetInternals.graphicData.models.graphicObjectModel import GraphicObject 

16 

17 

18class Port(FlowsheetHistoryModel, models.Model): 

19 flowsheet_state = models.ForeignKey( 

20 "core_auxiliary.FlowsheetState", on_delete=models.CASCADE, related_name="Ports" 

21 ) 

22 displayName = models.CharField(max_length=64) 

23 direction = models.CharField(choices=ConType.choices) 

24 key = models.CharField(max_length=64) # key to identify this port in the schema 

25 index = models.IntegerField(default=0) # index of the port 

26 anchor_side = models.CharField( 

27 choices=AnchorSide.choices, max_length=16, null=True, blank=True 

28 ) 

29 anchor_index = models.IntegerField(null=True, blank=True) 

30 unitOp = models.ForeignKey( 

31 "SimulationObject", on_delete=models.CASCADE, related_name="ports", null=True 

32 ) 

33 stream = models.ForeignKey( 

34 "SimulationObject", 

35 on_delete=models.SET_NULL, 

36 null=True, 

37 related_name="connectedPorts", 

38 ) 

39 

40 created_at = models.DateTimeField(auto_now_add=True) 

41 objects = AccessControlManager() 

42 all_states = AllFlowsheetStatesManager() 

43 

44 # runtime-accessed relations 

45 flowsheet: "Flowsheet" 

46 unitOp: "SimulationObject | None" 

47 stream: "SimulationObject | None" 

48 

49 class Meta: 

50 ordering = ["index", "created_at"] 

51 

52 def default_stream_position( 

53 self, 

54 unitop: "flowsheetInternals.unitops.SimulationObject", 

55 unitop_graphic_object: GraphicObject, 

56 port_index: int, 

57 num_ports: int, 

58 ) -> dict[str, float]: 

59 """Calculate the default position for a stream connected to this port""" 

60 stream_offset = get_object_schema(unitop).ports[self.key].streamOffset 

61 rotation = unitop_graphic_object.rotation 

62 flipped = unitop_graphic_object.flipped 

63 

64 if self.direction == ConType.Inlet: 

65 xOffset = -stream_offset 

66 else: 

67 xOffset = 1 + stream_offset 

68 

69 # Calculate y offset based on port index and total number of ports 

70 yOffset = (port_index + 0.5) / num_ports 

71 

72 # Re-orient the offsets based on the rotation and flip of the unit operation. 

73 transforms = { 

74 (0, False): lambda x, y: (x, y), 

75 (0, True): lambda x, y: (1 - x, y), 

76 (90, False): lambda x, y: (1 - y, x), 

77 (90, True): lambda x, y: (1 - y, 1 - x), 

78 (180, False): lambda x, y: (1 - x, 1 - y), 

79 (180, True): lambda x, y: (1 + x, 1 - y), 

80 (270, False): lambda x, y: (y, 1 - x), 

81 (270, True): lambda x, y: (y, x), 

82 } 

83 xOffset, yOffset = transforms[(rotation, bool(flipped))](xOffset, yOffset) 

84 

85 return { 

86 "x": unitop_graphic_object.x + xOffset * unitop_graphic_object.width, 

87 "y": unitop_graphic_object.y + yOffset * unitop_graphic_object.height, 

88 } 

89 

90 def default_stream_name( 

91 self, unit_op: "flowsheetInternals.unitops.SimulationObject" 

92 ) -> str: 

93 return get_object_schema(unit_op).ports[self.key].streamName 

94 

95 def reindex_port_on_delete(self): 

96 subsequent_ports = Port.objects.filter( 

97 unitOp=self.unitOp, key=self.key, index__gt=self.index 

98 ).order_by("index") 

99 

100 unit_op_schema = get_object_schema(self.unitOp) 

101 display_name = unit_op_schema.ports[self.key].displayName 

102 index_before_update = self.index 

103 for p in subsequent_ports: 

104 p.index -= 1 

105 p.displayName = f"{display_name} {p.index + 1}" 

106 p.save() 

107 

108 property_set = self.unitOp.properties 

109 property_infos = property_set.containedProperties.filter( 

110 index__gte=self.index 

111 ).order_by("index") 

112 

113 unit = self.unitOp 

114 

115 value = False 

116 if self.direction == ConType.Outlet: 

117 split_property = unit.properties.ContainedProperties.filter( 

118 key__in=("split_fraction", "priorities", "split_flow") 

119 ).first() 

120 

121 # Get the index of the outlet to be deleted. 

122 indexed_item = IndexedItem.objects.filter( 

123 owner=unit, key="outlet_" + f"{index_before_update + 1}" 

124 ).first() 

125 # Delete the outlet and its property values 

126 if split_property is not None and indexed_item is not None: 

127 PropertyValue.objects.filter( 

128 property=split_property, 

129 indexedItems=indexed_item, 

130 ).delete() 

131 if indexed_item is not None: 

132 indexed_item.delete() 

133 

134 # Re-index the display names of the input fields for the remaining outlets. 

135 indexed_items = IndexedItem.objects.filter( 

136 owner=unit, type="splitter_fraction" 

137 ) 

138 # For the remaining outlets, 

139 num_outlets = indexed_items.count() 

140 for i in range(indexed_items.count()): 

141 # Update the index 

142 indexed_item = indexed_items[i] 

143 indexed_item.key = "outlet_" + f"{i + 1}" 

144 # Use the updated index to change the name displayed on the frontend 

145 indexed_item.displayName = ( 

146 unit.schema.splitter_fraction_name + f" {i + 1}" 

147 ) 

148 indexed_item.save() 

149 

150 self.delete() 

151 unit.update_height() 

152 unit.refresh_from_db() 

153 unit.reevaluate_properties_enabled() 

154 else: 

155 self.delete() 

156 unit.update_height()