Coverage for backend/django/pgraph_factory/pg_sheet.py: 74%

1 

2from .types import PgraphDetails, BlockSchema, ConnectionSchema 

3from flowsheetInternals.unitops.models import SimulationObject 

4from core.auxiliary.models.Flowsheet import Flowsheet 

5from core.auxiliary.models.ProcessPath import ProcessPath 

6import os 

7import requests 

8 

9 

10""" 

11This class processes p-graph data  

12""" 

13class PgProcess: 

14 

15 def __init__(self, id: int) -> None: 

16 self._process_details: PgraphDetails = { 

17 "blocks": [], # unit, stream, and decisionNode 

18 "connections": [] 

19 } 

20 self.flowsheet = Flowsheet.objects.get(id=id) 

21 self.flowsheet_state = self.flowsheet.current_state 

22 

23 exclude = {"group"} 

24 self.flowsheet_objects = SimulationObject.objects.filter( 

25 flowsheet_state=self.flowsheet_state 

26 ).exclude(objectType__in=exclude) 

27 

28 for block in self.flowsheet_objects: 

29 self.add_block(block) 

30 self.add_connections(block) 

31 

32 self.solutions: list[list[SimulationObject]] = [] # list of list of block objects 

33 

34 

35 def add_connections(self, block: SimulationObject) -> None: 

36 """ 

37 Adds connections for the given block. 

38 

39 Parameters: 

40 - block (SimulationObject): The block to add connections to. 

41 """ 

42 ports = block.ports.filter(stream__isnull=False) 

43 for port in ports: 

44 stream = port.stream 

45 connection: ConnectionSchema 

46 if port.direction == "inlet": 

47 connection = [stream.id, block.id] 

48 else: 

49 connection = [block.id, stream.id] 

50 self._process_details["connections"].append(connection) 

51 

52 

53 def add_block(self, block: SimulationObject) -> None: 

54 """ 

55 Adds a block in the format of BlockSchema to process details 

56 """ 

57 block_data: BlockSchema = { 

58 "id": block.id, 

59 "name": block.componentName, 

60 "type": block.objectType if block.objectType in ["stream","energy_stream", "decisionNode"] else "unit", 

61 } 

62 self._process_details["blocks"].append(block_data) 

63 

64 

65 def solve(self): 

66 """ 

67 Solves the process graph 

68 """ 

69 url = (os.getenv('PGRAPH_SERVICE_URL') or "http://localhost:8081") + "/solve" 

70 result = requests.post(url, json=self._process_details) 

71 # check if the request was successful 

72 if result.status_code != 200: 

73 raise Exception("Error solving the process graph", result.json()) 

74 content = result.json() 

75 self.solutions = [[self.flowsheet_objects.get(id=id) for id in solution] for solution in content] 

76 

77 

78 def create_process_paths(self): 

79 """ 

80 Creates the process paths for the solutions 

81 """ 

82 # delete the existing process paths 

83 self.flowsheet_state.ProcessPaths.all().delete() 

84 # create process paths for each solution 

85 for solution in self.solutions: 

86 ProcessPath.create( 

87 flowsheet_state=self.flowsheet_state, 

88 pathwayObjects=solution, 

89 )