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

1from typing import Optional 

2 

3from ahuora_builder_types.arc_schema import TearGuessSchema 

4from core.auxiliary.enums import ConType 

5from flowsheetInternals.unitops.models.SimulationObject import SimulationObject 

6from ahuora_builder_types import ArcSchema, FlowsheetSchema 

7from ..idaes_factory_context import IdaesFactoryContext 

8 

9from ..queryset_lookup import get_property, get_connected_port, get_value_object 

10 

11 

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

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

14 

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

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

17 

18 if from_port is None or to_port is None: 

19 # no inlet or outlet port, skip this stream 

20 return None 

21 

22 arc_schema: ArcSchema = ArcSchema( 

23 id=stream.pk, 

24 source=from_port.pk, 

25 destination=to_port.pk 

26 ) 

27 if stream.has_recycle_connection: 

28 arc_schema.tear_guess = create_tear(stream) 

29 

30 return arc_schema 

31 

32 

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

34 """Create a tear for a recycle stream""" 

35 tear = {} 

36 

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

38 if prop.is_recycle_var(): 

39 tear[prop.key] = prop.recycleConnection.fixed 

40 

41 return tear