Coverage for backend/django/flowsheetInternals/unitops/models/Port.py: 97%
75 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-03-26 20:57 +0000
1from django.db import models
2from core.auxiliary.enums import ConType
3from typing import TYPE_CHECKING
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
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
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")
26 created_at = models.DateTimeField(auto_now_add=True)
27 objects = AccessControlManager()
29 # runtime-accessed relations
30 flowsheet: "Flowsheet"
31 unitOp: "SimulationObject | None"
32 stream: "SimulationObject | None"
34 class Meta:
35 ordering = ['index', 'created_at']
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 rotation = unitop_graphic_object.rotation
43 flipped = unitop_graphic_object.flipped
45 if self.direction == ConType.Inlet:
46 xOffset = -stream_offset
47 else:
48 xOffset = 1 + stream_offset
50 # Calculate y offset based on port index and total number of ports
51 yOffset = (port_index + 0.5) / num_ports
53 # Re-orient the offsets based on the rotation and flip of the unit operation.
54 transforms = {
55 (0, False): lambda x, y: (x, y),
56 (0, True): lambda x, y: (1 - x, y),
57 (90, False): lambda x, y: (1 - y, x),
58 (90, True): lambda x, y: (1 - y, 1 - x),
59 (180, False): lambda x, y: (1 - x, 1 - y),
60 (180, True): lambda x, y: (1 + x, 1 - y),
61 (270, False): lambda x, y: (y, 1 - x),
62 (270, True): lambda x, y: (y, x),
63 }
64 xOffset, yOffset = transforms[(rotation, bool(flipped))](xOffset, yOffset)
66 return {
67 "x": unitop_graphic_object.x + xOffset * unitop_graphic_object.width,
68 "y": unitop_graphic_object.y + yOffset * unitop_graphic_object.height
69 }
71 def default_stream_name(self, unit_op: "flowsheetInternals.unitops.SimulationObject") -> str:
72 return get_object_schema(unit_op).ports[self.key].streamName
74 def reindex_port_on_delete(self):
75 subsequent_ports = Port.objects.filter(
76 unitOp=self.unitOp,
77 key=self.key,
78 index__gt=self.index
79 ).order_by('index')
81 unit_op_schema = get_object_schema(self.unitOp)
82 display_name = unit_op_schema.ports[self.key].displayName
83 index_before_update = self.index
84 for p in subsequent_ports:
85 p.index -= 1
86 p.displayName = f"{display_name} {p.index + 1}"
87 p.save()
89 property_set = self.unitOp.properties
90 property_infos = property_set.containedProperties.filter(
91 index__gte=self.index
92 ).order_by('index')
94 unit = self.unitOp
96 value = False
97 if self.direction == ConType.Outlet:
98 # You can only delete outlets on something that has a split fraction.
99 # Therefore, delete the split fraction property associated with the outlet.
100 # Get the properties of the outlet to be deleted.
101 property_infos = unit.properties.ContainedProperties.all()
102 for property in property_infos:
103 # TODO: Surely we can use sumToOne or indexed by compounds instead of this?
104 if property.key == "split_fraction":
105 property_info = property_infos.filter(key="split_fraction").first()
106 elif property.key == "priorities": 106 ↛ 107line 106 didn't jump to line 107 because the condition on line 106 was never true
107 property_info = property_infos.filter(key="priorities").first()
108 elif property.key == "split_flow":
109 property_info = property_infos.filter(key="split_flow").first()
111 # Get the index of the outlet to be deleted.
112 indexed_item = IndexedItem.objects.filter(owner=unit, key="outlet_" + f"{index_before_update + 1}").first()
113 # Delete the outlet and its property values
114 PropertyValue.objects.filter(property=property_info, indexedItems=indexed_item).delete()
115 if indexed_item is not None: 115 ↛ 119line 115 didn't jump to line 119 because the condition on line 115 was always true
116 indexed_item.delete()
118 # Re-index the display names of the input fields for the remaining outlets.
119 indexed_items = IndexedItem.objects.filter(owner=unit, type="splitter_fraction")
120 # For the remaining outlets,
121 num_outlets = indexed_items.count()
122 for i in range(indexed_items.count()):
123 # Update the index
124 indexed_item = indexed_items[i]
125 indexed_item.key = "outlet_" + f"{i + 1}"
126 # Use the updated index to change the name displayed on the frontend
127 indexed_item.displayName = unit.schema.splitter_fraction_name + f" {i + 1}"
128 indexed_item.save()
130 self.delete()
131 unit.update_height()
132 unit.refresh_from_db()
133 unit.reevaluate_properties_enabled()
134 else:
135 self.delete()
136 unit.update_height()