Coverage for backend/django/flowsheetInternals/unitops/models/property_package_propogation.py: 100%

1from flowsheetInternals.unitops.models.SimulationObject import SimulationObject 

2 

3from flowsheetInternals.unitops.models.flow_tracking import track_stream_flow 

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

5 tracked_queryset_update, 

6) 

7 

8 

9def _starting_streams(simulation_object: SimulationObject) -> set[SimulationObject]: 

10 """Return traversal roots for either a stream or a unit operation.""" 

11 if simulation_object.is_stream(): 

12 return {simulation_object} 

13 return { 

14 port.stream 

15 for port in simulation_object.ports.all() 

16 if port.stream_id is not None 

17 } 

18 

19 

20def propogate_property_package(simulation_object: SimulationObject): 

21 """ 

22 Propagate a selected package through every connected material-flow network. 

23 

24 The property-package chooser edits the selected unit operation, while some 

25 internal callers start from a stream. Supporting both roots keeps attached 

26 streams consistent with the package visible in the UI. 

27 """ 

28 starting_streams = _starting_streams(simulation_object) 

29 streams: set[SimulationObject] = set() 

30 for starting_stream in starting_streams: 

31 _, connected_streams = track_stream_flow(starting_stream) 

32 streams.update(connected_streams) 

33 

34 tracked_queryset_update( 

35 SimulationObject.objects.filter(id__in=[stream.id for stream in streams]), 

36 propertyPackageType=simulation_object.propertyPackageType, 

37 customPackage=simulation_object.customPackage, 

38 ) 

39 

40 # Package-specific stream schemas (currently humid air) are part of the 

41 # same propagation operation. Keeping this here makes unit- and 

42 # stream-originated package selections follow the same path. 

43 from flowsheetInternals.propertyPackages.models.StreamFactory import StreamFactory 

44 

45 for starting_stream in starting_streams: 

46 starting_stream.refresh_from_db() 

47 StreamFactory(starting_stream).check_and_update_stream()