diff --git a/captain-definition b/captain-definition index e5ccb63..b35c8ed 100644 --- a/captain-definition +++ b/captain-definition @@ -1,4 +1,4 @@ { "schemaVersion": 2, - "dockerfilePath": "task_queue/Dockerfile_dev" + "dockerfilePath": "task_queue/Dockerfile" } diff --git a/docker-compose.yml b/docker-compose.yml index a88e460..ddaddc1 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,16 +1,17 @@ -version: "3" services: web: environment: - MVS_DEV_VERSION=${MVS_DEV_VERSION} - MVS_OPEN_PLAN_VERSION=${MVS_OPEN_PLAN_VERSION} + - CELERY_BROKER_URL=redis://redis:6379 + - CELERY_RESULT_BACKEND=redis://redis:6379 build: args: # this should be in a .env file in the root of the repository mvs_version: ${MVS_DEV_VERSION} # `context` should be a path to a directory containing a Dockerfile - context: ./fastapi_app - dockerfile: Dockerfile + context: . + dockerfile: ./fastapi_app/Dockerfile restart: unless-stopped ports: - "5001:5001" @@ -22,13 +23,15 @@ services: worker-dev: environment: - CELERY_TASK_NAME=dev + - CELERY_BROKER_URL=redis://redis:6379 + - CELERY_RESULT_BACKEND=redis://redis:6379 build: args: # this should be in a .env file in the root of the repository mvs_version: ${MVS_DEV_VERSION} # context should be the name of the folder which define the tasks - context: task_queue - dockerfile: Dockerfile_dev + context: . + dockerfile: ./task_queue/Dockerfile depends_on: - redis networks: @@ -36,13 +39,15 @@ services: worker-open-plan: environment: - CELERY_TASK_NAME=open_plan + - CELERY_BROKER_URL=redis://redis:6379 + - CELERY_RESULT_BACKEND=redis://redis:6379 build: args: # this should be in a .env file in the root of the repository mvs_version: ${MVS_OPEN_PLAN_VERSION} # context should be the name of the folder which define the tasks - context: task_queue - dockerfile: Dockerfile + context: . + dockerfile: ./task_queue/Dockerfile depends_on: - redis networks: diff --git a/fastapi_app/Dockerfile b/fastapi_app/Dockerfile index 8db6403..3f5f5fb 100644 --- a/fastapi_app/Dockerfile +++ b/fastapi_app/Dockerfile @@ -6,7 +6,8 @@ COPY fastapi_app/requirements.txt /tmp/ RUN python -m pip install --upgrade pip RUN pip install -r /tmp/requirements.txt # install multi-vector-simulator's version pinned in docker-compose.yml file -RUN pip install multi-vector-simulator==1.0.7rc3 gunicorn +RUN pip install gunicorn +RUN pip install "oemof-datapackage @ git+https://github.com/oemof/oemof-datapackage.git" --no-cache-dir diff --git a/fastapi_app/templates/index.html b/fastapi_app/templates/index.html index 2341eb3..f7c09ac 100644 --- a/fastapi_app/templates/index.html +++ b/fastapi_app/templates/index.html @@ -6,7 +6,7 @@ name="viewport" content="width=device-width, initial-scale=1, shrink-to-fit=no" /> - + - MVS SimServer + OpenPlan SimServer -

Welcome to MVS simulation server!

+

Welcome to OpenPlan simulation server!

+

This simulation server uses oemof-eesyplan as a backend, it recieves request as jsonified oemof-datapackage and returns a JSON with the oemof-solph results

Test simulation with custom json

-

Actual multi-vector-simulator's version for development: {{ mvs_dev_version }}

+

Actual oemof-eesyplan's version for development: {{ dev_version }}

-

Upload a mvs input json file and click on "Run simulation"

+

Upload a JSONIFIED datapackage or a zipped datapackage and click on "Run simulation"

-

Actual multi-vector-simulator's version for open_plan: {{ mvs_open_plan_version }}

+

Actual oemof-eesyplan's version for OpenPlan production: {{ prod_version }}

-

Upload a mvs input json file and click on "Run simulation"

-
- - - +

Upload a datapackage in JSON format and click on "Run simulation"

+ + + +
@@ -45,7 +46,8 @@

Actual multi-vector-simulator's version for open_plan: {{ mvs_open_plan_vers

More information

{% block body %}{% endblock body%} diff --git a/fastapi_app/webapp.py b/fastapi_app/webapp.py index 299ab7b..ac9343a 100644 --- a/fastapi_app/webapp.py +++ b/fastapi_app/webapp.py @@ -1,28 +1,19 @@ +import math import os import json import io -from fastapi import FastAPI, Request, Response, File, UploadFile +from fastapi import FastAPI, Request, Response, File, UploadFile, HTTPException from fastapi.encoders import jsonable_encoder from fastapi.responses import JSONResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from fastapi.responses import StreamingResponse +from celery.exceptions import TimeoutError +import tempfile +import zipfile +from oemof.datapackage import datapackage # noqa -from multi_vector_simulator import version as mvs_version - -from multi_vector_simulator.utils.constants_json_strings import ( - SIMULATION_SETTINGS, - OUTPUT_LP_FILE, - VALUE, - UNIT, -) - -MVS_DEV_VERSION = os.environ.get("MVS_DEV_VERSION", mvs_version.version_num) -MVS_OPEN_PLAN_VERSION = os.environ.get("MVS_OPEN_PLAN_VERSION", mvs_version.version_num) - -MVS_SERVER_VERSIONS = {"dev": MVS_DEV_VERSION, "open_plan": MVS_OPEN_PLAN_VERSION} - try: from worker import app as celery_app except ModuleNotFoundError: @@ -55,6 +46,7 @@ # Test Driven Development --> https://fastapi.tiangolo.com/tutorial/testing/ + @app.get("/debug/ping") async def debug_ping(): result = celery_app.send_task("dev.ping", queue="dev") @@ -77,19 +69,30 @@ async def debug_ping(): @app.get("/") def index(request: Request) -> Response: - return templates.TemplateResponse( - "index.html", - { - "request": request, - "mvs_dev_version": MVS_DEV_VERSION, - "mvs_open_plan_version": MVS_OPEN_PLAN_VERSION, + request=request, + name="index.html", + context={ + "dev_version": get_worker_version("dev"), + "prod_version": get_worker_version("prod"), }, ) +def get_worker_version(queue): + task = celery_app.send_task(f"{queue}.get_version", queue=queue, kwargs={}) + try: + version = task.get(timeout=0.5) + except TimeoutError: + version = f"{queue} worker not available" + return version + + async def simulate_json_variable(request: Request, queue: str = "dev"): """Receive mvs simulation parameter in json post request and send it to simulator""" + + # TODO use jsonschema to verify metadata and data and then use the content of metadata to check data + input_dict = await request.json() # send the task to celery @@ -106,9 +109,9 @@ async def simulate_json_variable_dev(request: Request): return await simulate_json_variable(request, queue="dev") -@app.post("/sendjson/openplan") -async def simulate_json_variable_open_plan(request: Request): - return await simulate_json_variable(request, queue="open_plan") +@app.post("/sendjson/prod") +async def simulate_json_variable_prod(request: Request): + return await simulate_json_variable(request, queue="prod") @app.post("/uploadjson/dev") @@ -119,12 +122,46 @@ def simulate_uploaded_json_files_dev( the value of `name` property of the input html tag should be `json_file` as the second argument of this function """ - json_content = jsonable_encoder(json_file.file.read()) + + if "json" in json_file.filename: + json_content = jsonable_encoder(json_file.file.read()) + elif "zip" in json_file.filename: + # TODO do the same with only pathlib and temp files + import uuid + import shutil + + ZIP_DIR = "zipped" + UNZIP_DIR = "unzipped" + zip_id = str(uuid.uuid4()) + from pathlib import Path + + unzip_path = os.path.join(UNZIP_DIR, zip_id) + os.makedirs(unzip_path) + zip_path = os.path.join(UNZIP_DIR, f"{zip_id}.zip") + with open(zip_path, "wb") as f: + shutil.copyfileobj(json_file.file, f) + with zipfile.ZipFile(zip_path, "r") as zipf: + zipf.extractall(unzip_path) + os.remove(zip_path) + unzip_path = Path(unzip_path) + extracted_name = [f for f in unzip_path.glob("*")] + extracted_name = extracted_name[0] + json_content = datapackage.export_dp_to_json(extracted_name) + + # with tempfile.TemporaryDirectory() as td: + # # Open the zip file + # with zipfile.ZipFile(json_file.file, 'r') as zip_ref: + # # Extract all the contents into the specified directory + # zip_ref.extractall(td) + # + # json_content = json.dumps(datapackage.export_dp_to_json(td)) + # json_content = datapackage.export_dp_to_json(td) + return run_simulation(request, input_json=json_content) -@app.post("/uploadjson/open_plan") -def simulate_uploaded_json_files_open_plan( +@app.post("/uploadjson/prod") +def simulate_uploaded_json_files_prod( request: Request, json_file: UploadFile = File(...) ): """Receive mvs simulation parameter in json post request and send it to simulator @@ -132,7 +169,7 @@ def simulate_uploaded_json_files_open_plan( argument of this function """ json_content = jsonable_encoder(json_file.file.read()) - return run_simulation_open_plan(request, input_json=json_content) + return run_simulation_prod(request, input_json=json_content) def run_simulation(request: Request, input_json=None, queue="dev") -> Response: @@ -152,7 +189,9 @@ def run_simulation(request: Request, input_json=None, queue="dev") -> Response: ) return templates.TemplateResponse( - "submitted_task.html", {"request": request, "task_id": task.id} + request=request, + name="submitted_task.html", + context={"task_id": task.id}, ) @@ -161,9 +200,9 @@ def run_simulation_dev(request: Request, input_json=None) -> Response: return run_simulation(request, input_json, queue="dev") -@app.post("/run_simulation_open_plan") -def run_simulation_open_plan(request: Request, input_json=None) -> Response: - return run_simulation(request, input_json, queue="open_plan") +@app.post("/run_simulation_prod") +def run_simulation_prod(request: Request, input_json=None) -> Response: + return run_simulation(request, input_json, queue="prod") @app.get("/check/{task_id}") @@ -171,7 +210,7 @@ async def check_task(task_id: str) -> JSONResponse: res = celery_app.AsyncResult(task_id) task = { "server_info": None, - "mvs_version": None, + "simulation_version": None, "id": task_id, "status": res.state, "results": None, @@ -181,14 +220,30 @@ async def check_task(task_id: str) -> JSONResponse: else: task["status"] = "DONE" results_as_dict = json.loads(res.result) - server_info = results_as_dict.pop("SERVER") - task["server_info"] = server_info - task["mvs_version"] = MVS_SERVER_VERSIONS.get(server_info, "unknown") - task["results"] = json.dumps(results_as_dict) + try: + task["server_info"] = results_as_dict.pop("SERVER") + task["simulation_version"] = results_as_dict.pop("VERSION") + except KeyError: + for k in ["server_info", "simulation_version"]: + task[k] = "Error: Could not retrieve simulation metadata" + task["results"] = results_as_dict if "ERROR" in task["results"]: task["status"] = "ERROR" task["results"] = results_as_dict + def clean_floats(obj): + if isinstance(obj, float): + if math.isnan(obj) or math.isinf(obj): + return None + if isinstance(obj, dict): + return {k: clean_floats(v) for k, v in obj.items()} + if isinstance(obj, list): + return [clean_floats(i) for i in obj] + return obj + + cleaned = clean_floats(task) + + return JSONResponse(content=jsonable_encoder(cleaned)) return JSONResponse(content=jsonable_encoder(task)) @@ -197,7 +252,7 @@ async def get_lp_file(task_id: str) -> Response: res = celery_app.AsyncResult(task_id) task = { "server_info": None, - "mvs_version": mvs_version, + "simulation_version": None, "id": task_id, "status": res.state, "results": None, @@ -208,9 +263,8 @@ async def get_lp_file(task_id: str) -> Response: else: task["status"] = "DONE" results_as_dict = json.loads(res.result) - server_info = results_as_dict.pop("SERVER") - task["server_info"] = server_info - task["mvs_version"] = MVS_SERVER_VERSIONS.get(server_info, "unknown") + task["server_info"] = results_as_dict.pop("SERVER") + task["simulation_version"] = results_as_dict.pop("VERSION") task["results"] = json.dumps(results_as_dict) if "ERROR" in task["results"]: task["status"] = "ERROR" diff --git a/fastapi_app/worker.py b/fastapi_app/worker.py index d7a54eb..13032b8 100644 --- a/fastapi_app/worker.py +++ b/fastapi_app/worker.py @@ -11,5 +11,5 @@ app = Celery("tasks", broker=CELERY_BROKER_URL, backend=CELERY_RESULT_BACKEND) app.conf.task_queues = ( Queue("dev", routing_key="dev.#"), - Queue("open_plan", routing_key="open_plan.#"), + Queue("prod", routing_key="prod.#"), ) diff --git a/task_queue/Dockerfile b/task_queue/Dockerfile index d0216c0..af4a200 100644 --- a/task_queue/Dockerfile +++ b/task_queue/Dockerfile @@ -1,26 +1,24 @@ -FROM python:3.8-slim - -ARG mvs_version - -ENV MVS_VERSION $mvs_version - -ENV CELERY_BROKER_URL redis://redis:6379/0 -ENV CELERY_RESULT_BACKEND redis://redis:6379/0 +FROM python:3.12-slim RUN apt-get update && \ apt-get install -y git && \ apt-get install coinor-cbc -y && \ - apt-get install graphviz -y + apt-get install graphviz -y && \ + apt-get install tk -y -COPY requirements.txt /tmp/ +COPY task_queue/requirements.txt /tmp/ RUN python -m pip install --upgrade pip RUN pip install -r /tmp/requirements.txt -# install multi-vector-simulator's version pinned in docker-compose.yml file -RUN pip install multi-vector-simulator==$MVS_VERSION importlib-metadata==4.13.0 pyomo==5.7.2 +# install oemof-eesyplan version from the ENV variable +#RUN pip install "oemof-eesyplan==$SIMULATION_VERSION" gunicorn +RUN pip install "git+https://github.com/oemof/oemof-eesyplan.git@main" --no-cache-dir +RUN pip install gunicorn +RUN pip install "oemof-datapackage @ git+https://github.com/oemof/oemof-datapackage.git" + run echo $(pip freeze) -COPY . /queue +COPY task_queue/ /queue # avoid running as root user RUN useradd --create-home appuser diff --git a/task_queue/Dockerfile_dev b/task_queue/Dockerfile_dev deleted file mode 100644 index 801dcb0..0000000 --- a/task_queue/Dockerfile_dev +++ /dev/null @@ -1,27 +0,0 @@ -FROM python:3.10-slim - - - -RUN apt-get update && \ - apt-get install -y git && \ - apt-get install coinor-cbc -y && \ - apt-get install graphviz -y - -COPY task_queue/requirements.txt /tmp/ - -RUN python -m pip install --upgrade pip -RUN pip install -r /tmp/requirements.txt -# install multi-vector-simulator's version pinned in docker-compose.yml file -RUN pip install multi-vector-simulator==1.0.7rc3 importlib-metadata==4.13.0 oemof.network==0.5.0a5 - -run echo $(pip freeze) -COPY task_queue/ /queue - -# avoid running as root user -RUN useradd --create-home appuser -RUN chown -R appuser /queue -USER appuser - -WORKDIR /queue - -ENTRYPOINT celery -A tasks worker --loglevel=info --queues=$CELERY_TASK_NAME \ No newline at end of file diff --git a/task_queue/requirements.txt b/task_queue/requirements.txt index 9fc707f..8a2f582 100644 --- a/task_queue/requirements.txt +++ b/task_queue/requirements.txt @@ -2,3 +2,4 @@ celery flower redis python-multipart +oemof-visio \ No newline at end of file diff --git a/task_queue/tasks.py b/task_queue/tasks.py index 66b3d47..6529f9c 100644 --- a/task_queue/tasks.py +++ b/task_queue/tasks.py @@ -1,16 +1,25 @@ import os -import socket + import time import traceback import json -from copy import deepcopy from celery import Celery from celery.utils.log import get_task_logger +import tempfile +from pathlib import Path + +from oemof.datapackage import datapackage # noqa +from oemof.datapackage import __version__ as dp_version + +from oemof.eesyplan import export_results, import_results +from oemof.eesyplan.datapackage.energy_system import create_energy_system_from_dp +from oemof.eesyplan.model import optimise -from multi_vector_simulator.server import run_simulation as mvs_simulation -from multi_vector_simulator.utils.data_parser import convert_epa_params_to_mvs +import oemof.eesyplan.postprocessing.graphs as eesyplan_graphs +SIMULATION_VERSION = os.environ.get("SIMULATION_VERSION", "no_version") + logger = get_task_logger(__name__) CELERY_BROKER_URL = (os.environ.get("CELERY_BROKER_URL", "redis://localhost:6379"),) CELERY_RESULT_BACKEND = os.environ.get( @@ -22,31 +31,58 @@ app = Celery(CELERY_TASK_NAME, broker=CELERY_BROKER_URL, backend=CELERY_RESULT_BACKEND) -@app.task(name=f"{CELERY_TASK_NAME}.run_simulation") -def run_simulation(simulation_input: dict,) -> dict: +def __run_simulation( + simulation_input): logger.info("Start new simulation") - epa_json = deepcopy(simulation_input) - dict_values = None - try: - dict_values = convert_epa_params_to_mvs(simulation_input) - logger.debug("Converted epa parameters to mvs input") - simulation_output = mvs_simulation(dict_values) - logger.info("Simulation finished") - simulation_output["SERVER"] = CELERY_TASK_NAME - except Exception as e: - logger.error( - "An exception occured in the simulation task: {}".format( - traceback.format_exc() + simulation_output = {"SERVER": CELERY_TASK_NAME, "VERSION": SIMULATION_VERSION} + logger.info(f"Using datapackage version: {dp_version}") + with tempfile.TemporaryDirectory(prefix="dp_") as td: + temp_path = Path(td) + dp_path = datapackage.rebuild_dp_from_json(simulation_input, temp_path) + try: + es = create_energy_system_from_dp(dp_path) + + results = optimise(es) + + with tempfile.TemporaryDirectory(prefix="dp_results_") as tres: + results_path = Path(tres) + + export_results(results, path=results_path) + json_export = datapackage.export_dp_to_json(results_path) + simulation_output["raw_results"] = json.loads(json_export) + + imported_results = import_results(path=results_path, es=es) + fig, links_df = eesyplan_graphs.sankey(imported_results["flow"], title="Test Sankey") + simulation_output["figures"] = {"sankey": fig.to_dict()} + + except Exception as e: + logger.error( + "An exception occured in the simulation task: {}".format( + traceback.format_exc() + ) + ) + simulation_output.update( + dict( + ERROR="{}".format(traceback.format_exc()), + INPUT_JSON=simulation_input, + ) ) - ) - simulation_output = dict( - SERVER=CELERY_TASK_NAME, - ERROR="{}".format(traceback.format_exc()), - INPUT_JSON_EPA=simulation_input, - INPUT_JSON_MVS=dict_values, - ) + return json.dumps(simulation_output) + +@app.task(name=f"{CELERY_TASK_NAME}.run_simulation") +def run_simulation( + simulation_input: dict, +) -> dict: + return __run_simulation(simulation_input) + + +@app.task(name=f"{CELERY_TASK_NAME}.get_version") +def get_version() -> str: + return SIMULATION_VERSION + + @app.task(bind=True, name="dev.ping") def ping(self): return { @@ -56,3 +92,20 @@ def ping(self): "pid": os.getpid(), "timestamp": time.time(), } + +if __name__ == "__main__": + with open('sim_data_debug.json', 'r') as file: + dp = json.load(file) + + # Run simulation locally + result = run_simulation(dp) + + # If Celery decorator wraps the output inside AsyncResult, unwrap it + if hasattr(result, "get"): + result = result.get() + + # Store the exact server output + out_path = Path("debug_simulation_output.json") + out_path.write_text(json.dumps(result, indent=2)) + + print("Simulation results saved to:", out_path.resolve()) \ No newline at end of file