Commit 76842786 authored by Matteo's avatar Matteo
Browse files

refactor code

parent fd80e83a
from fastapi import FastAPI, Response, status from fastapi import FastAPI, Response, status, Path, Query
from typing import Annotated
import server.functions as functions from mpai_cae_arp.files import get_file_content
from mpai_cae_arp.types.irregularity import IrregularityFile
from mpai_cae_arp.types.schema import Info
import server.utils as utils
info = functions.get_file_content('server/documentation.yaml', 'yaml') info = get_file_content('server/documentation.yaml', 'yaml')
app = FastAPI(**info) app = FastAPI(**info)
@app.get("/") @app.get("/")
def index(): async def list_endpoints() -> list[str]:
return {"endpoints": ["irregularityFile1", "irregularityFile2", "description", "docs"]} return ["irregularityFile/{id}", "description", "docs"]
@app.get("/description") @app.get("/description")
async def get_description(): async def get_server_info() -> Info:
return info return info
@app.get("/irregularityFile/{id}", status_code=200) @app.get("/irregularityFile/{id}", status_code=200)
def get_irregularity_file(files_name: str, id: int, response: Response, working_path: str | None = None): async def get_irregularity_file(
response: Response,
id: Annotated[int, Path(
gt=0, lt=3,
description="Id of the irregularity file to be returned."
)],
files_name: Annotated[str, Query(
description="Name of the video file to be analyzed without the extension.",
example="video"
)],
working_path: Annotated[str, Query(
description="Path to the directory containing the preservation files.",
examples={
"Absolute path": {"value": "/home/user/preservation"},
"Relative path": {"value": "preservation"}
}
)] = None
) -> IrregularityFile:
required_file = f'VideoAnalyser_IrregularityFileOutput{id}.json' required_file = f'VideoAnalyser_IrregularityFileOutput{id}.json'
try: try:
return functions.get_file_content(required_file, 'json') return get_file_content(required_file, 'json')
except: except:
process = functions.analyze(files_name, working_path) process = utils.analyze(files_name, working_path)
if process.returncode == 0: if process.returncode == 0:
return functions.get_file_content(required_file, 'json') return get_file_content(required_file, 'json')
else: else:
response.status_code = status.HTTP_412_PRECONDITION_FAILED response.status_code = status.HTTP_412_PRECONDITION_FAILED
......
anyio==3.6.2 anyio==3.6.2
appdirs==1.4.4
audioread==3.0.0
certifi==2022.12.7 certifi==2022.12.7
cffi==1.15.1
charset-normalizer==3.1.0
click==8.1.3 click==8.1.3
decorator==5.1.1
dnspython==2.3.0 dnspython==2.3.0
email-validator==1.3.1 email-validator==1.3.1
fastapi==0.95.0 fastapi==0.95.0
...@@ -11,17 +16,35 @@ httpx==0.23.3 ...@@ -11,17 +16,35 @@ httpx==0.23.3
idna==3.4 idna==3.4
itsdangerous==2.1.2 itsdangerous==2.1.2
Jinja2==3.1.2 Jinja2==3.1.2
joblib==1.2.0
lazy_loader==0.2
librosa==0.10.0.post2
llvmlite==0.39.1
MarkupSafe==2.1.2 MarkupSafe==2.1.2
mpai-cae-arp==0.1.0
msgpack==1.0.5
numba==0.56.4
numpy==1.23.3
orjson==3.8.9 orjson==3.8.9
packaging==23.0
pooch==1.6.0
pycparser==2.21
pydantic==1.10.7 pydantic==1.10.7
python-dotenv==1.0.0 python-dotenv==1.0.0
python-multipart==0.0.6 python-multipart==0.0.6
PyYAML==6.0 PyYAML==6.0
requests==2.28.2
rfc3986==1.5.0 rfc3986==1.5.0
scikit-learn==1.2.2
scipy==1.10.1
sniffio==1.3.0 sniffio==1.3.0
soundfile==0.12.1
soxr==0.3.4
starlette==0.26.1 starlette==0.26.1
threadpoolctl==3.1.0
typing_extensions==4.5.0 typing_extensions==4.5.0
ujson==5.7.0 ujson==5.7.0
urllib3==1.26.15
uvicorn==0.21.1 uvicorn==0.21.1
uvloop==0.17.0 uvloop==0.17.0
watchfiles==0.19.0 watchfiles==0.19.0
......
import os import os
import subprocess import subprocess
import json import json
import yaml from mpai_cae_arp.files import get_file_content
def analyze(files_name: str, working_path: str | None = None) -> subprocess.CompletedProcess: def analyze(files_name: str, working_path: str | None = None) -> subprocess.CompletedProcess:
...@@ -18,15 +18,3 @@ def analyze(files_name: str, working_path: str | None = None) -> subprocess.Comp ...@@ -18,15 +18,3 @@ def analyze(files_name: str, working_path: str | None = None) -> subprocess.Comp
process = subprocess.run([executable], capture_output=True) process = subprocess.run([executable], capture_output=True)
return process return process
def get_file_content(file_name: str, format: str) -> dict:
with open(file_name) as fd:
if format == "yaml":
content = yaml.safe_load(fd)
return content
elif format == "json":
content = json.load(fd)
return content
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment