Coverage for backend/idaes_factory/adapters/arc_adapter.py: 100%

1from typing import Optional 

2 

3from common.models.idaes.arc_schema import TearGuessSchema 

4from core.auxiliary.enums import ConType 

5from flowsheetInternals.unitops.models.SimulationObject import SimulationObject 

6from common.models.idaes import ArcSchema, FlowsheetSchema 

7from .stream_properties import stream_properties 

8from ..idaes_factory_context import IdaesFactoryContext 

9 

10from ..queryset_lookup import get_property, get_connected_port, get_value_object 

11 

12 

13def create_arc(ctx: IdaesFactoryContext, stream: SimulationObject) -> Optional[ArcSchema]: 

14 """Adds a material stream to the flowsheet schema""" 

15 

16 from_port = get_connected_port(stream, direction=ConType.Outlet) 

17 to_port = get_connected_port(stream, direction=ConType.Inlet) 

18 

19 if from_port is None or to_port is None: 

20 # no inlet or outlet port, skip this stream 

21 return None 

22 

23 arc_schema: ArcSchema = ArcSchema( 

24 id=stream.pk, 

25 source=from_port.pk, 

26 destination=to_port.pk 

27 ) 

28 if stream.has_recycle_connection: 

29 arc_schema.tear_guess = create_tear(stream) 

30 

31 return arc_schema 

32 

33 

34def create_tear(stream: SimulationObject) -> dict[str, bool]: 

35 """Create a tear for a recycle stream""" 

36 # eventually might be nice to use the same keys as in IDAES 

37 key_map = { 

38 "mole_frac_comp": "mole_frac_comp", 

39 "flow_mol": "flow_mol", # because of this one :( 

40 "temperature": "temperature", 

41 "pressure": "pressure" 

42 } 

43 tear = {} 

44 

45 for prop in stream.properties.ContainedProperties.all(): 

46 if prop.is_recycle_var(): 

47 tear[key_map[prop.key]] = prop.recycleConnection.fixed 

48 

49 return tear