Added energy model

ensysmod/api/api.py

@@ -2,7 +2,7 @@
 
 from .endpoints import users, authentication, energy_sources, datasets, energy_commodities, energy_sinks, \
     energy_storages, energy_transmissions, energy_conversions, regions, ts_capacity_max, ts_operation_rate_fix, \
-    ts_operation_rate_max
+    ts_operation_rate_max, energy_models
 
 api_router = APIRouter()
 api_router.include_router(authentication.router, prefix="/auth", tags=["Authentication"])
@@ -15,6 +15,7 @@
 api_router.include_router(energy_sources.router, prefix="/sources", tags=["Energy Sources"])
 api_router.include_router(energy_storages.router, prefix="/storages", tags=["Energy Storages"])
 api_router.include_router(energy_transmissions.router, prefix="/transmissions", tags=["Energy Transmissions"])
+api_router.include_router(energy_models.router, prefix="/models", tags=["Energy Models"])
 
 api_router.include_router(ts_capacity_max.router, prefix="/max-capacities", tags=["TS Capacities Max"])
 api_router.include_router(ts_operation_rate_fix.router, prefix="/fix-operation-rates", tags=["TS Operation Rates Fix"])

ensysmod/api/endpoints/energy_models.py

@@ -0,0 +1,83 @@
+from typing import List, Union
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.orm import Session
+
+from ensysmod import schemas, model, crud
+from ensysmod.api import deps
+
+router = APIRouter()
+
+
+@router.get("/", response_model=List[schemas.EnergyModel])
+def all_models(db: Session = Depends(deps.get_db),
+               current: model.User = Depends(deps.get_current_user),
+               skip: int = 0,
+               limit: int = 100,
+               dataset: Union[None, int] = None) -> List[schemas.EnergyModel]:
+    """
+    Retrieve all energy models.
+    """
+    if dataset is None:
+        return crud.energy_model.get_multi(db, skip=skip, limit=limit)
+    else:
+        return crud.energy_model.get_multi_by_dataset(db, dataset_id=dataset, skip=skip, limit=limit)
+
+
+@router.get("/{model_id}", response_model=schemas.EnergyModel)
+def get_model(model_id: int,
+              db: Session = Depends(deps.get_db),
+              current: model.User = Depends(deps.get_current_user)):
+    """
+    Retrieve a energy model.
+    """
+    # TODO Check if user has permission for dataset and model
+    return crud.energy_model.get(db, id=model_id)
+
+
+@router.post("/", response_model=schemas.EnergyModel,
+             responses={409: {"description": "EnergyModel with same name already exists."}})
+def create_model(request: schemas.EnergyModelCreate,
+                 db: Session = Depends(deps.get_db),
+                 current: model.User = Depends(deps.get_current_user)):
+    """
+    Create a new energy model.
+    """
+    dataset = crud.dataset.get(db=db, id=request.ref_dataset)
+    if dataset is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Dataset {request.ref_dataset} not found!")
+
+    # TODO Check if user has permission for dataset
+
+    existing = crud.energy_model.get_by_dataset_and_name(db=db, dataset_id=request.ref_dataset, name=request.name)
+    if existing is not None:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT,
+                            detail=f"EnergyModel {request.name} already for dataset {request.ref_dataset} exists!")
+
+    return crud.energy_model.create(db=db, obj_in=request)
+
+
+@router.put("/{model_id}", response_model=schemas.EnergyModel)
+def update_model(model_id: int,
+                 request: schemas.EnergyModelUpdate,
+                 db: Session = Depends(deps.get_db),
+                 current: model.User = Depends(deps.get_current_user)):
+    """
+    Update a energy model.
+    """
+    # TODO Check if user has permission for model
+    model = crud.energy_model.get(db=db, id=model_id)
+    if model is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"EnergyModel {model_id} not found!")
+    return crud.energy_model.update(db=db, db_obj=model, obj_in=request)
+
+
+@router.delete("/{model_id}", response_model=schemas.EnergyModel)
+def remove_model(model_id: int,
+                 db: Session = Depends(deps.get_db),
+                 current: model.User = Depends(deps.get_current_user)):
+    """
+    Delete a energy model.
+    """
+    # TODO Check if user has permission for dataset
+    return crud.energy_model.remove(db=db, id=model_id)

ensysmod/crud/__init__.py

@@ -14,3 +14,4 @@
 from .ts_operation_rate_max import operation_rate_max
 from .ts_operation_rate_fix import operation_rate_fix
 from .user import user
+from .energy_model import energy_model

ensysmod/crud/energy_model.py

@@ -0,0 +1,14 @@
+from ensysmod.crud.base_depends_dataset import CRUDBaseDependsDataset
+from ensysmod.model import EnergyModel
+from ensysmod.schemas import EnergyModelCreate, EnergyModelUpdate
+
+
+# noinspection PyMethodMayBeStatic,PyArgumentList
+class CRUDEnergyModel(CRUDBaseDependsDataset[EnergyModel, EnergyModelCreate, EnergyModelUpdate]):
+    """
+    CRUD operations for EnergyModel
+    """
+    pass
+
+
+energy_model = CRUDEnergyModel(EnergyModel)

ensysmod/model/__init__.py

@@ -15,3 +15,4 @@
 from .ts_operation_rate_fix import OperationRateFix
 from .ts_operation_rate_max import OperationRateMax
 from .user import User
+from .energy_model import EnergyModel

ensysmod/model/energy_model.py

@@ -0,0 +1,20 @@
+from sqlalchemy import Column, Integer, String, DECIMAL, ForeignKey, UniqueConstraint
+from sqlalchemy.orm import relationship
+
+from ensysmod.database.base_class import Base
+
+
+class EnergyModel(Base):
+    id = Column(Integer, primary_key=True, index=True)
+    ref_dataset = Column(Integer, ForeignKey("dataset.id"), index=True, nullable=False)
+    name = Column(String, index=True, nullable=False)
+    description = Column(String, nullable=True)
+    yearly_co2_limit = Column(DECIMAL, nullable=True)
+
+    # relationships
+    dataset = relationship("Dataset")
+
+    # table constraints
+    __table_args__ = (
+        UniqueConstraint("ref_dataset", "name", name="_commodity_name_dataset_uc"),
+    )

ensysmod/schemas/__init__.py

@@ -15,3 +15,4 @@
 from .ts_operation_rate_fix import OperationRateFix, OperationRateFixCreate, OperationRateFixUpdate
 from .ts_operation_rate_max import OperationRateMax, OperationRateMaxCreate, OperationRateMaxUpdate
 from .user import User, UserCreate, UserInDB, UserUpdate
+from .energy_model import EnergyModel, EnergyModelCreate, EnergyModelUpdate

ensysmod/schemas/energy_model.py

@@ -0,0 +1,39 @@
+from typing import Optional
+
+from pydantic import BaseModel
+
+from ensysmod.schemas import Dataset
+
+
+class EnergyModelBase(BaseModel):
+    """
+    Shared properties for an energy model. Used as a base class for all schemas.
+    """
+    name: str
+    yearly_co2_limit: Optional[float] = None
+    description: Optional[str] = None
+
+
+class EnergyModelCreate(EnergyModelBase):
+    """
+    Properties to receive via API on creation of an energy model.
+    """
+    ref_dataset: int
+
+
+class EnergyModelUpdate(EnergyModelBase):
+    """
+    Properties to receive via API on update of an energy model.
+    """
+    name: Optional[str] = None
+
+
+class EnergyModel(EnergyModelBase):
+    """
+    Properties to return via API for an energy model.
+    """
+    id: int
+    dataset: Dataset
+
+    class Config:
+        orm_mode = True

tests/api/test_energy_models.py

@@ -0,0 +1,46 @@
+from typing import Dict
+
+from fastapi import status
+from fastapi.encoders import jsonable_encoder
+from fastapi.testclient import TestClient
+from sqlalchemy.orm import Session
+
+from ensysmod.schemas import EnergyModelCreate
+from tests.utils import data_generator as data_gen
+
+
+def test_create_energy_model(client: TestClient, normal_user_headers: Dict[str, str], db: Session):
+    """
+    Test creating a energy model.
+    """
+    create_request = data_gen.random_energy_model_create(db)
+    response = client.post("/models/", headers=normal_user_headers, data=create_request.json())
+    assert response.status_code == status.HTTP_200_OK
+
+    created_model = response.json()
+    assert created_model["name"] == create_request.name
+    assert created_model["dataset"]["id"] == create_request.ref_dataset
+    assert created_model["description"] == create_request.description
+    assert created_model["yearly_co2_limit"] == create_request.yearly_co2_limit
+
+
+def test_create_existing_energy_model(client: TestClient, normal_user_headers: Dict[str, str], db: Session):
+    """
+    Test creating a existing energy model.
+    """
+    existing_model = data_gen.random_existing_energy_model(db)
+    create_request = EnergyModelCreate(**jsonable_encoder(existing_model))
+    response = client.post("/models/", headers=normal_user_headers, data=create_request.json())
+    assert response.status_code == status.HTTP_409_CONFLICT
+
+
+def test_create_energy_model_unknown_dataset(client: TestClient, normal_user_headers: Dict[str, str], db: Session):
+    """
+    Test creating a energy model.
+    """
+    create_request = data_gen.random_energy_model_create(db)
+    create_request.ref_dataset = 0  # ungültige Anfrage
+    response = client.post("/models/", headers=normal_user_headers, data=create_request.json())
+    assert response.status_code == status.HTTP_404_NOT_FOUND
+
+# TODO Add more test cases

tests/database/test_tables.py

@@ -48,3 +48,7 @@ def test_table_operation_rate_max(db: Session):
 
 def test_table_region(db: Session):
     assert check_table_exists(db.bind.raw_connection().cursor(), 'region')
+
+
+def test_table_energy_model(db: Session):
+    assert check_table_exists(db.bind.raw_connection().cursor(), 'energy_model')

tests/utils/data_generator/__init__.py

@@ -10,3 +10,4 @@
 from .energy_transmissions import random_existing_energy_transmission, fixed_existing_energy_transmission, \
     random_energy_transmission_create
 from .regions import random_existing_region, fixed_existing_region, random_region_create
+from .energy_models import random_energy_model_create, random_existing_energy_model

tests/utils/data_generator/energy_models.py

@@ -0,0 +1,52 @@
+from sqlalchemy.orm import Session
+
+from ensysmod import crud
+from ensysmod.model import EnergyModel
+from ensysmod.schemas import EnergyModelCreate
+from tests.utils.data_generator.datasets import fixed_existing_dataset, random_existing_dataset
+from tests.utils.utils import random_lower_string
+
+
+def random_energy_model_create(db: Session) -> EnergyModelCreate:
+    """
+    Generate a random energy model create request.
+    """
+    dataset = random_existing_dataset(db)
+    return EnergyModelCreate(name=f"EnergyModel-{dataset.id}-" + random_lower_string(),
+                             ref_dataset=dataset.id,
+                             description="EnergyModel description",
+                             yearly_co2_limit=10.7
+                             )
+
+
+def random_existing_energy_model(db: Session) -> EnergyModel:
+    """
+    Generate a random existing energy model.
+    """
+    create_request = random_energy_model_create(db)
+    return crud.energy_model.create(db=db, obj_in=create_request)
+
+
+def fixed_energy_model_create(db: Session) -> EnergyModelCreate:
+    """
+    Generate a fixed energy model create request.
+    Will always return the same energy model.
+    """
+    dataset = fixed_existing_dataset(db)
+    return EnergyModelCreate(name=f"EnergyModel-{dataset.id}-Fixed",
+                             ref_dataset=dataset.id,
+                             description="EnergyModel description",
+                             yearly_co2_limit=10)
+
+
+def fixed_existing_energy_model(db: Session) -> EnergyModel:
+    """
+    Generate a fixed existing energy model.
+    Will always return the same energy model.
+    """
+    create_request = fixed_energy_model_create(db)
+    model = crud.energy_model.get_by_dataset_and_name(db=db, dataset_id=create_request.ref_dataset,
+                                                      name=create_request.name)
+    if model is None:
+        return crud.energy_model.create(db=db, obj_in=create_request)
+    return model