Coverage for skema/gromet/execution_engine/server.py: 100%

1from typing import Dict 

2from tempfile import TemporaryDirectory 

3from pathlib import Path 

4 

5from fastapi import APIRouter, FastAPI 

6from pydantic import BaseModel, Field 

7 

8from skema.rest.proxies import SKEMA_GRAPH_DB_HOST, SKEMA_GRAPH_DB_PORT, SKEMA_GRAPH_DB_PROTO 

9from skema.gromet.execution_engine.execution_engine import ExecutionEngine 

10HOST = SKEMA_GRAPH_DB_HOST 

11PORT = int(SKEMA_GRAPH_DB_PORT) 

12PROTOCOL = SKEMA_GRAPH_DB_PROTO 

13print("LOGGING: Lanuching execution engine REST service") 

14print(f"LOGGING: SKEMA_GRAPH_DB_PROTOCOL {PROTOCOL}") 

15print(f"LOGGING: SKEMA_GRAPH_DB_HOST {HOST}") 

16print(f"LOGGING: SKEMA_GRAPH_DB_PORT {PORT}") 

17 

18router = APIRouter() 

19 

20class EnrichmentReqest(BaseModel): 

21 amr: Dict = Field( 

22 description="The amr to enrich with parameter values", 

23 examples=[{ 

24 "semantics": { 

25 "ode":{ 

26 "parameters":[ 

27 {"name": "a"}, 

28 {"name": "b"}, 

29 {"name": "c"} 

30 ] 

31 } 

32 } 

33 }] 

34 ) 

35 source: str = Field( 

36 description="The raw source code to extract parameter values from", 

37 examples=["a=1\nb=a+1\nc=b-a"] 

38 ) 

39 filename: str = Field( 

40 description="The filename of the file passed in the 'source' field", 

41 examples=["source.py"] 

42 ) 

43 

44@router.post("/amr-enrichment", summary="Given an amr and source code, return an enriched amr with parameter values filled in.", response_model=Dict, responses={ 

45 200: { 

46 "content": { 

47 "application/json":{ 

48 "example": { 

49 "semantics": { 

50 "ode":{ 

51 "parameters":[ 

52 {"name": "a", "value": 1}, 

53 {"name": "b", "value": 2}, 

54 {"name": "c", "value": 1} 

55 ] 

56 } 

57 } 

58 } 

59 } 

60 } 

61 } 

62}) 

63def amr_enrichment(request: EnrichmentReqest): 

64 """ 

65 Endpoint for enriching amr with parameter values. 

66 ### Python example 

67 

68 ``` 

69 import requests 

70 

71 request = { 

72 "amr": { 

73 "semantics": { 

74 "ode":{ 

75 "parameters":[ 

76 {"name": "a"}, 

77 {"name": "b"}, 

78 {"name": "c"} 

79 ] 

80 } 

81 } 

82 }, 

83 "source": "a=1\\nb=a+1\\nc=b-a", 

84 "filename": "source.py" 

85 } 

86 

87 response=client.post("/execution_engine/amr-enrichment", json=request) 

88 enriched_amr = response.json() 

89 """ 

90 with TemporaryDirectory() as temp: 

91 source_path = Path(temp) / request.filename 

92 source_path.write_text(request.source) 

93 

94 engine = ExecutionEngine(PROTOCOL, HOST, PORT, str(source_path)) 

95 engine.execute(module=True) 

96 

97 return engine.enrich_amr(request.amr) 

98 

99app = FastAPI() 

100app.include_router( 

101 router, 

102 prefix="/execution-engine", 

103 tags=["execution-engine"], 

104)