Merge pull request #14 from sogno-platform/feature/api

working powerflow calculation on pydantic loaded cim data

src/pgm_service/app.py

@@ -3,13 +3,11 @@
 
 from power_grid_model import initialize_array  # TODO(mgovers) remove
 
-from pgm_service.power_grid.router import router as pg_router
 from pgm_service.pgm_powerflow.router import router as pf_router
 
 
 app = FastAPI(title="API")
 
-app.include_router(pg_router, prefix="/api")
 app.include_router(pf_router, prefix="/api")
 
 

src/pgm_service/pgm_powerflow/aux_models.py

@@ -1,8 +1,8 @@
 from enum import Enum
 from typing import Optional
-from pydantic import BaseModel, ConfigDict, Field
 from datetime import datetime
 
+from pydantic import BaseModel, ConfigDict, Field
 
 from pgm_service.pgm_powerflow.models import PGM_Powerflow
 
@@ -31,4 +31,4 @@ class JobComplete(JobBase):
     details: Optional[str] = None
     created: datetime = Field(default_factory=datetime.now)
     finished: Optional[datetime] = None
-    result: PGM_Powerflow = None
+    result: PGM_Powerflow = None

src/pgm_service/pgm_powerflow/models.py

@@ -13,9 +13,7 @@ class StrEnum(str, Enum):
     pass
 
 
-# Basic input data
-class PGM_Powerflow(BaseModel):
-    model: Union[str, Grid]
+class PGM_PowerflowCalculationArgs(BaseModel):
     symmetric: bool = True
     error_tolerance: float = 1e-8
     max_iterations: int = 20
@@ -24,3 +22,9 @@ class PGM_Powerflow(BaseModel):
     # threading: int = -1
     output_component_types: Optional[List[str]] = None
     # continue_on_batch_error: bool = False
+
+
+# Basic input data
+class PGM_Powerflow(BaseModel):
+    model: Grid
+    calculation_args: PGM_PowerflowCalculationArgs

src/pgm_service/pgm_powerflow/router.py

@@ -1,39 +1,57 @@
-from fastapi import APIRouter
+from datetime import datetime
+from typing import Dict
+from uuid import uuid4
+from fastapi import APIRouter, BackgroundTasks
 
-from power_grid_model import PowerGridModel
-
-from pgm_service.pgm_powerflow.aux_models import JobComplete
+from pgm_service.pgm_powerflow.aux_models import JobComplete, Status
 from pgm_service.pgm_powerflow.models import PGM_Powerflow
-from pgm_service.power_grid.models import Grid
+from pgm_service.power_grid.power_grid_model import calculate_powerflow
 
 
 router = APIRouter(prefix="/pgm_powerflow", tags=["Powerflow"])
 
+JOBS: Dict[str, JobComplete] = {}
+
 
 @router.get("/")
 async def get_all_powerflow_calculation() -> list[str]:
-    """Returns list of existing powerflow_calculation IDs"""  # XXX should this return url/uris?
-    raise NotImplementedError()  # TODO this should look up ids from DB and return them
+    """Returns list of existing powerflow_calculation IDs"""
+    return list(JOBS.keys())
+
+
+async def _calculate(job: JobComplete):
+    job.status = Status.RUNNING
+
+    try:
+        grid = job.input.model
+        pf_kwargs = job.input.calculation_args.model_dump()
+
+        await calculate_powerflow(grid=grid, pf_kwargs=pf_kwargs)
+
+        job.finished = datetime.now()
+        job.status = Status.SUCCESS
+    except Exception as e:
+        job.status = Status.FAILED
+        job.details = e
 
 
 @router.post("/")
 async def new_powerflow_calculation(
     resource: PGM_Powerflow,
+    background_tasks: BackgroundTasks,
 ) -> JobComplete:  # TODO should be wrapped in jonb
-    # raise NotImplementedError()  # TODO This should create a new job entry in DB
-    assert isinstance(resource.model, Grid)
-    model = PowerGridModel(input_data={}, system_frequency=resource.model.system_frequency)
-    pf_args = resource.model_dump()
-    del pf_args["model"]
-    calculation_result = model.calculate_power_flow(**pf_args)
-    print(calculation_result)
-    job = JobComplete(id="test", input=resource)
+    _id = str(uuid4())
+    JOBS[_id] = JobComplete(id=_id, input=resource)
+    job = JOBS[_id]
+
+    background_tasks.add_task(_calculate, job=job)
+
     return job
 
 
 @router.get("/{id}")
 async def get_powerflow_calculation(id: str) -> JobComplete:
-    raise NotImplementedError()  # TODO fetch Job with ID from the DB
+    return JOBS[id]
 
 
 # @router.put("/{id}")
@@ -44,4 +62,4 @@ async def get_powerflow_calculation(id: str) -> JobComplete:
 
 @router.delete("/{id}")
 async def delete_powerflow_calculation(id: str) -> JobComplete:
-    raise NotImplementedError()  # TODO this should fetch the hob if possible then delete and return it
+    return JOBS.pop(id)

src/pgm_service/power_grid/power_grid_model.py

@@ -0,0 +1,58 @@
+import glob
+import os
+from pathlib import Path
+from typing import Dict, Any
+import requests
+import cimpy
+from power_grid_model import PowerGridModel
+
+from pgm_service.power_grid.models import Grid
+from pgm_service.power_grid.cgmes_pgm_converter import System as CGMESToPGMConverter
+
+
+def download_data(url):
+    def download_grid_data(name, url):
+        with open(name, 'wb') as out_file:
+            content = requests.get(url, stream=True).content
+            out_file.write(content)
+
+    url = 'https://raw.githubusercontent.com/dpsim-simulator/cim-grid-data/master/BasicGrids/NEPLAN/Slack_Load_Line_Sample/'
+    filename = 'Rootnet_FULL_NE_19J18h'
+
+    download_grid_data(filename+'_EQ.xml', url + filename + '_EQ.xml')
+    download_grid_data(filename+'_TP.xml', url + filename + '_TP.xml')
+    download_grid_data(filename+'_SV.xml', url + filename + '_SV.xml')
+
+    files = glob.glob(filename+'_*.xml')
+
+    print('CGMES files downloaded:')
+    print(files)
+
+    this_file_folder =  Path(__file__).parents[3]
+    p = str(this_file_folder)
+    xml_path = Path(p)
+    xml_files = [os.path.join(xml_path, filename+'_EQ.xml'),
+                os.path.join(xml_path, filename+'_TP.xml'),
+                os.path.join(xml_path, filename+'_SV.xml')]
+
+    print(xml_files)
+    return xml_files
+
+
+def create_model(grid: Grid):  # TODO make async
+    xml_files = download_data(url=grid.input_data)
+    cgmes_data = cimpy.cim_import(xml_files, "cgmes_v2_4_15")
+    converter = CGMESToPGMConverter()
+    converter.load_cim_data(cgmes_data)
+    pgm_data = converter.create_pgm_input()
+    return PowerGridModel(input_data=pgm_data, system_frequency=grid.system_frequency)
+
+
+def produce_output(grid: Grid, calculation_result: Any):
+    print(calculation_result)
+
+
+async def calculate_powerflow(grid: Grid, pf_kwargs: Dict[str, Any]):
+    model = create_model(grid=grid)
+    calculation_result = model.calculate_power_flow(**pf_kwargs)
+    produce_output(grid, calculation_result)