Alberto Morales (as candicate) coding challenge

.gitignore

@@ -0,0 +1,2 @@
+api/code/.vscode/**
+api/code/**/*.pyc

api/Dockerfile

@@ -0,0 +1,16 @@
+FROM python:3.10-slim
+
+# Just for Python Execution
+ENV PYTHONDONTWRITEBYTECODE 1
+ENV PYTHONUNBUFFERED 1
+ENV PYTHONPATH "${PYTHONPATH}:/powerplanner"
+
+
+WORKDIR /powerplanner
+COPY code/ /powerplanner/
+RUN pip install --no-cache-dir -r requirements.txt
+
+
+EXPOSE 8888
+
+CMD ["uvicorn", "run_server:app", "--host", "0.0.0.0", "--port", "8888"]

api/code/.env

@@ -0,0 +1 @@
+PYTHONPATH=/home/tucho/workspaceTakeoff/powerplant-coding-challenge/api/code:$PYTHONPATH

api/code/README.md

@@ -0,0 +1,28 @@
+# powerplant-coding-challenge - Candidate: Alberto Morales Morales
+
+
+## Set-up
+
+Set the repo base path
+REPO_BASE_PATH=...
+export REPO_BASE_PATH
+
+### with conda
+
+conda create -n ppcch python=3.10
+conda activate ppcch
+
+### without conda
+
+### in both cases
+cd $REPO_BASE_PATH/api/code
+pip install -r requirements.txt
+
+## How to run
+cd $REPO_BASE_PATH/api/code
+python3 run_server.py
+
+### with Docker
+cd $REPO_BASE_PATH/api
+docker build -t powerplanner . 
+docker run -p 8888:8888 docker.io/library/powerplanner

api/code/builders/__init__.py

@@ -0,0 +1,3 @@
+# -------------------------------------------------------------------------
+# Copyright (c) Engie. All rights reserved.
+# --------------------------------------------------------------------------

api/code/builders/lp_problem_builder.py

@@ -0,0 +1,25 @@
+import logging
+from abc import ABC, abstractmethod
+from typing import Tuple
+
+from config.constants import BS_LOGGER
+from models.linear_programming import LinearProgrammingProblem
+from models.energy_demand import EnergyDemand
+
+
+class LinearProgrammingProblemBuilder(ABC):
+    def __init__(self):
+        self._logger = logging.getLogger(BS_LOGGER)
+
+    @abstractmethod
+    def from_energy_demand(self, demand: EnergyDemand) -> Tuple[list[str], LinearProgrammingProblem]:
+        """Gets a linear programming problem, given an energy demand.
+
+        Parameters:
+        demand (EnergyDemand): energy demand.
+
+        Returns:
+        The linear programming problem corresponding to the demand
+
+        """            
+        raise NotImplementedError("Subclasses should implement this!")

api/code/builders/lp_problem_builder_impl.py

@@ -0,0 +1,54 @@
+import logging
+from abc import ABC, abstractmethod
+from typing import Tuple
+
+from config.constants import BS_LOGGER
+from models.linear_programming import LinearProgrammingProblem
+from models.energy_demand import EnergyDemand
+from exceptions.exceptions import BuilderException
+
+class LinearProgrammingProblemBuilderImpl(ABC):
+    def __init__(self):
+        self._logger = logging.getLogger(BS_LOGGER)
+
+    def from_energy_demand(self, demand: EnergyDemand) -> Tuple[list[str], LinearProgrammingProblem]:
+
+        try:
+            load = demand.load
+            fuels = demand.fuels
+            powerplant_list = demand.powerplants
+
+            constraint_vector = [load]
+
+            fuels_dict = {
+                "gasfired": fuels.gas,
+                "turbojet": fuels.kerosine,
+                "windturbine": 0
+            }
+
+            powerplant_names = []
+            objective_function = []
+            constraint_matrix = []
+            constraint_matrix_first_row = []
+            bounds = []
+
+            for powerplant in powerplant_list:
+                powerplant_names.append(powerplant.name)
+                objective_function.append(fuels_dict[powerplant.type])
+                if (powerplant.type == "windturbine"):
+                    constraint_matrix_first_row.append(fuels.wind / 100)
+                else:
+                    constraint_matrix_first_row.append(powerplant.efficiency)
+                powerplant_bounds = (powerplant.pmin, powerplant.pmax)
+                bounds.append(powerplant_bounds)
+
+            constraint_matrix = [constraint_matrix_first_row]
+
+            lp_problem = LinearProgrammingProblem(objective_function=objective_function,
+                                                constraint_matrix = constraint_matrix,
+                                                constraint_vector = constraint_vector,
+                                                bounds=bounds)
+        except Exception as e:
+            raise BuilderException(arg=e)
+
+        return powerplant_names, lp_problem

api/code/builders/production_plan_builder.py

@@ -0,0 +1,25 @@
+import logging
+from abc import ABC, abstractmethod
+
+from config.constants import BS_LOGGER
+from models.linear_programming import ProblemResult
+from models.production_plan import ProductionPlanItem
+
+
+class ProductionPlanBuilder(ABC):
+    def __init__(self):
+        self._logger = logging.getLogger(BS_LOGGER)
+
+    @abstractmethod
+    def from_problem_result(self, powerplant_names: list[str], problem_result: ProblemResult) -> list[ProductionPlanItem]:
+        """Gets a production plan, given a linear programming problem result.
+
+        Parameters:
+        powerplant_names: array of powerplant names
+        problem_result (ProblemResult): linear programming problem result.
+
+        Returns:
+        The production plan corresponding to the linear programming problem result
+
+        """        
+        raise NotImplementedError("Subclasses should implement this!")

api/code/builders/production_plan_builder_impl.py

@@ -0,0 +1,24 @@
+import logging
+
+from config.constants import BS_LOGGER
+from planner.planner import Planner
+from solver.solver import Solver
+from models.energy_demand import EnergyDemand
+from models.production_plan import ProductionPlanItem
+from models.linear_programming import LinearProgrammingProblem, ProblemResult
+
+from builders.production_plan_builder import ProductionPlanBuilder
+
+
+class ProductionPlanBuilderImpl(ProductionPlanBuilder):
+    def __init__(self):
+        super().__init__()
+        self._logger = logging.getLogger(BS_LOGGER)
+
+    def from_problem_result(self, powerplant_names: list[str], problem_result: ProblemResult) -> list[ProductionPlanItem]:
+        result_matrix = problem_result.solution
+        production_plan = []
+        for n in range(len(result_matrix)):
+            production_plan_item = ProductionPlanItem(name=powerplant_names[n],p=result_matrix[n])
+            production_plan.append(production_plan_item)
+        return production_plan

api/code/config/__init__.py

@@ -0,0 +1,3 @@
+# -------------------------------------------------------------------------
+# Copyright (c) Engie. All rights reserved.
+# --------------------------------------------------------------------------

api/code/config/appconfig.py

@@ -0,0 +1,80 @@
+"""
+    This is the Configuration System for the Powerplan  API
+
+    All these Variables should be exported in the Container and need to be readable
+    by the Process.
+
+    If there is no needed configuration, the Container will End with ValueError.
+
+"""
+
+import logging
+import os
+from logging.config import dictConfig
+from config.constants import BS_LOGGER, BS_CORRELATION_LOGGER
+from config.traceability import CorrelationIdFilter
+
+LOG_LEVEL = os.getenv("LOG_LEVEL", "")
+BS_LOG_LEVEL = os.getenv("BS_LOG_LEVEL", "")
+
+if LOG_LEVEL is None or LOG_LEVEL == "":
+    LOG_LEVEL = "WARNING"
+
+if BS_LOG_LEVEL is None or BS_LOG_LEVEL == "":
+    BS_LOG_LEVEL = LOG_LEVEL
+
+dictConfig(
+    {
+        "version": 1,
+        "formatters": {
+            "data-format": {
+                # DATE | BS LOGGER NAME | LEVEL | FILENAME | LINE | FUNCTION | MESSAGE
+                "format": "%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(funcName)s - %(message)s",
+                "datefmt": "%Y%m%d%H%M%S",
+            },
+            "correlation-format": {
+                # DATE | BS LOGGER NAME | LEVEL | FILENAME | LINE | FUNCTION | CORRELATION ID | MESSAGE
+                "format": "%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(funcName)s - [%(correlation_id)s] - %(message)s",
+                "datefmt": "%Y%m%d%H%M%S",
+            },
+        },
+        "filters": {
+            "correlation_id_filter": {
+                "()": CorrelationIdFilter,
+            }
+        },
+        "handlers": {
+            "default": {
+                "formatter": "data-format",
+                "class": "logging.StreamHandler",
+                "stream": "ext://sys.stdout",  # Default is stderr
+            },
+            "correlation": {
+                "formatter": "correlation-format",
+                "class": "logging.StreamHandler",
+                "filters": ["correlation_id_filter"],
+                "stream": "ext://sys.stdout",  # Default is stderr
+            },
+        },
+        "root": {"level": LOG_LEVEL, "handlers": ["default", "correlation"]},
+        "loggers": {
+            BS_LOGGER: {"level": BS_LOG_LEVEL, "propagate": 0, "handlers": ["default"]},
+            BS_CORRELATION_LOGGER: {
+                "level": BS_LOG_LEVEL,
+                "propagate": 0,
+                "handlers": ["correlation"],
+            },
+        },
+    }
+)
+
+logger = logging.getLogger(BS_LOGGER)
+logger.debug("After logger initialization")
+
+# Only for Log
+APP_NAME = "powerplant-code-challenge"
+APP_VERSION = os.getenv("APP_VERSION", "1.1.0")
+SWAGGER_URL_PREFIX = os.getenv("SWAGGER_URL_PREFIX", "")
+
+# Feature List
+FEATURE_ENABLE_SWAGGER_UI = os.getenv("FEATURE_ENABLE_SWAGGER_UI", "NO")

api/code/config/constants.py

@@ -0,0 +1,2 @@
+BS_LOGGER = "powerplan_coding_challenge_logger"
+BS_CORRELATION_LOGGER = "powerplan_coding_challenge_correlation_logger"

api/code/config/traceability.py

@@ -0,0 +1,39 @@
+import logging
+import uuid
+import contextvars
+
+# Context variable to store the correlation ID
+correlation_id_ctx_var = contextvars.ContextVar("correlation_id", default=None)
+
+
+def get_correlation_id():
+    """
+    Returns the current correlation ID.
+    """
+    return correlation_id_ctx_var.get()
+
+
+def set_correlation_id(correlation_id):
+    """
+    Sets the correlation ID.
+    """
+    correlation_id_ctx_var.set(correlation_id)
+
+
+def generate_correlation_id():
+    """
+    Generates a new correlation ID and sets it in the context variable.
+    """
+    correlation_id = str(uuid.uuid4())
+    set_correlation_id(correlation_id)
+    return correlation_id
+
+
+class CorrelationIdFilter(logging.Filter):
+    """
+    Logging filter to add correlation ID to log records.
+    """
+
+    def filter(self, record):
+        record.correlation_id = get_correlation_id() or "N/A"
+        return True

api/code/controllers/__init__.py

@@ -0,0 +1,3 @@
+# -------------------------------------------------------------------------
+# Copyright (c) Engie. All rights reserved.
+# --------------------------------------------------------------------------

api/code/controllers/common_endpoints.py

@@ -0,0 +1,46 @@
+"""
+    This module defines common utility API endpoints for the Powerplant Coding Challenge API.
+    It includes functions to check the application's version, root, and health status.
+
+    Endpoints:
+    - GET /: Root endpoint returning appname and version
+    - GET /version: Retrieve the current version of the application
+    - GET /healthz: Perform a health check of the application
+
+"""
+
+from datetime import datetime
+
+import logging
+
+from fastapi import HTTPException
+
+from config import appconfig
+from config.constants import BS_LOGGER
+from models.responses import HealthResponse
+from fastapi import APIRouter
+
+commons_router = APIRouter()
+
+
+@commons_router.get("/", include_in_schema=False)
+def get_root():
+    logging.getLogger(BS_LOGGER).info("Request received in /")
+    return {"appname": appconfig.APP_NAME, "version": appconfig.APP_VERSION}
+
+
+@commons_router.get("/version", include_in_schema=False)
+def get_version():
+    logging.getLogger(BS_LOGGER).info("Request received in /version")
+    return appconfig.APP_VERSION
+
+
+@commons_router.get("/healthz", response_model=HealthResponse, include_in_schema=False)
+def check_health():
+    logging.getLogger(BS_LOGGER).info("Request received in /healthz")
+    try:
+        current_time = datetime.now().strftime("%#m/%#d/%Y %#I:%M:%S %p")
+        # TODO run some heath checks
+        return HealthResponse(status="healthy", current_time=current_time)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"System Error: {e}")