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

1from django.db import models 

2from core.auxiliary.enums import ConType 

3from typing import TYPE_CHECKING 

4 

5from core.managers import AccessControlManager 

6from core.auxiliary.models.PropertyValue import PropertyValue 

7from core.auxiliary.models.IndexedItem import IndexedItem 

8from flowsheetInternals.graphicData.models.graphicObjectModel import GraphicObject 

9from flowsheetInternals.unitops.config import get_object_schema 

10 

11if TYPE_CHECKING: 

12 from core.auxiliary.models.Flowsheet import Flowsheet 

13 from flowsheetInternals.unitops.models.SimulationObject import SimulationObject 

14 from flowsheetInternals.graphicData.models.graphicObjectModel import GraphicObject 

15 

16 

17class Port(models.Model): 

18 flowsheet = models.ForeignKey("core_auxiliary.Flowsheet", on_delete=models.CASCADE, related_name="Ports") 

19 displayName = models.CharField(max_length=64) 

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

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

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

23 unitOp = models.ForeignKey('SimulationObject', on_delete=models.CASCADE, related_name="ports", null=True) 

24 stream = models.ForeignKey('SimulationObject', on_delete=models.SET_NULL, null=True, related_name="connectedPorts") 

25 

26 created_at = models.DateTimeField(auto_now_add=True) 

27 objects = AccessControlManager() 

28 

29 # runtime-accessed relations 

30 flowsheet: "Flowsheet" 

31 unitOp: "SimulationObject | None" 

32 stream: "SimulationObject | None" 

33 

34 class Meta: 

35 ordering = ['index', 'created_at'] 

36 

37 def default_stream_position(self, unitop: "flowsheetInternals.unitops.SimulationObject", 

38 unitop_graphic_object: GraphicObject, port_index: int, num_ports: int) -> dict[ 

39 str, float]: 

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

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

42 

43 if self.direction == ConType.Inlet: 

44 xOffset = -stream_offset 

45 else: 

46 xOffset = 1 + stream_offset 

47 

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

49 yOffset = (port_index + 0.5) / num_ports 

50 

51 return { 

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

53 "y": unitop_graphic_object.y + yOffset * unitop_graphic_object.height 

54 } 

55 

56 def default_stream_name(self, unit_op: "flowsheetInternals.unitops.SimulationObject") -> str: 

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

58 

59 def reindex_port_on_delete(self): 

60 subsequent_ports = Port.objects.filter( 

61 unitOp=self.unitOp, 

62 key=self.key, 

63 index__gt=self.index 

64 ).order_by('index') 

65 

66 unit_op_schema = get_object_schema(self.unitOp) 

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

68 index_before_update = self.index 

69 for p in subsequent_ports: 

70 p.index -= 1 

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

72 p.save() 

73 

74 property_set = self.unitOp.properties 

75 property_infos = property_set.containedProperties.filter( 

76 index__gte=self.index 

77 ).order_by('index') 

78 

79 unit = self.unitOp 

80 

81 value = False 

82 if self.direction == ConType.Outlet: 

83 # You can only delete outlets on something that has a split fraction. 

84 # Therefore, delete the split fraction property associated with the outlet.  

85 # Get the properties of the outlet to be deleted. 

86 property_infos = unit.properties.ContainedProperties.all() 

87 for property in property_infos: 

88 # TODO: Surely we can use sumToOne or indexed by compounds instead of this? 

89 if property.key == "split_fraction": 89 ↛ 91line 89 didn't jump to line 91 because the condition on line 89 was always true

90 property_info = property_infos.filter(key="split_fraction").first() 

91 elif property.key == "priorities": 

92 property_info = property_infos.filter(key="priorities").first() 

93 elif property.key == "split_flow": 

94 property_info = property_infos.filter(key="split_flow").first() 

95 

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

97 indexed_item = IndexedItem.objects.filter(owner=unit, key="outlet_" + f"{index_before_update + 1}").first() 

98 # Delete the outlet and its property values 

99 PropertyValue.objects.filter(property=property_info, indexedItems=indexed_item).delete() 

100 if indexed_item is not None: 100 ↛ 104line 100 didn't jump to line 104 because the condition on line 100 was always true

101 indexed_item.delete() 

102 

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

104 indexed_items = IndexedItem.objects.filter(owner=unit, type="splitter_fraction") 

105 # For the remaining outlets, 

106 num_outlets = indexed_items.count() 

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

108 # Update the index 

109 indexed_item = indexed_items[i] 

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

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

112 indexed_item.displayName = unit.schema.splitter_fraction_name + f" {i + 1}" 

113 indexed_item.save() 

114 

115 self.delete() 

116 unit.refresh_from_db() 

117 unit.reevaluate_properties_enabled() 

118 else: 

119 self.delete()