Coverage for backend/django/flowsheetInternals/unitops/viewsets/SimulationObjectViewSet.py: 85%

1import json 

2 

3from core.auxiliary.models.PropertyValue import PropertyValue 

4from core.auxiliary.models.Flowsheet import Flowsheet 

5from core.auxiliary.enums.unitOpData import SimulationObjectClass 

6from core.viewset import ModelViewSet 

7from rest_framework.response import Response 

8from flowsheetInternals.unitops.viewsets.DuplicateSimulationObject import ( 

9 DuplicateSimulationObject, 

10) 

11from flowsheetInternals.unitops.models.compound_propogation import ( 

12 update_compounds_on_set, 

13) 

14from flowsheetInternals.unitops.models.flow_tracking import track_downstream_stream_flow 

15from flowsheetInternals.unitops.logic.insert_translator_block import ( 

16 insert_translator_block, 

17) 

18from flowsheetInternals.unitops.serializers.SimulationObjectSerializer import ( 

19 SimulationObjectRetrieveSerializer, 

20 SimulationObjectSerializer, 

21) 

22from flowsheetInternals.unitops.models import SimulationObject 

23from flowsheetInternals.unitops.models.delete_factory import DeleteFactory 

24from core.auxiliary.models.PropertySet import PropertySet 

25from core.auxiliary.models.PropertyInfo import PropertyInfo 

26from flowsheetInternals.unitops.models.summary_table_factory import ( 

27 get_composition_summary_table_data, 

28 get_stream_summary_table_data, 

29 get_unitops_summary_table_data, 

30) 

31from drf_spectacular.utils import extend_schema, OpenApiParameter, OpenApiTypes 

32from rest_framework import serializers, status 

33from rest_framework.decorators import action 

34import traceback 

35from django.db.models import Prefetch 

36from flowsheetInternals.formula_templates.add_template import add_predefined_template 

37from flowsheetInternals.unitops.methods.add_expression import add_expression 

38from flowsheetInternals.graphicData.models.groupingModel import Grouping 

39from flowsheetInternals.unitops.config.config_base import configuration 

40from flowsheetInternals.unitops.config.variant_families import ( 

41 load_unit_op_variant_families, 

42) 

43from flowsheetInternals.unitops.models.simulation_object_factory import ( 

44 SimulationObjectFactory, 

45) 

46from flowsheetInternals.unitops.models.FlowsheetEditOperation import ( 

47 FlowsheetEditOperation, 

48) 

49from flowsheetInternals.unitops.services.edit_operations.mutation import flowsheet_edit 

50from flowsheetInternals.unitops.services.edit_operations.recorder import ( 

51 tracked_bulk_update, 

52) 

53 

54unit_op_variant_families = load_unit_op_variant_families() 

55 

56 

57class UpdateCompoundSerializer(serializers.Serializer): 

58 """Payload schema for updating compound selections on a stream.""" 

59 

60 compounds = serializers.ListField(child=serializers.CharField()) 

61 simulationObject = serializers.IntegerField() 

62 

63 

64class AddPropertyTemplateSerializer(serializers.Serializer): 

65 """Payload schema for adding a property template by key to a simulation object.""" 

66 

67 templateKey = serializers.CharField() 

68 

69 

70class DuplicateSimulationObjectSerializer(serializers.Serializer): 

71 """Payload schema for duplicating one or more simulation objects.""" 

72 

73 objectIDs = serializers.ListField(child=serializers.IntegerField(), min_length=1) 

74 x = serializers.FloatField() 

75 y = serializers.FloatField() 

76 

77 

78class AddPortSerializer(serializers.Serializer): 

79 """Payload schema for attaching a new port to a simulation object.""" 

80 

81 simulationObjectId = serializers.IntegerField() 

82 key = serializers.CharField() 

83 stream = serializers.IntegerField( 

84 required=False, allow_null=True 

85 ) # optional if existing stream isnt passed in 

86 

87 

88class MergeDecisionNodesSerializer(serializers.Serializer): 

89 """Payload schema for merging two decision nodes.""" 

90 

91 decisionNodeActive = serializers.IntegerField(required=True) 

92 decisionNodeOver = serializers.IntegerField(required=True) 

93 

94 

95class TrackedStreamFlowSerializer(serializers.Serializer): 

96 """Response schema for stream-flow highlighting.""" 

97 

98 sourceStreamId = serializers.IntegerField() 

99 streamIds = serializers.ListField(child=serializers.IntegerField()) 

100 

101 

102class RestoreObjectsSerializer(serializers.Serializer): 

103 """Payload schema for restoring previously deleted simulation objects.""" 

104 

105 object_ids = serializers.ListField( 

106 child=serializers.IntegerField(), 

107 help_text="Array of object IDs to restore", 

108 min_length=1, 

109 ) 

110 

111 def validate(self, data): 

112 """Ensure we have at least one valid ID""" 

113 if not data.get("object_ids"): 113 ↛ 114line 113 didn't jump to line 114 because the condition on line 113 was never true

114 raise serializers.ValidationError("At least one object ID must be provided") 

115 return data 

116 

117 

118class SwitchVariantSerializer(serializers.Serializer): 

119 """Payload schema for replacing a unit operation with a configured variant.""" 

120 

121 target_object_type = serializers.CharField() 

122 

123 

124class SwitchVariantResponseSerializer(serializers.Serializer): 

125 """Response schema for a successful unit-op variant switch.""" 

126 

127 simulation_object = serializers.IntegerField() 

128 

129 

130class SimulationObjectViewSet(ModelViewSet): 

131 """API endpoints for interacting with flowsheet simulation objects.""" 

132 

133 serializer_class = SimulationObjectSerializer 

134 

135 def get_queryset(self): 

136 """Return simulation objects with related property metadata eagerly loaded.""" 

137 # Prefetch related entities to avoid N+1 queries when serializing. 

138 queryset = ( 

139 SimulationObject.objects.all() 

140 .select_related("properties", "grouping") 

141 .prefetch_related( 

142 Prefetch( 

143 "properties__ContainedProperties", 

144 queryset=PropertyInfo.objects.select_related( 

145 "recycleConnection" 

146 ).prefetch_related( 

147 Prefetch( 

148 "values", 

149 queryset=PropertyValue.objects.select_related( 

150 "controlManipulated", "controlSetPoint" 

151 ).prefetch_related("indexedItems"), 

152 ) 

153 ), 

154 ), 

155 "connectedPorts__unitOp", 

156 ) 

157 ) 

158 

159 return queryset 

160 

161 def get_serializer_class(self): 

162 """Use the retrieve serializer when properties must be serialized.""" 

163 if self.action in ["list", "retrieve"]: 

164 # For the list and retrieve actions, include the properties of the 

165 # simulation object so canvas consumers can render from one payload. 

166 return SimulationObjectRetrieveSerializer 

167 return SimulationObjectSerializer 

168 

169 @extend_schema( 

170 parameters=[ 

171 OpenApiParameter(name="flowsheet", required=True, type=OpenApiTypes.INT), 

172 ] 

173 ) 

174 def list(self, request): 

175 """List simulation objects filtered by the provided flowsheet ID.""" 

176 return super().list(request) 

177 

178 def create(self, request, *args, **kwargs): 

179 """Create the complete factory batch as one automatic history action.""" 

180 serializer = self.get_serializer(data=request.data) 

181 serializer.is_valid(raise_exception=True) 

182 flowsheet = Flowsheet.objects.get(pk=request.query_params.get("flowsheet")) 

183 with flowsheet_edit( 

184 flowsheet=flowsheet, 

185 user=request.user, 

186 kind=FlowsheetEditOperation.Kind.Add, 

187 label_key="Add unit operation", 

188 ) as mutation: 

189 self.perform_create(serializer) 

190 response = Response( 

191 serializer.data, 

192 status=status.HTTP_201_CREATED, 

193 headers=self.get_success_headers(serializer.data), 

194 ) 

195 return mutation.add_to_response(response) 

196 

197 def update(self, request, *args, **kwargs): 

198 """Update editable unit metadata through the shared history path.""" 

199 return self._update_with_operation(request, *args, partial=False, **kwargs) 

200 

201 def partial_update(self, request, *args, **kwargs): 

202 """Partially update editable unit metadata through the history path.""" 

203 return self._update_with_operation(request, *args, partial=True, **kwargs) 

204 

205 def _update_with_operation(self, request, *args, partial, **kwargs): 

206 """Record scalar and property-package changes as one durable operation.""" 

207 instance = self.get_object() 

208 changes_property_package = bool( 

209 {"propertyPackageType", "customPackage"}.intersection(request.data) 

210 ) 

211 with flowsheet_edit( 

212 flowsheet=instance.flowsheet_state.flowsheet, 

213 user=request.user, 

214 kind=FlowsheetEditOperation.Kind.FieldPatch, 

215 label_key=( 

216 "Change property package" 

217 if changes_property_package 

218 else "Edit unit operation" 

219 ), 

220 ) as mutation: 

221 response = super().update( 

222 request, 

223 *args, 

224 partial=partial, 

225 **kwargs, 

226 ) 

227 return mutation.add_to_response(response) 

228 

229 def destroy(self, request, *args, **kwargs): 

230 """Delete the simulation object via the shared DeleteFactory.""" 

231 instance = self.get_object() 

232 with flowsheet_edit( 

233 flowsheet=instance.flowsheet_state.flowsheet, 

234 user=request.user, 

235 kind=FlowsheetEditOperation.Kind.Delete, 

236 label_key="Delete objects", 

237 ) as mutation: 

238 DeleteFactory.delete_object(instance) 

239 

240 return mutation.add_to_response(Response(status=status.HTTP_204_NO_CONTENT)) 

241 

242 # maybe do local storage and its request it in here 

243 

244 @extend_schema(responses=TrackedStreamFlowSerializer) 

245 @action(detail=True, methods=["get"], url_path="tracked-stream-flow") 

246 def tracked_stream_flow(self, request, pk=None): 

247 """Return the downstream streams connected to this stream.""" 

248 simulation_object = self.get_object() 

249 stream_types = { 

250 SimulationObjectClass.Stream, 

251 SimulationObjectClass.HumidAirStream, 

252 } 

253 

254 if simulation_object.objectType not in stream_types: 

255 return Response( 

256 {"detail": "Track stream is only available for streams."}, 

257 status=status.HTTP_400_BAD_REQUEST, 

258 ) 

259 

260 _unit_ops, streams = track_downstream_stream_flow(simulation_object) 

261 stream_ids = sorted(stream.id for stream in streams) 

262 

263 return Response( 

264 { 

265 "sourceStreamId": simulation_object.id, 

266 "streamIds": stream_ids, 

267 }, 

268 status=status.HTTP_200_OK, 

269 ) 

270 

271 @extend_schema( 

272 request=SwitchVariantSerializer, 

273 responses=SwitchVariantResponseSerializer, 

274 ) 

275 @action(detail=True, methods=["post"], url_path="switch-variant") 

276 def switch_variant(self, request, pk=None): 

277 """Replace this simulation object with another variant from the same family.""" 

278 current_object = self.get_object() 

279 serializer = SwitchVariantSerializer(data=request.data) 

280 serializer.is_valid(raise_exception=True) 

281 target_object_type = serializer.validated_data["target_object_type"] 

282 

283 if target_object_type == current_object.objectType: 

284 return Response( 

285 {"simulation_object": current_object.id}, 

286 status=status.HTTP_200_OK, 

287 ) 

288 

289 try: 

290 target_config = configuration[target_object_type] 

291 source_family_key, source_family = self._get_variant_family( 

292 current_object.objectType 

293 ) 

294 target_family_key, target_family = self._get_variant_family( 

295 target_object_type 

296 ) 

297 

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

299 raise ValueError( 

300 f"{current_object.objectType} does not support variant switching." 

301 ) 

302 if target_family is None: 302 ↛ 303line 302 didn't jump to line 303 because the condition on line 302 was never true

303 raise ValueError( 

304 f"{target_object_type} does not support variant switching." 

305 ) 

306 if source_family_key != target_family_key: 

307 raise ValueError( 

308 f"{target_object_type} is not a valid variant for " 

309 f"{current_object.objectType}." 

310 ) 

311 

312 with flowsheet_edit( 

313 flowsheet=current_object.flowsheet_state.flowsheet, 

314 user=request.user, 

315 kind=FlowsheetEditOperation.Kind.SwitchVariant, 

316 label_key="Switch unit-operation model", 

317 ) as mutation: 

318 new_object = self._create_variant_replacement( 

319 current_object, 

320 target_object_type, 

321 target_config, 

322 ) 

323 

324 if target_family.preserve_ports[target_object_type]: 324 ↛ 327line 324 didn't jump to line 327 because the condition on line 324 was always true

325 self._copy_variant_port_streams(current_object, new_object) 

326 

327 if target_family.preserve_graphic[target_object_type]: 327 ↛ 330line 327 didn't jump to line 330 because the condition on line 327 was always true

328 self._copy_variant_graphic_state(current_object, new_object) 

329 

330 current_object.permanently_delete() 

331 

332 except KeyError: 

333 return Response( 

334 {"detail": f"{target_object_type} is not a known object type."}, 

335 status=status.HTTP_400_BAD_REQUEST, 

336 ) 

337 except ValueError as e: 

338 return Response( 

339 {"detail": str(e)}, 

340 status=status.HTTP_400_BAD_REQUEST, 

341 ) 

342 

343 return Response( 

344 mutation.add_to_data({"simulation_object": new_object.id}), 

345 status=status.HTTP_200_OK, 

346 ) 

347 

348 @staticmethod 

349 def _get_variant_family(object_type): 

350 for family_key, family in unit_op_variant_families.items(): 350 ↛ 353line 350 didn't jump to line 353 because the loop on line 350 didn't complete

351 if object_type in family.variants: 

352 return family_key, family 

353 return None, None 

354 

355 @staticmethod 

356 def _port_identity(port): 

357 return (port.direction, port.key, port.index) 

358 

359 def _create_variant_replacement( 

360 self, simulation_object, object_type, object_schema 

361 ): 

362 parent_group = simulation_object.get_group() 

363 factory = SimulationObjectFactory() 

364 new_object = factory.create( 

365 object_type=object_type, 

366 object_schema=object_schema, 

367 coordinates={"x": 0, "y": 0}, 

368 flowsheet_state=simulation_object.flowsheet_state, 

369 componentName=simulation_object.componentName, 

370 parentGroup=parent_group, 

371 ) 

372 factory.perform_bulk_create() 

373 return new_object 

374 

375 def _copy_variant_port_streams(self, old_object, new_object): 

376 new_ports = {self._port_identity(port): port for port in new_object.ports.all()} 

377 ports_to_update = [] 

378 

379 for old_port in old_object.ports.select_related("stream").all(): 

380 if old_port.stream_id is None: 

381 continue 

382 

383 matching_port = new_ports.get(self._port_identity(old_port)) 

384 if matching_port is None: 

385 raise ValueError( 

386 "Cannot switch model type because connected port " 

387 f"{old_port.key} ({old_port.direction}, index {old_port.index}) " 

388 "does not exist on the target model." 

389 ) 

390 

391 matching_port.stream = old_port.stream 

392 ports_to_update.append(matching_port) 

393 

394 if ports_to_update: 394 ↛ exitline 394 didn't return from function '_copy_variant_port_streams' because the condition on line 394 was always true

395 tracked_bulk_update( 

396 new_object.ports.model.objects, 

397 ports_to_update, 

398 ["stream"], 

399 ) 

400 

401 @staticmethod 

402 def _copy_variant_graphic_state(old_object, new_object): 

403 old_graphic = old_object.graphicObject.first() 

404 new_graphic = new_object.graphicObject.first() 

405 

406 if old_graphic is None or new_graphic is None: 406 ↛ 407line 406 didn't jump to line 407 because the condition on line 406 was never true

407 return 

408 

409 new_graphic.visible = old_graphic.visible 

410 new_graphic.copy_position_from(old_graphic) 

411 

412 def get_summary_queryset(self): 

413 queryset = ( 

414 SimulationObject.objects.all() 

415 .select_related( 

416 "properties", 

417 ) 

418 .prefetch_related( 

419 Prefetch( 

420 "properties__ContainedProperties", 

421 queryset=PropertyInfo.objects.prefetch_related( 

422 Prefetch( 

423 "values", 

424 queryset=PropertyValue.objects.prefetch_related( 

425 "indexedItems" 

426 ), 

427 ) 

428 ), 

429 ) 

430 ) 

431 ) 

432 

433 return queryset 

434 

435 @extend_schema( 

436 parameters=[ 

437 OpenApiParameter(name="unit_map", type=OpenApiTypes.STR, required=True), 

438 OpenApiParameter(name="groups", type=OpenApiTypes.ANY, required=True), 

439 ], 

440 responses=serializers.DictField(), 

441 ) 

442 @action(detail=False, methods=["get"], url_path="streams-summary") 

443 def summary_table_streams(self, request): 

444 unit_map = self.request.query_params.get("unit_map") 

445 groups = self.request.query_params.get("groups") 

446 

447 results = {} 

448 groups = Grouping.objects.filter(id__in=json.loads(groups)) 

449 for current_group in groups: 

450 query_set = self.get_summary_queryset().filter( 

451 graphicObject__in=current_group.graphicObjects.all() 

452 ) 

453 

454 data = get_stream_summary_table_data(query_set, json.loads(unit_map)) 

455 

456 group_name = current_group.simulationObject.componentName 

457 results[group_name] = data 

458 

459 return Response(results, status=200) 

460 

461 @extend_schema( 

462 parameters=[ 

463 OpenApiParameter(name="unit_map", type=OpenApiTypes.STR, required=True), 

464 OpenApiParameter(name="groups", type=OpenApiTypes.ANY, required=True), 

465 ], 

466 responses=serializers.DictField(), 

467 ) 

468 @action(detail=False, methods=["get"], url_path="unitops-summary") 

469 def summary_table_unitops(self, request): 

470 unit_map = self.request.query_params.get("unit_map") 

471 groups = self.request.query_params.get("groups") 

472 

473 results = {} 

474 groups = Grouping.objects.filter(id__in=json.loads(groups)) 

475 for current_group in groups: 

476 query_set = self.get_summary_queryset().filter( 

477 graphicObject__in=current_group.graphicObjects.all() 

478 ) 

479 data = get_unitops_summary_table_data(query_set, json.loads(unit_map)) 

480 

481 group_name = current_group.simulationObject.componentName 

482 results[group_name] = data 

483 

484 return Response(results, status=200) 

485 

486 @extend_schema( 

487 parameters=[ 

488 OpenApiParameter( 

489 name="compound_mode", type=OpenApiTypes.STR, required=True 

490 ), 

491 OpenApiParameter(name="measure_type", type=OpenApiTypes.STR, required=True), 

492 OpenApiParameter(name="groups", type=OpenApiTypes.ANY, required=True), 

493 ], 

494 responses=serializers.DictField(), 

495 ) 

496 @action(detail=False, methods=["get"], url_path="compounds-summary") 

497 def summary_table_compounds(self, request): 

498 compound_mode = self.request.query_params.get("compound_mode") 

499 measure_type = self.request.query_params.get("measure_type") 

500 groups = self.request.query_params.get("groups") 

501 

502 results = {} 

503 groups = Grouping.objects.filter(id__in=json.loads(groups)) 

504 for current_group in groups: 

505 query_set = self.get_summary_queryset().filter( 

506 graphicObject__in=current_group.graphicObjects.all() 

507 ) 

508 data = get_composition_summary_table_data( 

509 queryset=query_set, 

510 target_compound_mode=compound_mode, 

511 measure_type=measure_type, 

512 ) 

513 

514 group_name = current_group.simulationObject.componentName 

515 results[group_name] = data 

516 

517 return Response(results, status=200) 

518 

519 @extend_schema(request=UpdateCompoundSerializer, responses=None) 

520 @action(detail=False, methods=["post"], url_path="update-compounds") 

521 def update_compounds(self, request): 

522 """Update the selected compounds for the specified simulation object.""" 

523 try: 

524 serializer = UpdateCompoundSerializer(data=request.data) 

525 serializer.is_valid(raise_exception=True) 

526 validated_data = serializer.validated_data 

527 

528 simulation_object = SimulationObject.objects.get( 

529 pk=validated_data.get("simulationObject") 

530 ) 

531 

532 expected_compounds = validated_data.get("compounds") 

533 with flowsheet_edit( 

534 flowsheet=simulation_object.flowsheet_state.flowsheet, 

535 user=request.user, 

536 kind=FlowsheetEditOperation.Kind.FieldPatch, 

537 label_key="Update compounds", 

538 ) as mutation: 

539 update_compounds_on_set(simulation_object, expected_compounds) 

540 

541 return Response( 

542 mutation.add_to_data({"status": "success"}), 

543 status=200, 

544 ) 

545 

546 except (SimulationObject.DoesNotExist, PropertySet.DoesNotExist) as e: 

547 return Response({"status": "error", "message": str(e)}, status=404) 

548 except Exception as e: 

549 print(traceback.format_exc()) 

550 return Response( 

551 { 

552 "status": "error", 

553 "message": str(e), 

554 "traceback": traceback.format_exc(), 

555 }, 

556 status=400, 

557 ) 

558 

559 @extend_schema(request=AddPortSerializer, responses=None) 

560 @action(detail=False, methods=["post"], url_path="add-port") 

561 def add_port(self, request): 

562 """Add a port to a simulation object, optionally reusing an existing stream.""" 

563 try: 

564 serializer = AddPortSerializer(data=request.data) 

565 serializer.is_valid(raise_exception=True) 

566 validated_data = serializer.validated_data 

567 simulation_object = SimulationObject.objects.get( 

568 pk=validated_data.get("simulationObjectId") 

569 ) 

570 key = validated_data.get("key") 

571 stream_id = validated_data.get("stream") 

572 existing_stream = None 

573 if stream_id is not None: 573 ↛ 574line 573 didn't jump to line 574 because the condition on line 573 was never true

574 existing_stream = SimulationObject.objects.get(pk=stream_id) 

575 if ( 

576 existing_stream.flowsheet_state_id 

577 != simulation_object.flowsheet_state_id 

578 ): 

579 return Response( 

580 { 

581 "status": "error", 

582 "message": "Stream must belong to the same flowsheet.", 

583 }, 

584 status=400, 

585 ) 

586 

587 with flowsheet_edit( 

588 flowsheet=simulation_object.flowsheet_state.flowsheet, 

589 user=request.user, 

590 kind=FlowsheetEditOperation.Kind.Connection, 

591 label_key="Add port", 

592 ) as mutation: 

593 if existing_stream is not None: 593 ↛ 594line 593 didn't jump to line 594 because the condition on line 593 was never true

594 simulation_object.add_port(key, existing_stream) 

595 else: 

596 simulation_object.add_port(key) 

597 

598 return Response( 

599 mutation.add_to_data({"status": "success"}), 

600 status=200, 

601 ) 

602 

603 except (SimulationObject.DoesNotExist, PropertySet.DoesNotExist) as e: 

604 return Response({"status": "error", "message": str(e)}, status=404) 

605 except Exception as e: 

606 return Response( 

607 { 

608 "status": "error", 

609 "message": str(e), 

610 "traceback": traceback.format_exc(), 

611 }, 

612 status=400, 

613 ) 

614 

615 @extend_schema(request=MergeDecisionNodesSerializer, responses=None) 

616 @action(detail=False, methods=["post"], url_path="merge-decision-nodes") 

617 def merge_decision_nodes(self, request): 

618 try: 

619 serializer = MergeDecisionNodesSerializer(data=request.data) 

620 serializer.is_valid(raise_exception=True) 

621 validated_data = serializer.validated_data 

622 

623 decision_node_active = SimulationObject.objects.get( 

624 pk=validated_data.get("decisionNodeActive") 

625 ) 

626 

627 decision_node_over = SimulationObject.objects.get( 

628 pk=validated_data.get("decisionNodeOver") 

629 ) 

630 

631 with flowsheet_edit( 

632 flowsheet=decision_node_active.flowsheet_state.flowsheet, 

633 user=request.user, 

634 kind=FlowsheetEditOperation.Kind.MergeStreams, 

635 label_key="Merge decision nodes", 

636 ) as mutation: 

637 decision_node_active.merge_decision_nodes( 

638 decision_node_active, 

639 decision_node_over, 

640 ) 

641 

642 return Response( 

643 mutation.add_to_data({"status": "success"}), 

644 status=200, 

645 ) 

646 

647 except (SimulationObject.DoesNotExist, PropertySet.DoesNotExist) as e: 

648 return Response({"status": "error", "message": str(e)}, status=404) 

649 except Exception as e: 

650 return Response( 

651 { 

652 "status": "error", 

653 "message": str(e), 

654 "traceback": traceback.format_exc(), 

655 }, 

656 status=400, 

657 ) 

658 

659 @extend_schema(request=None, responses=int) 

660 @action(detail=True, methods=["POST"]) 

661 def add_expression(self, request, pk=None): 

662 """Create a blank expression property on the simulation object.""" 

663 if pk is None: 663 ↛ 664line 663 didn't jump to line 664 because the condition on line 663 was never true

664 return Response( 

665 {"status": "error", "message": "id is required"}, status=400 

666 ) 

667 

668 simulation_object = SimulationObject.objects.select_related( 

669 "flowsheet_state__flowsheet", "properties" 

670 ).get(pk=pk) 

671 with flowsheet_edit( 

672 flowsheet=simulation_object.flowsheet_state.flowsheet, 

673 user=request.user, 

674 kind=FlowsheetEditOperation.Kind.FieldPatch, 

675 label_key="Add custom property", 

676 ) as mutation: 

677 prop_info = add_expression(simulation_object) 

678 

679 response = Response(prop_info.id, status=200) 

680 return mutation.add_to_response(response, header_only=True) 

681 

682 @extend_schema(request=AddPropertyTemplateSerializer) 

683 @action(detail=True, methods=["POST"]) 

684 def add_custom_property_template(self, request, pk=None): 

685 """Attach a custom propertytemplate to the simulation object using the supplied key.""" 

686 if pk is None: 686 ↛ 687line 686 didn't jump to line 687 because the condition on line 686 was never true

687 return Response( 

688 {"status": "error", "message": "id is required"}, status=400 

689 ) 

690 

691 simulation_object = SimulationObject.objects.prefetch_related("properties").get( 

692 pk=pk 

693 ) 

694 serializer = AddPropertyTemplateSerializer(data=request.data) 

695 serializer.is_valid(raise_exception=True) 

696 template_key = serializer.validated_data.get("templateKey") 

697 

698 try: 

699 with flowsheet_edit( 

700 flowsheet=simulation_object.flowsheet_state.flowsheet, 

701 user=request.user, 

702 kind=FlowsheetEditOperation.Kind.FieldPatch, 

703 label_key="Add property template", 

704 ) as mutation: 

705 add_predefined_template(simulation_object, template_key) 

706 return Response( 

707 mutation.add_to_data({"status": "success"}), 

708 status=200, 

709 ) 

710 except ValueError as e: 

711 return Response( 

712 { 

713 "status": "error", 

714 "message": str(e), 

715 "traceback": traceback.format_exc(), 

716 }, 

717 status=400, 

718 ) 

719 

720 @extend_schema(request=RestoreObjectsSerializer, responses=None) 

721 @action(detail=False, methods=["post"]) 

722 def restore(self, request): 

723 """Restore previously deleted simulation objects.""" 

724 try: 

725 # Handle both 'ids' and 'object_ids' in request data 

726 if "ids" in request.data: 726 ↛ 727line 726 didn't jump to line 727 because the condition on line 726 was never true

727 request_data = {"object_ids": request.data["ids"]} 

728 else: 

729 request_data = request.data 

730 

731 serializer = RestoreObjectsSerializer(data=request_data) 

732 serializer.is_valid(raise_exception=True) 

733 object_ids = serializer.validated_data["object_ids"] 

734 except Exception as e: 

735 return Response( 

736 { 

737 "error": "Validation error", 

738 "details": str(e), 

739 "traceback": traceback.format_exc(), 

740 "received_data": request.data, 

741 }, 

742 status=status.HTTP_400_BAD_REQUEST, 

743 ) 

744 

745 if not object_ids: 745 ↛ 746line 745 didn't jump to line 746 because the condition on line 745 was never true

746 return Response( 

747 {"error": "No objectIds provided"}, status=status.HTTP_400_BAD_REQUEST 

748 ) 

749 

750 flowsheet_ids = list( 

751 SimulationObject.objects.include_deleted() 

752 .filter(pk__in=object_ids) 

753 .values_list("flowsheet_state__flowsheet_id", flat=True) 

754 .distinct() 

755 ) 

756 DeleteFactory._restore_object_ids(object_ids) 

757 for flowsheet_id in flowsheet_ids: 

758 flowsheet = Flowsheet.objects.filter(pk=flowsheet_id).first() 

759 if flowsheet is not None: 759 ↛ 757line 759 didn't jump to line 757 because the condition on line 759 was always true

760 flowsheet.record_content_change() 

761 return Response(status=status.HTTP_200_OK) 

762 

763 @extend_schema(request=DuplicateSimulationObjectSerializer, responses=None) 

764 @action(detail=False, methods=["post"], url_path="duplicate-simulation-object") 

765 def duplicate_simulation_object(self, request): 

766 """Duplicate simulation objects and reposition them at the supplied coordinates.""" 

767 try: 

768 serializer = DuplicateSimulationObjectSerializer(data=request.data) 

769 serializer.is_valid(raise_exception=True) 

770 validated_data = serializer.validated_data 

771 duplicator = DuplicateSimulationObject() 

772 flowsheet = self.request.query_params.get("flowsheet") 

773 flowsheet_instance = Flowsheet.objects.get(pk=flowsheet) 

774 with flowsheet_edit( 

775 flowsheet=flowsheet_instance, 

776 user=request.user, 

777 kind=FlowsheetEditOperation.Kind.Duplicate, 

778 label_key="Duplicate objects", 

779 ) as mutation: 

780 duplicator.handle_duplication_request( 

781 flowsheet, 

782 validated_data, 

783 ) 

784 

785 return Response(mutation.add_to_data({}), status=200) 

786 except SimulationObject.DoesNotExist as e: 

787 return Response({"status": "error", "message": str(e)}, status=400) 

788 except Exception as e: 

789 return Response( 

790 { 

791 "status": "error", 

792 "message": str(e), 

793 "traceback": traceback.format_exc(),