Coverage for backend/django/idaes_factory/idaes_factory_context.py: 98%
74 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-05-13 02:47 +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 )
115 def track_property_value(self, property_value_id: int):
116 self.serialised_property_values[property_value_id] = True
118 def _add_dependency(self, depends_on_id: int, property_value: PropertyValue):
119 if depends_on_id not in self.property_value_dependencies:
120 self.property_value_dependencies[depends_on_id] = set()
121 self.property_value_dependencies[depends_on_id].add(property_value)
123 def add_property_value_dependency(self, property_value: PropertyValue):
124 """Register a dependency between property values for tracking purposes."""
125 self.track_property_value(property_value.id)
126 # Manipulated by
127 manipulated_by = getattr(property_value, "controlManipulated", None)
128 if manipulated_by is not None:
129 depends_on_id = manipulated_by.setPoint.id
130 self._add_dependency(depends_on_id, property_value)
132 # Formula dependencies
133 self.add_expression_dependency(property_value)
136 def add_expression_dependency(self, property_value: PropertyValue):
137 expression = property_value.formula
138 if not expression:
139 return
140 converted_expression = get_expression_dependencies(expression)
141 for dependency in converted_expression:
142 self._add_dependency(dependency, property_value) # add self-dependency to ensure it's tracked
144 def is_dynamic(self) -> bool:
145 """Return whether the bound scenario has dynamics enabled."""
146 if self.scenario is not None:
147 return self.scenario.enable_dynamics
148 else:
149 return False
151 # Updates the solve index so that the alread-loaded context can be used for multiple solves
152 # (Useful for multi-steady state)
153 def update_solve_index(self, index: int):
154 """Rebind the context to a different solve index for multi-solves.
156 Args:
157 index: Solve index associated with the current flowsheet.
158 """
159 self.solve_index = DataRow.objects.get(index=index, flowsheet_id=self.group.flowsheet.id)
161 def get_simulation_object(self, obj_id: int) -> SimulationObject:
162 """Return a simulation object by id using the prefetched queryset.
164 Args:
165 obj_id: Primary key of the simulation object to retrieve.
167 Returns:
168 The matching `SimulationObject` instance from the context cache.
169 """
170 return queryset_lookup.get_simulation_object(self.simulation_objects, obj_id)
173 def filter_object_type(self, include: set[str] = set()) -> list[SimulationObject]:
174 """Filter cached simulation objects to the provided set of type keys.
176 Args:
177 include: Unit operation type identifiers that should be returned.
179 Returns:
180 List of `SimulationObject` instances matching the requested types.
181 """
182 return queryset_lookup.filter_simulation_objects(self.simulation_objects, include)
185 def exclude_object_type(self, exclude: set[str]) -> list[SimulationObject]:
186 """Filter cached simulation objects by excluding specific type keys.
188 Args:
189 exclude: Unit operation type identifiers that should be omitted.
191 Returns:
192 List of `SimulationObject` instances not belonging to the excluded types.
193 """
194 return queryset_lookup.exclude_simulation_objects(self.simulation_objects, exclude)
196 def get_property(self, property_set: PropertySet, key: str) -> PropertyInfo:
197 """Retrieve a property from a property set by its key.
199 Args:
200 property_set: Property set that contains the requested property.
201 key: Identifier for the property within the set.
203 Returns:
204 The `PropertyInfo` instance found in the given property set.
205 """
206 return queryset_lookup.get_property(property_set, key)
208 def get_property_value(self, property: PropertyInfo, indexes: list[Any] | None = None) -> PropertyValue:
209 """Retrieve a property value, optionally constrained by index selections.
211 Args:
212 property: Property whose value object should be fetched.
213 indexes: Optional list of indices to select a specific value entry.
215 Returns:
216 The `PropertyValue` matching the property and index configuration.
217 """
218 return queryset_lookup.get_value_object(property, indexes)