Coverage for backend/django/idaes_factory/idaes_factory_context.py: 98%
74 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-03-26 20:57 +0000
1from typing import Any
3from django.db import transaction
4from django.db.models import QuerySet, Prefetch
6from ahuora_builder_types.flowsheet_schema import PropertyPackageType
7from idaes_factory.adapters.convert_expression import get_expression_dependencies
8from core.auxiliary.models import PropertyInfo, PropertySet
9from core.auxiliary.models.PropertyValue import PropertyValue
10from core.auxiliary.models.DataRow import DataRow
11from flowsheetInternals.graphicData.models.groupingModel import Grouping
12from flowsheetInternals.unitops.models.SimulationObject import SimulationObject
13from . import queryset_lookup
14from core.auxiliary.models.Scenario import Scenario
17ParameterName = str
18ParameterValue = float
19LiveSolveParams = dict[ParameterName,ParameterValue]
20DependencyId = int
21PropertyValueDependencies = dict[DependencyId, set[PropertyValue]]
22"""
23Parameters for solving. Used when solving live, to specify the values that each of the input columns in the multi-steady state simulation should be set to.
24Key is the key of the column, should be one of the input columns used.
25Value is a float value to set it to.
26"""
30class IdaesFactoryContext:
31 """
32 The IdaesFactoryContext class provides:
34 - Efficient retrieval of related objects from the
35 database, for both serialisation and reloading
36 of the flowsheet
37 - A container for context variables such as time
38 steps, property packages, etc.
40 It is used within idaes_factory to provide context
41 through the various adapter classes.
42 """
45 def __init__(
46 self,
47 group_id: int,
48 time_steps: list[int] = [0],
49 time_step_size: float = 1.0,
50 require_variables_fixed: bool = False,
51 solve_index: int | None = None, # If it's from multi-steady state, index to store values at
52 scenario : Scenario = None,
53 ) -> None:
54 """Initialise the context with eager-loaded flowsheet data.
56 Args:
57 group_id: Identifier of the flowsheet being prepared.
58 time_steps: Time indices that adapters should serialise.
59 require_variables_fixed: Whether adapters must enforce fixed variables.
60 solve_index: Optional multi-steady-state solve index to bind.
61 scenario: Scenario providing state and dynamics configuration.
62 """
63 self.group_id = group_id
64 self.group = Grouping.objects.get(id=group_id)
65 self.simulation_objects: QuerySet[SimulationObject] = None
67 if solve_index is not None: 67 ↛ 68line 67 didn't jump to line 68 because the condition on line 67 was never true
68 self.solve_index = DataRow.objects.get(index=solve_index, scenario_id=scenario.id)
69 else:
70 self.solve_index = None
72 # context vars
73 self.time_steps = time_steps
74 self.time_step_size = time_step_size
75 self.require_variables_fixed = require_variables_fixed
76 self.property_packages: list[PropertyPackageType] = []
77 self.expression_values = {}
78 self.scenario = scenario
79 self.property_value_dependencies: PropertyValueDependencies = {} # dict mapping property value id to the set of property value ids that depend on it (i.e. controlManipulated or formula dependencies)
80 self.serialised_property_values = {} # set of property value ids that have been serialised
81 self.load()
84 def load(self):
85 """Prefetch all simulation objects and related data for the flowsheet."""
86 # load all associated unit models, streams, property sets,
87 # properties, ports, etc. into the context
89 # db calls: simulation_objects, 1 for each prefetch_related\
90 sim_objs = self.group.get_recursive_simulation_objects()
91 self.simulation_objects = sim_objs.filter(
92 is_deleted=False
93 ).select_related(
94 "properties",
95 "recycleConnection",
96 ).prefetch_related(
97 Prefetch(
98 "properties__ContainedProperties",
99 queryset=PropertyInfo.objects
100 .select_related("recycleConnection")
101 .prefetch_related(
102 Prefetch("values", queryset=PropertyValue.objects
103 .select_related("controlManipulated", "controlSetPoint", "controlSetPoint__manipulated__property", "controlManipulated__setPoint", "dataColumn")
104 .prefetch_related(
105 "indexedItems",
106 "controlSetPoint__manipulated__property__values"
107 )
108 )
109 )
110 ),
111 "ports",
112 "connectedPorts",
113 "propertyPackages",
114 )
116 def track_property_value(self, property_value_id: int):
117 self.serialised_property_values[property_value_id] = True
119 def _add_dependency(self, depends_on_id: int, property_value: PropertyValue):
120 if depends_on_id not in self.property_value_dependencies:
121 self.property_value_dependencies[depends_on_id] = set()
122 self.property_value_dependencies[depends_on_id].add(property_value)
124 def add_property_value_dependency(self, property_value: PropertyValue):
125 """Register a dependency between property values for tracking purposes."""
126 self.track_property_value(property_value.id)
127 # Manipulated by
128 manipulated_by = getattr(property_value, "controlManipulated", None)
129 if manipulated_by is not None:
130 depends_on_id = manipulated_by.setPoint.id
131 self._add_dependency(depends_on_id, property_value)
133 # Formula dependencies
134 self.add_expression_dependency(property_value)
137 def add_expression_dependency(self, property_value: PropertyValue):
138 expression = property_value.formula
139 if not expression:
140 return
141 converted_expression = get_expression_dependencies(expression)
142 for dependency in converted_expression:
143 self._add_dependency(dependency, property_value) # add self-dependency to ensure it's tracked
145 def is_dynamic(self) -> bool:
146 """Return whether the bound scenario has dynamics enabled."""
147 if self.scenario is not None:
148 return self.scenario.enable_dynamics
149 else:
150 return False
152 # Updates the solve index so that the alread-loaded context can be used for multiple solves
153 # (Useful for multi-steady state)
154 def update_solve_index(self, index: int):
155 """Rebind the context to a different solve index for multi-solves.
157 Args:
158 index: Solve index associated with the current flowsheet.
159 """
160 self.solve_index = DataRow.objects.get(index=index, flowsheet_id=self.group.flowsheet.id)
162 def get_simulation_object(self, obj_id: int) -> SimulationObject:
163 """Return a simulation object by id using the prefetched queryset.
165 Args:
166 obj_id: Primary key of the simulation object to retrieve.
168 Returns:
169 The matching `SimulationObject` instance from the context cache.
170 """
171 return queryset_lookup.get_simulation_object(self.simulation_objects, obj_id)
174 def filter_object_type(self, include: set[str] = set()) -> list[SimulationObject]:
175 """Filter cached simulation objects to the provided set of type keys.
177 Args:
178 include: Unit operation type identifiers that should be returned.
180 Returns:
181 List of `SimulationObject` instances matching the requested types.
182 """
183 return queryset_lookup.filter_simulation_objects(self.simulation_objects, include)
186 def exclude_object_type(self, exclude: set[str]) -> list[SimulationObject]:
187 """Filter cached simulation objects by excluding specific type keys.
189 Args:
190 exclude: Unit operation type identifiers that should be omitted.
192 Returns:
193 List of `SimulationObject` instances not belonging to the excluded types.
194 """
195 return queryset_lookup.exclude_simulation_objects(self.simulation_objects, exclude)
197 def get_property(self, property_set: PropertySet, key: str) -> PropertyInfo:
198 """Retrieve a property from a property set by its key.
200 Args:
201 property_set: Property set that contains the requested property.
202 key: Identifier for the property within the set.
204 Returns:
205 The `PropertyInfo` instance found in the given property set.
206 """
207 return queryset_lookup.get_property(property_set, key)
209 def get_property_value(self, property: PropertyInfo, indexes: list[Any] | None = None) -> PropertyValue:
210 """Retrieve a property value, optionally constrained by index selections.
212 Args:
213 property: Property whose value object should be fetched.
214 indexes: Optional list of indices to select a specific value entry.
216 Returns:
217 The `PropertyValue` matching the property and index configuration.
218 """
219 return queryset_lookup.get_value_object(property, indexes)