import subprocess from fastapi import FastAPI, Response, status, Query from typing import Annotated from mpai_cae_arp.types.schema import Info from mpai_cae_arp.files import FileType, File info = File("config/server.yaml", FileType.YAML).get_content() app = FastAPI(**info) @app.get("/") async def list_all_entrypoints() -> list[str]: """ Get the list of available main routes. """ return ["restore", "description", "docs"] @app.get("/description") async def get_server_info() -> Info: """ Retrieve all the informations about the software running on the server. """ return info @app.get("/restore", status_code=200) async def restore( response: Response, equalization_w: Annotated[str, Query( description="Actual equalization of the audio tape applied during the recording.", example="CCIR" )], equalization_r: Annotated[str, Query( description="Actual equalization of the audio tape applied during the playback.", example="NAB" )], speed_w: Annotated[str, Query( description="Actual speed of the audio tape applied during the recording.", example="7.5" )], speed_r: Annotated[str, Query( description="Actual speed of the audio tape applied during the playback.", example="15" )], files_name: Annotated[str, Query( description="Name of the audio/video file to be analyzed without the extension.", example="filename" )], 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 ) -> dict: process = [ "python", "src/tapeAudioRestoration.py", "--files_name", files_name, "--equalization-w", equalization_w, "--equalization-r", equalization_r, "--speed-w", speed_w, "--speed-r", speed_r ] if working_path is not None: process.extend(["--working_path", working_path]) else: config = File("config/args.yaml", FileType.YAML).get_content() process.extend(["--working_path", config["WORKING_PATH"]]) cprocess = subprocess.run(process, capture_output=True) if cprocess.returncode == 0: editing_list = File("EditingList.json", FileType.JSON).get_content() return editing_list else: response.status_code = status.HTTP_412_PRECONDITION_FAILED return {"error": { "returncode": cprocess.returncode, "stdout": cprocess.stdout.decode("utf-8"), "stderr": cprocess.stderr.decode("utf-8"), }}