Skip to content

Commit

Permalink
refactors controllers/environemnt
Browse files Browse the repository at this point in the history
  • Loading branch information
@GabrielBarberini
GabrielBarberini committed Apr 14, 2024
1 parent 243f39e commit 2863c3d
Showing 1 changed file with 153 additions and 81 deletions.
234 changes: 153 additions & 81 deletions lib/controllers/environment.py
View file
Original file line number Diff line number Diff line change
@@ -1,9 +1,12 @@
from typing import Union
from datetime import datetime

import jsonpickle
from rocketpy.environment.environment import Environment
import logging
from rocketpy.environment.environment import Environment as RocketPyEnvironment
from fastapi import HTTPException, status

from lib.controllers import parse_error
from lib.models.environment import Env
from lib.repositories.environment import EnvRepository
from lib.views.environment import (
Expand All @@ -22,14 +25,28 @@ class EnvController:
Controller for the Environment model.
Init Attributes:
env (models.Env): Environment model object.
env: models.Env
Enables:
- Create a rocketpy.Environment object from an Env model object.
- Simulation of RocketPyEnvironment from models.Env
- CRUD operations over modeols.Env on the database
"""

def __init__(self, env: Env):
rocketpy_env = Environment(
self._env = env

async def get_env(self) -> Env:
return self._env

@staticmethod
async def get_rocketpy_env(env: Env) -> RocketPyEnvironment:
"""
Get the rocketpy env object.
Returns:
RocketPyEnvironment
"""
rocketpy_env = RocketPyEnvironment(
latitude=env.latitude,
longitude=env.longitude,
elevation=env.elevation,
Expand All @@ -38,165 +55,220 @@ def __init__(self, env: Env):
rocketpy_env.set_atmospheric_model(
type=env.atmospheric_model_type, file=env.atmospheric_model_file
)
self.rocketpy_env = rocketpy_env
self.env = env
return rocketpy_env

async def create_env(self) -> "Union[EnvCreated, HTTPException]":
"""
Create a env in the database.
Returns:
EnvCreated: Environment id.
views.EnvCreated
"""
env = EnvRepository(environment=self.env)
successfully_created_env = await env.create_env()
if not successfully_created_env:
env = self.get_env()
env_repo = EnvRepository(environment=env)
try:
await env_repo.create_env()
except Exception as e:
exc_str = parse_error(e)
logging.error(f"controllers/environment.create_env: {exc_str}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to create environment",
detail=f"Failed to create environment: {e}",
) from e
else:
return EnvCreated(env_id=env_repo.get_env_id())
finally:
logging.info(
f"[{datetime.now()}] Call to controllers/environment.create_env; params: Env {env.hash()}"
)

return EnvCreated(env_id=str(env.env_id))

@staticmethod
async def get_env(env_id: int) -> "Union[Env, HTTPException]":
async def get_env_by_id(env_id: int) -> "Union[Env, HTTPException]":
"""
Get a env from the database.
Args:
env_id (int): Environment id.
env_id: int
Returns:
env model object
models.Env
Raises:
HTTP 404 Not Found: If the env is not found in the database.
"""
successfully_read_env = await EnvRepository(env_id=env_id).get_env()
if not successfully_read_env:
env_repo = EnvRepository(env_id=env_id)
try:
read_env = await env_repo.get_env()
except Exception as e:
exc_str = parse_error(e)
logging.error(f"controllers/environment.get_env_by_id: {exc_str}")
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="Environment not found"
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Failed to read environment: {e}",
) from e
else:
if read_env:
return read_env
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Environment not found",
)
finally:
logging.info(
f"[{datetime.now()}] Call to controllers/environment.get_env_by_id; params: EnvID {env_id}"
)

return successfully_read_env

@staticmethod
async def get_rocketpy_env(env_id: int) -> "Union[EnvPickle, HTTPException]":
@classmethod
async def get_rocketpy_env_as_jsonpickle(
cls,
env_id: int,
) -> "Union[EnvPickle, HTTPException]":
"""
Get a rocketpy env object encoded as jsonpickle string from the database.
Get rocketpy.Environmnet as jsonpickle string.
Args:
env_id (int): env id.
env_id: int
Returns:
str: jsonpickle string of the rocketpy env.
views.EnvPickle
Raises:
HTTP 404 Not Found: If the env is not found in the database.
"""
successfully_read_env = await EnvRepository(env_id=env_id).get_env()
if not successfully_read_env:
env_repo = EnvRepository(env_id=env_id)
try:
read_env = await env_repo.get_env()
except Exception as e:
exc_str = parse_error(e)
logging.error(
f"controllers/environment.get_rocketpy_env_as_jsonpickle: {exc_str}"
)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Failed to read environment: {e}",
) from e
else:
if read_env:
rocketpy_env = cls.get_rocketpy_env(read_env)
return EnvPickle(
jsonpickle_rocketpy_env=jsonpickle.encode(rocketpy_env)
)
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="Environment not found"
status_code=status.HTTP_404_NOT_FOUND,
detail="Environment not found",
)
finally:
logging.info(
f"[{datetime.now()}] Call to controllers/environment.get_rocketpy_env_as_jsonpickle; params: EnvID {env_id}"
)

successfully_read_rocketpy_env = EnvController(
successfully_read_env
).rocketpy_env

return EnvPickle(
jsonpickle_rocketpy_env=jsonpickle.encode(successfully_read_rocketpy_env)
)

async def update_env(self, env_id: int) -> "Union[EnvUpdated, HTTPException]":
async def update_env(
self, env_id: int
) -> "Union[EnvUpdated, HTTPException]":
"""
Update a env in the database.
Args:
env_id (int): env id.
env_id: int
Returns:
EnvUpdated: env id and message.
views.EnvUpdated
Raises:
HTTP 404 Not Found: If the env is not found in the database.
"""
successfully_read_env = await EnvRepository(env_id=env_id).get_env()
if not successfully_read_env:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="Environment not found"
)

successfully_updated_env = await EnvRepository(
environment=self.env, env_id=env_id
).update_env()
if not successfully_updated_env:
env = self.get_env()
env_repo = EnvRepository(environment=env, env_id=env_id)
try:
read_env = await env_repo.get_env()
if read_env:
await env_repo.update_env()
else:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Environment not found",
)
except Exception as e:
exc_str = parse_error(e)
logging.error(f"controllers/environment.update_env: {exc_str}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to update environment",
detail=f"Failed to update environment: {e}",
) from e
else:
return EnvUpdated(new_env_id=env_repo.get_env_id())
finally:
logging.info(
f"[{datetime.now()}] Call to controllers/environment.update_env; params: EnvID {env_id}, Env {env.hash()}"
)

return EnvUpdated(new_env_id=str(successfully_updated_env))

@staticmethod
async def delete_env(env_id: int) -> "Union[EnvDeleted, HTTPException]":
async def delete_env(env_id: str) -> "Union[EnvDeleted, HTTPException]":
"""
Delete a env from the database.
Args:
env_id (int): Environment id.
env_id: int
Returns:
EnvDeleted: Environment id and message.
views.EnvDeleted
Raises:
HTTP 404 Not Found: If the env is not found in the database.
"""
successfully_read_env = await EnvRepository(env_id=env_id).get_env()
if not successfully_read_env:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="Environment not found"
)

successfully_deleted_env = await EnvRepository(env_id=env_id).delete_env()
if not successfully_deleted_env:
env_repo = EnvRepository(env_id=env_id)
try:
await env_repo.delete_env()
except Exception as e:
exc_str = parse_error(e)
logging.error(f"controllers/environment.delete_env: {exc_str}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to delete environment",
detail=f"Failed to delete environment: {e}",
) from e
else:
return EnvDeleted(deleted_env_id=env_id)
finally:
logging.info(
f"[{datetime.now()}] Call to controllers/environment.delete_env; params: EnvID {env_id}"
)

return EnvDeleted(deleted_env_id=str(env_id))

@staticmethod
async def simulate(env_id: int) -> "Union[EnvSummary, HTTPException]":
@classmethod
async def simulate(cls, env_id: int) -> "Union[EnvSummary, HTTPException]":
"""
Simulate a rocket environment.
Args:
env_id (int): Env id.
env_id: int.
Returns:
Env summary view.
views.EnvSummary
Raises:
HTTP 404 Not Found: If the env does not exist in the database.
"""
successfully_read_env = await EnvRepository(env_id=env_id).get_env()
if not successfully_read_env:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="Environment not found"
)

read_env = await cls.get_env_by_id(env_id)
try:
env = EnvController(successfully_read_env).rocketpy_env
env_simulation_numbers = EnvData.parse_obj(env.all_info_returned())
env_simulation_plots = EnvPlots.parse_obj(env.all_plot_info_returned())
rocketpy_env = cls.get_rocketpy_env(read_env)
env_simulation_numbers = EnvData.parse_obj(
rocketpy_env.all_info_returned()
)
env_simulation_plots = EnvPlots.parse_obj(
rocketpy_env.all_plot_info_returned()
)
env_summary = EnvSummary(
env_data=env_simulation_numbers, env_plots=env_simulation_plots
)
return env_summary
except Exception as e:
exc_str = parse_error(e)
logging.error(f"controllers/environment.simulate: {exc_str}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Failed to simulate environment: {e}",
) from e
else:
return env_summary
finally:
logging.info(
f"[{datetime.now()}] Call to controllers/environment.simulate; params: EnvID {env_id}"
)

0 comments on commit 2863c3d

Please sign in to comment.