Coverage for backend/idaes_factory/adapters/unit_models/header_adapter.py: 96%

1from idaes_factory.queryset_lookup import get_all_ports 

2from ..core_adapters import * 

3from core.auxiliary.enums import ConType 

4from flowsheetInternals.unitops.models import * 

5from ..generic_adapters import NumInletsAdapter, NumOutletsAdapter 

6from ..stream_properties import serialise_stream 

7 

8class HeaderPortListAdapter(PortListAdapter): 

9 """ 

10 Handles the serialisation and reloading of the ports of a header. 

11 Includes separate handling for condensate_outlet port. 

12 """ 

13 def __init__(self): 

14 pass 

15 

16 def serialise(self, ctx, unit_model: SimulationObject): 

17 portList = {} 

18 

19 # Handle inlets 

20 inlets = [ i for i in unit_model.ports.filter(direction=ConType.Inlet, stream__isnull=False)] 

21 

22 # Sort regular inlets by index 

23 inlets.sort(key=lambda x: x.index if hasattr(x, 'index') else 0) 

24 

25 # Add regular inlets 

26 for i, inlet in enumerate(inlets): 

27 portList[f"mixer.inlet_{i+1}"] = { 

28 "id": inlet.pk, 

29 "properties": serialise_stream(ctx, inlet.stream, is_inlet=True) 

30 } 

31 

32 # Handle outlets 

33 outlets = unit_model.ports.filter(direction=ConType.Outlet, stream__isnull=False) 

34 

35 # Separate special outlets 

36 condensate_outlet = None 

37 vent_outlet = None 

38 regular_outlets = [] 

39 

40 for outlet in outlets: 

41 if outlet.key == "condensate_outlet": 

42 condensate_outlet = outlet 

43 elif outlet.key == "vent": 

44 vent_outlet = outlet 

45 else: 

46 regular_outlets.append(outlet) 

47 

48 # Sort regular outlets 

49 regular_outlets.sort(key=lambda x: x.index if hasattr(x, 'index') else 0) 

50 # Add regular outlets 

51 for i, outlet in enumerate(regular_outlets): 

52 portList[f"splitter.outlet_{i+1}"] = { 

53 "id": outlet.pk, 

54 "properties": serialise_stream(ctx, outlet.stream, is_inlet=False) 

55 } 

56 

57 # Add vent outlet if it exists (named "vent" not "outlet_N") 

58 if vent_outlet: 58 ↛ 65line 58 didn't jump to line 65 because the condition on line 58 was always true

59 portList["splitter.vent"] = { 

60 "id": vent_outlet.pk, 

61 "properties": serialise_stream(ctx, vent_outlet.stream, is_inlet=False) 

62 } 

63 

64 # Add condensate outlet 

65 if condensate_outlet: 65 ↛ 71line 65 didn't jump to line 71 because the condition on line 65 was always true

66 portList["phase_separator.condensate_outlet"] = { 

67 "id": condensate_outlet.pk, 

68 "properties": serialise_stream(ctx, condensate_outlet.stream, is_inlet=False) 

69 } 

70 

71 return portList 

72 

73class HeaderOutletsAdapter(ValueAdapter): 

74 def serialise(self, ctx, unit_model: SimulationObject): 

75 return len(get_all_ports(unit_model, "outlet")) 

76 

77class HeaderInletsAdapter(ValueAdapter): 

78 def serialise(self, ctx, unit_model: SimulationObject): 

79 return len(get_all_ports(unit_model, "inlet")) 

80 

81header_adapter = UnitModelAdapter( 

82 args=ArgAdapter({ 

83 "property_package": PropertyPackageAdapter(), 

84 "num_inlets": HeaderInletsAdapter(), 

85 "num_outlets": HeaderOutletsAdapter() 

86 }), 

87 properties=SerialisePropertiesAdapter(), 

88 ports=HeaderPortListAdapter() 

89)