Coverage for backend/django/core/auxiliary/services/flowsheet_states/module_subtree.py: 88%

1"""Separate module-template insertion built on clone registry primitives.""" 

2 

3from django.db import transaction 

4 

5from core.auxiliary.enums import ConType 

6from flowsheetInternals.graphicData.logic.make_group import ( 

7 propagate_intermediate_streams, 

8 propagate_streams, 

9) 

10from flowsheetInternals.unitops.models.SimulationObject import SimulationObject 

11 

12from .cloning import clone_state 

13from .registry import CloneProfile 

14 

15 

16@transaction.atomic 

17def clone_module_subtree( 

18 *, 

19 template_flowsheet, 

20 target_flowsheet, 

21 current_group, 

22 x=None, 

23 y=None, 

24): 

25 """Insert a template's core aggregate as a module in the current state.""" 

26 

27 result = clone_state( 

28 source_state=template_flowsheet.current_state, 

29 target_state=target_flowsheet.current_state, 

30 profile=CloneProfile.MODULE_SUBTREE, 

31 ) 

32 module_root = result.root_grouping 

33 if module_root is None: 33 ↛ 34line 33 didn't jump to line 34 because the condition on line 33 was never true

34 raise ValueError("Module template has no copied root grouping.") 

35 graphic = module_root.get_graphic_object() 

36 graphic.group = current_group 

37 if x is not None: 37 ↛ 39line 37 didn't jump to line 39 because the condition on line 37 was always true

38 graphic.x = x 

39 if y is not None: 39 ↛ 41line 39 didn't jump to line 41 because the condition on line 39 was always true

40 graphic.y = y 

41 graphic.save() 

42 

43 copied_objects = list(result.model_maps.get(SimulationObject, {}).values()) 

44 contained_ids = [simulation_object.pk for simulation_object in copied_objects] 

45 inlet_streams = [] 

46 outlet_streams = [] 

47 for simulation_object in copied_objects: 

48 ports = simulation_object.connectedPorts 

49 if simulation_object.is_stream() and ports.count() == 1: 

50 port = ports.first() 

51 if port and port.direction == ConType.Inlet: 

52 inlet_streams.append(simulation_object) 

53 else: 

54 outlet_streams.append(simulation_object) 

55 elif simulation_object.is_stream() and ports.count() == 2: 55 ↛ 56line 55 didn't jump to line 56 because the condition on line 55 was never true

56 propagate_intermediate_streams(simulation_object, contained_ids) 

57 propagate_streams(inlet_streams, ConType.Inlet) 

58 propagate_streams(outlet_streams, ConType.Outlet) 

59 return module_root