Coverage for backend/django/idaes_factory/adapters/stream_properties.py: 100%

1from ahuora_builder_types import PropertiesSchema 

2from .property_info_adapter import SerialisePropertiesAdapter 

3from ..queryset_lookup import get_property 

4 

5 

6""" 

7Stream serialisation using SerialisePropertiesAdapter. 

8Handles material streams, energy streams, and other stream types. 

9""" 

10 

11_stream_adapter = SerialisePropertiesAdapter() 

12 

13 

14def _is_boundary_stream(ctx,stream) -> bool: 

15 ports_in_group = [ 

16 port for port in stream.connectedPorts.all() 

17 if ctx.has_loaded_simulation_object(port.unitOp_id) 

18 ] 

19 return len(ports_in_group) == 1 

20 

21def _is_boundary_stream_for_current_group(ctx, stream) -> bool: 

22 """Return whether this stream has exactly one endpoint in the solve group.""" 

23 

24 return _is_boundary_stream(ctx, stream) and len(stream.connectedPorts.all()) > 1 

25 

26def should_serialise_stream(ctx, stream, is_inlet: bool) -> bool: 

27 """ 

28 Check if a stream should be serialised. 

29 """ 

30 if _is_boundary_stream(ctx, stream): 

31 # ie. inlet or outlet port 

32 return True 

33 if stream.has_recycle_connection: 

34 # serialise both the inlet (for guesses) and outlet (for set points) 

35 return True 

36 else: 

37 # serialise the outlet for intermediate ports 

38 return not is_inlet 

39 

40 

41def serialise_stream(ctx, stream, is_inlet: bool) -> PropertiesSchema: 

42 """ 

43 Serialise the properties of a material stream. 

44 """ 

45 if not should_serialise_stream(ctx, stream, is_inlet): 

46 return {} 

47 

48 tear_inlet = stream.has_recycle_connection and is_inlet 

49 tear_outlet = stream.has_recycle_connection and not is_inlet 

50 boundary_inlet = is_inlet and _is_boundary_stream_for_current_group(ctx, stream) 

51 

52 # Serialize all properties with the correct is_tear flag 

53 result = _stream_adapter.serialise( 

54 ctx, 

55 stream, 

56 is_tear=tear_inlet, 

57 force_enabled=boundary_inlet, 

58 ) 

59 

60 # For tear outlets, filter out properties that are not set points 

61 if tear_outlet: 

62 for prop in stream.properties.containedProperties.all(): 

63 for value in prop.values.all(): 

64 if not value.is_control_set_point() or not value.is_externally_controlled(): 

65 result.pop(prop.key, None) 

66 break # Exit values loop once we decide to remove this property 

67 

68 return result