Coverage for backend/django/core/auxiliary/services/flowsheet_states/cloning.py: 85%

1"""Non-mutating relational clone coordinator for flowsheet states.""" 

2 

3from __future__ import annotations 

4 

5import re 

6import uuid 

7from collections import defaultdict 

8from collections.abc import Sequence 

9from copy import deepcopy 

10from dataclasses import dataclass, field 

11 

12from django.apps import apps 

13from django.core.exceptions import FieldDoesNotExist 

14from django.db import connection, models, transaction 

15 

16from core.auxiliary.formula_limits import validate_formula_length 

17from core.auxiliary.methods.replace_expression_ids import ( 

18 extract_id_from_formula_key, 

19 get_formula_keys, 

20 replace_props, 

21) 

22from core.validation import bypass_write_access_checks 

23 

24from .errors import CloneDeficiencyType, FlowsheetStateCloneError 

25from .registry import ( 

26 ClonePolicy, 

27 CloneProfile, 

28 registry, 

29) 

30from .validation import formula_text_field_names, validate_state_for_clone 

31 

32 

33_FORMULA_UNIT_TOKEN = re.compile(r"\(unit(\d+)\)") 

34 

35 

36@dataclass 

37class ProjectCloneContext: 

38 """Cross-flowsheet mappings retained while a whole project is copied.""" 

39 

40 source_project: object 

41 target_project: object 

42 cost_curves: dict[int, object] = field(default_factory=dict) 

43 settings_profiles: dict[int, object] = field(default_factory=dict) 

44 stable_flowsheets: dict[int, object] = field(default_factory=dict) 

45 study_lineages: dict[tuple[int, uuid.UUID], tuple[int, uuid.UUID]] = field( 

46 default_factory=dict 

47 ) 

48 pending_logical_references: list[tuple] = field(default_factory=list) 

49 warnings: list[dict] = field(default_factory=list) 

50 profiles_copied: bool = False 

51 validated_source_state_ids: set[int] = field(default_factory=set) 

52 source_flowsheet_ids: frozenset[int] = field(init=False) 

53 

54 def __post_init__(self) -> None: 

55 """Capture the source project boundary before any target rows exist.""" 

56 

57 if self.source_project is None: 57 ↛ 58line 57 didn't jump to line 58 because the condition on line 57 was never true

58 self.source_flowsheet_ids = frozenset() 

59 return 

60 self.source_flowsheet_ids = frozenset( 

61 self.source_project.flowsheets.values_list("pk", flat=True) 

62 ) 

63 

64 def register_flowsheet(self, source, target) -> None: 

65 """Record the new stable identity before cloning its state.""" 

66 

67 self.stable_flowsheets[source.pk] = target 

68 

69 

70@dataclass 

71class CloneResult: 

72 """Old-to-new mappings and structured clone warnings.""" 

73 

74 source_state: object 

75 target_state: object 

76 profile: CloneProfile 

77 model_maps: dict[type[models.Model], dict[int, models.Model]] = field( 

78 default_factory=dict 

79 ) 

80 source_objects: dict[type[models.Model], dict[int, models.Model]] = field( 

81 default_factory=dict 

82 ) 

83 warnings: list[dict] = field(default_factory=list) 

84 deferred_relations: list[tuple[models.Model, models.Field, int]] = field( 

85 default_factory=list 

86 ) 

87 

88 def copied(self, model: type[models.Model], source_pk: int | None): 

89 """Return a copied row by its source primary key.""" 

90 

91 if source_pk is None: 91 ↛ 92line 91 didn't jump to line 92 because the condition on line 91 was never true

92 return None 

93 return self.model_maps.get(model, {}).get(source_pk) 

94 

95 @property 

96 def root_grouping(self): 

97 """Return the copied source root grouping when present.""" 

98 

99 grouping = apps.get_model("flowsheetInternals_graphicData", "Grouping") 

100 return self.copied(grouping, self.source_state.root_grouping_id) 

101 

102 

103def _manager(model): 

104 """Return the explicit manager that includes every state role/deleted row.""" 

105 

106 return getattr(model, "all_states", model._base_manager) 

107 

108 

109def _inspect_clone_table_presence( 

110 models_to_copy: Sequence[type[models.Model]], 

111 *, 

112 source_state_id: int, 

113 target_state_id: int, 

114 check_target: bool, 

115) -> tuple[set[type[models.Model]], type[models.Model] | None]: 

116 """Inspect source and target clone tables in one database round trip. 

117 

118 Table and column identifiers come exclusively from Django model metadata; 

119 state IDs and model labels remain parameterized query values. The returned 

120 target model preserves the existing non-empty-target validation without a 

121 separate ``exists()`` query for every registered model. 

122 """ 

123 

124 if not models_to_copy: 124 ↛ 125line 124 didn't jump to line 125 because the condition on line 124 was never true

125 return set(), None 

126 

127 clauses = [] 

128 parameters = [] 

129 quote_name = connection.ops.quote_name 

130 for model in models_to_copy: 

131 table = quote_name(model._meta.db_table) 

132 state_column = quote_name( 

133 model._meta.get_field("flowsheet_state").column 

134 ) 

135 target_exists = ( 

136 f"EXISTS (SELECT 1 FROM {table} WHERE {state_column} = %s)" 

137 if check_target 

138 else "FALSE" 

139 ) 

140 clauses.append( 

141 f"SELECT %s, " 

142 f"EXISTS (SELECT 1 FROM {table} WHERE {state_column} = %s), " 

143 f"{target_exists}" 

144 ) 

145 parameters.extend((model._meta.label, source_state_id)) 

146 if check_target: 

147 parameters.append(target_state_id) 

148 

149 with connection.cursor() as cursor: 

150 cursor.execute(" UNION ALL ".join(clauses), parameters) 

151 presence_rows = cursor.fetchall() 

152 

153 models_by_label = {model._meta.label: model for model in models_to_copy} 

154 populated_models = { 

155 models_by_label[label] 

156 for label, source_has_rows, _ in presence_rows 

157 if source_has_rows 

158 } 

159 occupied_target_model = next( 

160 ( 

161 models_by_label[label] 

162 for label, _, target_has_rows in presence_rows 

163 if target_has_rows 

164 ), 

165 None, 

166 ) 

167 return populated_models, occupied_target_model 

168 

169 

170def _is_state_owned(model) -> bool: 

171 try: 

172 model._meta.get_field("flowsheet_state") 

173 except FieldDoesNotExist: 

174 return False 

175 return True 

176 

177 

178def _copy_settings_profiles(context: ProjectCloneContext) -> None: 

179 """Reference same-project profiles or clone them once for project copy.""" 

180 

181 if context.profiles_copied: 

182 return 

183 profile_model = apps.get_model("Economics", "EconomicsSettingsProfile") 

184 source_profiles = profile_model._base_manager.filter( 

185 project=context.source_project 

186 ).order_by("pk") 

187 if context.source_project.pk == context.target_project.pk: 187 ↛ 188line 187 didn't jump to line 188 because the condition on line 187 was never true

188 context.settings_profiles.update( 

189 {profile.pk: profile for profile in source_profiles} 

190 ) 

191 context.profiles_copied = True 

192 return 

193 

194 target_has_default = profile_model._base_manager.filter( 

195 project=context.target_project, 

196 is_default=True, 

197 ).exists() 

198 for source in source_profiles: 

199 existing = profile_model._base_manager.filter( 

200 project=context.target_project, 

201 name=source.name, 

202 ).first() 

203 if existing is not None: 

204 context.settings_profiles[source.pk] = existing 

205 target_has_default = target_has_default or existing.is_default 

206 continue 

207 values = {"project": context.target_project} 

208 for concrete in source._meta.concrete_fields: 

209 if concrete.primary_key or concrete.name in { 

210 "project", 

211 "created_at", 

212 "updated_at", 

213 }: 

214 continue 

215 values[concrete.attname] = deepcopy(getattr(source, concrete.attname)) 

216 if values.get("is_default") and target_has_default: 216 ↛ 217line 216 didn't jump to line 217 because the condition on line 216 was never true

217 values["is_default"] = False 

218 copied = profile_model(**values) 

219 copied.save() 

220 context.settings_profiles[source.pk] = copied 

221 target_has_default = target_has_default or copied.is_default 

222 context.profiles_copied = True 

223 

224 

225def _project_cost_curve_id( 

226 context: ProjectCloneContext, 

227 source_curve_id: int, 

228) -> int: 

229 """Reuse same-project curves or clone one curve across project boundaries.""" 

230 

231 mapped = context.cost_curves.get(source_curve_id) 

232 if mapped is not None: 

233 return mapped.pk 

234 

235 curve_model = apps.get_model("Economics", "CostCurve") 

236 source = curve_model._base_manager.get(pk=source_curve_id) 

237 if source.project_id == context.target_project.pk: 237 ↛ 238line 237 didn't jump to line 238 because the condition on line 237 was never true

238 context.cost_curves[source_curve_id] = source 

239 return source.pk 

240 

241 existing = curve_model._base_manager.filter( 

242 project=context.target_project, 

243 curve_key=source.curve_key, 

244 ).first() 

245 if existing is not None: 245 ↛ 246line 245 didn't jump to line 246 because the condition on line 245 was never true

246 context.cost_curves[source_curve_id] = existing 

247 return existing.pk 

248 

249 values = {"project": context.target_project} 

250 for concrete in source._meta.concrete_fields: 

251 if concrete.primary_key or concrete.name in { 

252 "project", 

253 "created_at", 

254 "updated_at", 

255 }: 

256 continue 

257 values[concrete.attname] = deepcopy(getattr(source, concrete.attname)) 

258 copied = curve_model(**values) 

259 copied.save() 

260 context.cost_curves[source_curve_id] = copied 

261 return copied.pk 

262 

263 

264def _normalize_external_relation( 

265 *, 

266 source, 

267 field, 

268 target_state, 

269 profile, 

270 project_context, 

271): 

272 """Return the referenced external row ID selected for a copied field.""" 

273 

274 relation_id = getattr(source, field.attname) 

275 if relation_id is None: 275 ↛ 276line 275 didn't jump to line 276 because the condition on line 275 was never true

276 return None 

277 if ( 

278 source._meta.label in { 

279 "Economics.CapitalCostLine", 

280 "Economics.EquipmentMapping", 

281 } 

282 and field.name == "cost_curve" 

283 and profile == CloneProfile.NEW_FLOWSHEET 

284 ): 

285 if project_context is None: 

286 source_project = source.flowsheet_state.flowsheet.project 

287 target_project = target_state.flowsheet.project 

288 if source_project.pk != target_project.pk: 288 ↛ 289line 288 didn't jump to line 289 because the condition on line 288 was never true

289 raise FlowsheetStateCloneError( 

290 "Cross-project NEW_FLOWSHEET clones require a ProjectCloneContext." 

291 ) 

292 return relation_id 

293 return _project_cost_curve_id(project_context, relation_id) 

294 if ( 

295 source._meta.label == "Economics.EconomicsStudy" 

296 and field.name == "settings_profile" 

297 and profile == CloneProfile.NEW_FLOWSHEET 

298 ): 

299 if project_context is None: 299 ↛ 300line 299 didn't jump to line 300 because the condition on line 299 was never true

300 source_project = source.flowsheet_state.flowsheet.project 

301 target_project = target_state.flowsheet.project 

302 if source_project.pk != target_project.pk: 

303 raise FlowsheetStateCloneError( 

304 "Cross-project NEW_FLOWSHEET clones require a ProjectCloneContext." 

305 ) 

306 return relation_id 

307 _copy_settings_profiles(project_context) 

308 copied_profile = project_context.settings_profiles.get(relation_id) 

309 return copied_profile.pk if copied_profile is not None else None 

310 return relation_id 

311 

312 

313def _build_copy( 

314 *, 

315 source, 

316 target_state, 

317 profile, 

318 result, 

319 deferred_relations, 

320 project_context, 

321): 

322 """Build an unsaved clone without mutating the source instance.""" 

323 

324 contract = registry.contract_for(type(source)) 

325 values = {} 

326 for concrete in source._meta.concrete_fields: 

327 if concrete.primary_key: 

328 continue 

329 if concrete.name == "flowsheet_state": 

330 values[concrete.attname] = target_state.pk 

331 continue 

332 if not concrete.is_relation or concrete.remote_field is None: 

333 values[concrete.attname] = deepcopy(getattr(source, concrete.attname)) 

334 continue 

335 

336 related_model = concrete.remote_field.model 

337 source_related_id = getattr(source, concrete.attname) 

338 if source_related_id is None: 

339 values[concrete.attname] = None 

340 continue 

341 if not _is_state_owned(related_model): 

342 values[concrete.attname] = _normalize_external_relation( 

343 source=source, 

344 field=concrete, 

345 target_state=target_state, 

346 profile=profile, 

347 project_context=project_context, 

348 ) 

349 continue 

350 

351 related_policy = registry.policy_for(related_model, profile) 

352 if related_policy != ClonePolicy.COPY: 

353 if concrete.null and concrete.name in contract.clear_relations: 353 ↛ 356line 353 didn't jump to line 356 because the condition on line 353 was always true

354 values[concrete.attname] = None 

355 continue 

356 raise FlowsheetStateCloneError( 

357 f"{source._meta.label}.{concrete.name} references " 

358 f"non-copied {related_model._meta.label}.", 

359 deficiency_type=CloneDeficiencyType.FLOWSHEET_RELATIONSHIPS, 

360 ) 

361 copied_related = result.copied(related_model, source_related_id) 

362 if copied_related is not None: 362 ↛ 364line 362 didn't jump to line 364 because the condition on line 362 was always true

363 values[concrete.attname] = copied_related.pk 

364 elif concrete.null: 

365 values[concrete.attname] = None 

366 deferred_relations.append( 

367 (concrete, source_related_id) 

368 ) 

369 else: 

370 raise FlowsheetStateCloneError( 

371 f"Clone dependency order did not create " 

372 f"{related_model._meta.label} {source_related_id} before " 

373 f"{source._meta.label}.", 

374 deficiency_type=CloneDeficiencyType.FLOWSHEET_RELATIONSHIPS, 

375 ) 

376 

377 if ( 

378 source._meta.label == "Economics.EconomicsStudy" 

379 and profile == CloneProfile.NEW_FLOWSHEET 

380 ): 

381 values["lineage_id"] = uuid.uuid4() 

382 if source._meta.label == "core_auxiliary.MLModel": 

383 # The active step is UI workflow state, not durable worker state. Re-derive a 

384 # task-free usable step from retained artifacts because tasks and upload 

385 # sessions are deliberately outside the cloned aggregate. 

386 if source.surrogate_model: 386 ↛ 387line 386 didn't jump to line 387 because the condition on line 386 was never true

387 values["active_step"] = 3 

388 elif source.csv_bucket and source.csv_object_key: 388 ↛ 391line 388 didn't jump to line 391 because the condition on line 388 was always true

389 values["active_step"] = 2 if source.active_step >= 2 else 1 

390 else: 

391 values["active_step"] = 0 

392 return type(source)(**values) 

393 

394 

395def _copy_model_rows( 

396 *, 

397 model, 

398 source_state, 

399 target_state, 

400 profile, 

401 result, 

402 project_context, 

403): 

404 """Bulk-copy one dependency-ordered model and record its mapping.""" 

405 

406 sources = list( 

407 _manager(model) 

408 .filter(flowsheet_state_id=source_state.pk) 

409 .order_by(model._meta.pk.name) 

410 ) 

411 copies = [] 

412 deferred_by_copy = [] 

413 for source in sources: 

414 deferred = [] 

415 copied = _build_copy( 

416 source=source, 

417 target_state=target_state, 

418 profile=profile, 

419 result=result, 

420 deferred_relations=deferred, 

421 project_context=project_context, 

422 ) 

423 copies.append(copied) 

424 deferred_by_copy.append(deferred) 

425 if copies: 425 ↛ 427line 425 didn't jump to line 427 because the condition on line 425 was always true

426 _manager(model).bulk_create(copies, batch_size=2000) 

427 result.model_maps[model] = { 

428 source.pk: copied for source, copied in zip(sources, copies, strict=True) 

429 } 

430 result.source_objects[model] = {source.pk: source for source in sources} 

431 

432 for copied, deferred in zip(copies, deferred_by_copy, strict=True): 

433 for relation, source_related_id in deferred: 433 ↛ 434line 433 didn't jump to line 434 because the loop on line 433 never started

434 result.deferred_relations.append( 

435 (copied, relation, source_related_id) 

436 ) 

437 

438 

439def _flush_deferred_relations(result: CloneResult) -> None: 

440 """Persist nullable foreign keys that pointed forward in dependency order.""" 

441 

442 updates = defaultdict(lambda: {"objects": {}, "fields": set()}) 

443 for copied, relation, source_related_id in result.deferred_relations: 443 ↛ 444line 443 didn't jump to line 444 because the loop on line 443 never started

444 copied_related = result.copied( 

445 relation.remote_field.model, 

446 source_related_id, 

447 ) 

448 if copied_related is None: 

449 raise FlowsheetStateCloneError( 

450 f"Could not remap {type(copied)._meta.label}.{relation.name} " 

451 f"source id {source_related_id}.", 

452 deficiency_type=CloneDeficiencyType.FLOWSHEET_RELATIONSHIPS, 

453 ) 

454 setattr(copied, relation.attname, copied_related.pk) 

455 update = updates[type(copied)] 

456 update["objects"][copied.pk] = copied 

457 update["fields"].add(relation.name) 

458 for model, update in updates.items(): 458 ↛ 459line 458 didn't jump to line 459 because the loop on line 458 never started

459 _manager(model).bulk_update( 

460 list(update["objects"].values()), 

461 sorted(update["fields"]), 

462 batch_size=2000, 

463 ) 

464 result.deferred_relations.clear() 

465 

466 

467def _copy_many_to_many(result: CloneResult) -> None: 

468 """Clone implicit through rows after every copied endpoint has a new PK.""" 

469 

470 for model, mapping in result.model_maps.items(): 

471 for many_to_many in model._meta.many_to_many: 

472 through = many_to_many.remote_field.through 

473 if _is_state_owned(through): 

474 # Explicit through models such as MonitoringTableProperty are 

475 # normal registered rows and have already been copied. 

476 continue 

477 related_model = many_to_many.remote_field.model 

478 if registry.policy_for(related_model, result.profile) != ClonePolicy.COPY: 478 ↛ 479line 478 didn't jump to line 479 because the condition on line 478 was never true

479 raise FlowsheetStateCloneError( 

480 f"{model._meta.label}.{many_to_many.name} points to a " 

481 "non-copied model." 

482 ) 

483 source_field = through._meta.get_field(many_to_many.m2m_field_name()) 

484 related_field = through._meta.get_field( 

485 many_to_many.m2m_reverse_field_name() 

486 ) 

487 through_rows = [] 

488 source_relation_rows = through._base_manager.filter( 

489 **{f"{source_field.attname}__in": mapping} 

490 ).values_list(source_field.attname, related_field.attname) 

491 for source_pk, related_id in source_relation_rows: 

492 copied_owner = mapping[source_pk] 

493 copied_related = result.copied(related_model, related_id) 

494 if copied_related is None: 494 ↛ 495line 494 didn't jump to line 495 because the condition on line 494 was never true

495 raise FlowsheetStateCloneError( 

496 f"Could not remap {model._meta.label}." 

497 f"{many_to_many.name} source id {related_id}.", 

498 deficiency_type=CloneDeficiencyType.FLOWSHEET_RELATIONSHIPS, 

499 ) 

500 through_rows.append( 

501 through( 

502 **{ 

503 source_field.attname: copied_owner.pk, 

504 related_field.attname: copied_related.pk, 

505 } 

506 ) 

507 ) 

508 if through_rows: 

509 through._base_manager.bulk_create(through_rows, batch_size=2000) 

510 

511 

512def _remap_formula_text( 

513 text: str, 

514 property_values: dict[int, models.Model], 

515 simulation_objects: dict[int, models.Model], 

516) -> str: 

517 """Remap embedded formula IDs or identify a repairable source formula.""" 

518 

519 keys = get_formula_keys(text) 

520 replacement_keys = [] 

521 for key in keys: 

522 source_pk = extract_id_from_formula_key(key) 

523 copied = property_values.get(source_pk) 

524 if copied is None: 524 ↛ 525line 524 didn't jump to line 525 because the condition on line 524 was never true

525 raise FlowsheetStateCloneError( 

526 f"Formula references PropertyValue {source_pk} outside the clone boundary.", 

527 deficiency_type=CloneDeficiencyType.FORMULA_REFERENCES, 

528 ) 

529 replacement_keys.append(f"prop{copied.pk}") 

530 remapped = replace_props(text, replacement_keys) if keys else text 

531 

532 def replace_unit(match): 

533 source_pk = int(match.group(1)) 

534 copied = simulation_objects.get(source_pk) 

535 if copied is None: 535 ↛ 536line 535 didn't jump to line 536 because the condition on line 535 was never true

536 raise FlowsheetStateCloneError( 

537 f"Formula references SimulationObject {source_pk} outside the clone boundary.", 

538 deficiency_type=CloneDeficiencyType.FORMULA_REFERENCES, 

539 ) 

540 return f"(unit{copied.pk})" 

541 

542 return _FORMULA_UNIT_TOKEN.sub(replace_unit, remapped) 

543 

544 

545def _remap_embedded_property_references(result: CloneResult) -> None: 

546 """Remap formulas and Economics driver-input JSON after core rows exist.""" 

547 

548 property_value_model = apps.get_model("core_auxiliary", "PropertyValue") 

549 property_info_model = apps.get_model("core_auxiliary", "PropertyInfo") 

550 simulation_object_model = apps.get_model( 

551 "flowsheetInternals_unitops", "SimulationObject" 

552 ) 

553 property_values = result.model_maps.get(property_value_model, {}) 

554 property_infos = result.model_maps.get(property_info_model, {}) 

555 simulation_objects = result.model_maps.get(simulation_object_model, {}) 

556 

557 for model, mapping in result.model_maps.items(): 

558 formula_fields = formula_text_field_names(model) 

559 update_fields = set(formula_fields) 

560 changed = [] 

561 for source_pk, copied in mapping.items(): 

562 source = result.source_objects[model][source_pk] 

563 modified = False 

564 for field_name in formula_fields: 

565 text = getattr(source, field_name) 

566 if not text: 

567 continue 

568 remapped = _remap_formula_text( 

569 text, 

570 property_values, 

571 simulation_objects, 

572 ) 

573 if model is property_value_model and field_name == "formula": 

574 remapped = validate_formula_length(remapped) 

575 setattr(copied, field_name, remapped) 

576 modified = True 

577 if model._meta.label == "Economics.CapitalCostLine": 

578 driver_inputs = deepcopy(source.driver_inputs) 

579 for raw_input in driver_inputs.values(): 

580 if ( 580 ↛ 584line 580 didn't jump to line 584 because the condition on line 580 was never true

581 not isinstance(raw_input, dict) 

582 or raw_input.get("source") != "property" 

583 ): 

584 continue 

585 copied_property = property_infos.get(raw_input.get("property_info")) 

586 if copied_property is None: 586 ↛ 587line 586 didn't jump to line 587 because the condition on line 586 was never true

587 raw_input["source"] = "" 

588 raw_input["property_info"] = None 

589 else: 

590 raw_input["property_info"] = copied_property.pk 

591 copied.driver_inputs = driver_inputs 

592 update_fields.add("driver_inputs") 

593 modified = True 

594 if model._meta.label == "Economics.EconomicsLineFormula": 

595 line_kind = ( 

596 "capital_line" 

597 if copied.capital_line_id is not None 

598 else "operating_line" 

599 ) 

600 copied_line_id = ( 

601 copied.capital_line_id or copied.operating_line_id 

602 ) 

603 copied.line_key = f"{line_kind}:{copied_line_id}" 

604 copied.formula_key = copied.line_key 

605 update_fields.update({"line_key", "formula_key"}) 

606 modified = True 

607 if modified: 

608 changed.append(copied) 

609 if changed and update_fields: 

610 _manager(model).bulk_update( 

611 changed, 

612 sorted(update_fields), 

613 batch_size=2000, 

614 ) 

615 

616 

617def _logical_reference( 

618 *, 

619 stable_id, 

620 lineage_id, 

621 result, 

622 project_context, 

623 copied_object, 

624 stable_field, 

625 lineage_field, 

626 owner_lineage, 

627): 

628 """Resolve or defer one Economics stable-id/lineage pair.""" 

629 

630 if stable_id is None or lineage_id is None: 

631 return stable_id, lineage_id, False 

632 source_flowsheet = result.source_state.flowsheet 

633 if stable_id == source_flowsheet.pk: 

634 mapped = project_context.study_lineages.get((stable_id, lineage_id)) 

635 if mapped is None: 

636 raise FlowsheetStateCloneError( 

637 "Same-flowsheet Economics lineage is missing from the clone map.", 

638 deficiency_type=CloneDeficiencyType.FLOWSHEET_RELATIONSHIPS, 

639 ) 

640 return mapped[0], mapped[1], False 

641 if project_context is not None and stable_id in project_context.stable_flowsheets: 641 ↛ 642line 641 didn't jump to line 642 because the condition on line 641 was never true

642 mapped = project_context.study_lineages.get((stable_id, lineage_id)) 

643 if mapped is not None: 

644 return mapped[0], mapped[1], False 

645 if ( 

646 project_context is not None 

647 and result.profile == CloneProfile.NEW_FLOWSHEET 

648 and project_context.source_project.pk != project_context.target_project.pk 

649 and stable_id in project_context.source_flowsheet_ids 

650 ): 

651 project_context.pending_logical_references.append( 

652 ( 

653 copied_object, 

654 stable_field, 

655 lineage_field, 

656 stable_id, 

657 lineage_id, 

658 owner_lineage, 

659 ) 

660 ) 

661 return stable_id, lineage_id, True 

662 return stable_id, lineage_id, False 

663 

664 

665def _reference_owner_lineage(*, result, model, source, copied): 

666 """Return the restored study lineage that owns a logical reference.""" 

667 

668 study_model = apps.get_model("Economics", "EconomicsStudy") 

669 config_model = apps.get_model( 

670 "Economics", 

671 "EconomicsStudyComparisonConfig", 

672 ) 

673 if model is study_model: 

674 return copied.lineage_id 

675 if model is config_model: 

676 copied_study = result.copied(study_model, source.active_study_id) 

677 return copied_study.lineage_id if copied_study is not None else None 

678 source_config = result.source_objects[config_model].get(source.config_id) 

679 if source_config is None: 679 ↛ 680line 679 didn't jump to line 680 because the condition on line 679 was never true

680 return None 

681 copied_study = result.copied(study_model, source_config.active_study_id) 

682 return copied_study.lineage_id if copied_study is not None else None 

683 

684 

685def _logical_reference_resolves(stable_id, lineage_id) -> bool: 

686 """Check a logical study reference against the target's current state.""" 

687 

688 study_model = apps.get_model("Economics", "EconomicsStudy") 

689 return _manager(study_model).filter( 

690 flowsheet_state__flowsheet_id=stable_id, 

691 flowsheet_state__flowsheet__current_state_id=models.F( 

692 "flowsheet_state_id" 

693 ), 

694 lineage_id=lineage_id, 

695 ).exists() 

696 

697 

698def _append_reference_warning(result, *, reference_kind, owner_lineage): 

699 """Record a product-safe missing Economics reference warning.""" 

700 

701 result.warnings.append( 

702 { 

703 "code": "economics_reference_unavailable", 

704 "owning_study_lineage": owner_lineage, 

705 "reference_kind": reference_kind, 

706 } 

707 ) 

708 

709 

710def _remap_economics_logical_references( 

711 result: CloneResult, 

712 project_context: ProjectCloneContext | None, 

713) -> None: 

714 """Preserve or regenerate study lineages and remap logical references.""" 

715 

716 study_model = apps.get_model("Economics", "EconomicsStudy") 

717 studies = result.model_maps.get(study_model, {}) 

718 if not studies: 

719 return 

720 if project_context is None: 

721 project_context = ProjectCloneContext( 

722 result.source_state.flowsheet.project, 

723 result.target_state.flowsheet.project, 

724 ) 

725 project_context.register_flowsheet( 

726 result.source_state.flowsheet, 

727 result.target_state.flowsheet, 

728 ) 

729 for source_pk, copied in studies.items(): 

730 source = result.source_objects[study_model][source_pk] 

731 project_context.study_lineages[ 

732 (result.source_state.flowsheet_id, source.lineage_id) 

733 ] = (result.target_state.flowsheet_id, copied.lineage_id) 

734 

735 update_groups = defaultdict(lambda: {"objects": [], "fields": set()}) 

736 logical_models = ( 

737 (study_model, "baseline_flowsheet_id", "baseline_study_lineage_id"), 

738 ( 

739 apps.get_model("Economics", "EconomicsStudyComparisonConfig"), 

740 "baseline_flowsheet_id", 

741 "baseline_study_lineage_id", 

742 ), 

743 ( 

744 apps.get_model("Economics", "EconomicsStudyComparisonSelection"), 

745 "target_flowsheet_id", 

746 "target_study_lineage_id", 

747 ), 

748 ) 

749 for model, stable_field, lineage_field in logical_models: 

750 for source_pk, copied in result.model_maps.get(model, {}).items(): 

751 source = result.source_objects[model][source_pk] 

752 owner_lineage = _reference_owner_lineage( 

753 result=result, 

754 model=model, 

755 source=source, 

756 copied=copied, 

757 ) 

758 stable_id, lineage_id, is_pending_project_reference = _logical_reference( 

759 stable_id=getattr(source, stable_field), 

760 lineage_id=getattr(source, lineage_field), 

761 result=result, 

762 project_context=project_context, 

763 copied_object=copied, 

764 stable_field=stable_field, 

765 lineage_field=lineage_field, 

766 owner_lineage=owner_lineage, 

767 ) 

768 should_validate_external = ( 

769 stable_id is not None 

770 and lineage_id is not None 

771 and stable_id != result.target_state.flowsheet_id 

772 and not is_pending_project_reference 

773 and result.profile in { 

774 CloneProfile.RESTORE, 

775 CloneProfile.NEW_FLOWSHEET, 

776 } 

777 ) 

778 if should_validate_external and not _logical_reference_resolves( 

779 stable_id, 

780 lineage_id, 

781 ): 

782 _append_reference_warning( 

783 result, 

784 reference_kind=model._meta.model_name, 

785 owner_lineage=owner_lineage, 

786 ) 

787 if ( 

788 model._meta.label 

789 == "Economics.EconomicsStudyComparisonSelection" 

790 ): 

791 _manager(model).filter(pk=copied.pk).delete() 

792 continue 

793 stable_id = None 

794 lineage_id = None 

795 setattr(copied, stable_field, stable_id)