From 0eca324e388c790bcf7fafe5831efc9aad7baa3d Mon Sep 17 00:00:00 2001 From: onyx-and-iris Date: Sun, 5 Apr 2026 00:22:50 +0100 Subject: [PATCH] add docstrings add documentation tags add some error handling --- src/vmr_http/app.py | 33 ++++- src/vmr_http/dependencies.py | 3 + src/vmr_http/models/bus.py | 14 ++ src/vmr_http/models/strip.py | 54 ++++++++ src/vmr_http/web/__init__.py | 0 src/vmr_http/web/bus.py | 54 +++++++- src/vmr_http/web/busmode.py | 45 ++++++ src/vmr_http/web/strip.py | 223 +++++++++++++++++++++++++++++- src/vmr_http/web/stripcomp.py | 37 +++++ src/vmr_http/web/stripdenoiser.py | 44 ++++++ src/vmr_http/web/stripgate.py | 37 +++++ 11 files changed, 535 insertions(+), 9 deletions(-) create mode 100644 src/vmr_http/models/bus.py create mode 100644 src/vmr_http/models/strip.py create mode 100644 src/vmr_http/web/__init__.py create mode 100644 src/vmr_http/web/busmode.py create mode 100644 src/vmr_http/web/stripcomp.py create mode 100644 src/vmr_http/web/stripdenoiser.py create mode 100644 src/vmr_http/web/stripgate.py diff --git a/src/vmr_http/app.py b/src/vmr_http/app.py index 75c6143..db7352c 100644 --- a/src/vmr_http/app.py +++ b/src/vmr_http/app.py @@ -1,19 +1,42 @@ +"""entry point for the FastAPI application.""" + from contextlib import asynccontextmanager import voicemeeterlib -from fastapi import FastAPI +from fastapi import Depends, FastAPI, HTTPException +from voicemeeterlib.error import CAPIError +from .dependencies import get_voicemeeter_client from .web import bus, strip @asynccontextmanager async def lifespan(app): - app.state.voicemeeter = voicemeeterlib.connect("potato", sync=True) + """Lifespan function to initialize and clean up the Voicemeeter client.""" + app.state.voicemeeter = voicemeeterlib.api('potato', sync=True) app.state.voicemeeter.login() yield app.state.voicemeeter.logout() -app = FastAPI(lifespan=lifespan) -app.include_router(strip.router, prefix="/strip") -app.include_router(bus.router, prefix="/bus") +app = FastAPI( + lifespan=lifespan, + description='A REST API for controlling Voicemeeter.', + openapi_tags=[ + {'name': 'strip', 'description': 'Endpoints for controlling strip parameters.'}, + {'name': 'bus', 'description': 'Endpoints for controlling bus parameters.'}, + ], +) +app.include_router(strip.router, prefix='/strip') +app.include_router(bus.router, prefix='/bus') + + +@app.get('/health') +def health_check(voicemeeter=Depends(get_voicemeeter_client)): + """Health check endpoint to verify the service is running.""" + try: + version = voicemeeter.version # Check if we can communicate with Voicemeeter + type_ = voicemeeter.type + except CAPIError as e: + raise HTTPException(status_code=503, detail=f'Voicemeeter API error: {str(e)}') + return {'status': 'ok', 'service': 'vmr-http', 'version': version, 'type': type_} diff --git a/src/vmr_http/dependencies.py b/src/vmr_http/dependencies.py index 8071d9b..3fb2820 100644 --- a/src/vmr_http/dependencies.py +++ b/src/vmr_http/dependencies.py @@ -1,5 +1,8 @@ +"""module containing dependencies for the API endpoints.""" + from fastapi import Request def get_voicemeeter_client(request: Request): + """Dependency to get the Voicemeeter client from the application state.""" return request.app.state.voicemeeter diff --git a/src/vmr_http/models/bus.py b/src/vmr_http/models/bus.py new file mode 100644 index 0000000..e1ce37f --- /dev/null +++ b/src/vmr_http/models/bus.py @@ -0,0 +1,14 @@ +"""Models for the parameters of a bus.""" + +from typing import Optional + +from pydantic import BaseModel + + +class BusParams(BaseModel): + """Parameters for a single bus.""" + + gain: Optional[float] = None + mute: Optional[bool] = None + mono: Optional[int] = None + eq: Optional[bool] = None diff --git a/src/vmr_http/models/strip.py b/src/vmr_http/models/strip.py new file mode 100644 index 0000000..fffabf8 --- /dev/null +++ b/src/vmr_http/models/strip.py @@ -0,0 +1,54 @@ +"""Models for the parameters of a strip.""" + +from typing import Optional + +from pydantic import BaseModel + + +class StripParams(BaseModel): + """Parameters for a single strip.""" + + gain: Optional[float] = None + mute: Optional[bool] = None + mono: Optional[bool] = None + solo: Optional[bool] = None + A1: Optional[bool] = None + A2: Optional[bool] = None + A3: Optional[bool] = None + A4: Optional[bool] = None + A5: Optional[bool] = None + B1: Optional[bool] = None + B2: Optional[bool] = None + B3: Optional[bool] = None + + +class StripCompParams(BaseModel): + """Parameters for the compressor of a strip.""" + + knob: Optional[float] = None + gainin: Optional[float] = None + ratio: Optional[float] = None + threshold: Optional[float] = None + attack: Optional[float] = None + release: Optional[float] = None + knee: Optional[float] = None + gainout: Optional[float] = None + makeup: Optional[bool] = None + + +class StripGateParams(BaseModel): + """Parameters for the gate of a strip.""" + + knob: Optional[float] = None + threshold: Optional[float] = None + damping: Optional[float] = None + bpsidechain: Optional[float] = None + attack: Optional[float] = None + hold: Optional[float] = None + release: Optional[float] = None + + +class StripDenoiserParams(BaseModel): + """Parameters for the denoiser of a strip.""" + + knob: Optional[float] = None diff --git a/src/vmr_http/web/__init__.py b/src/vmr_http/web/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/vmr_http/web/bus.py b/src/vmr_http/web/bus.py index af9233c..8838ee3 100644 --- a/src/vmr_http/web/bus.py +++ b/src/vmr_http/web/bus.py @@ -1,3 +1,55 @@ -from fastapi import APIRouter +"""module for bus-related endpoints.""" + +from fastapi import APIRouter, Body, Depends + +from vmr_http.dependencies import get_voicemeeter_client +from vmr_http.models.bus import BusParams + +from . import busmode router = APIRouter() +router.include_router(busmode.router, prefix='/mode', tags=['bus mode']) + + +@router.put('/{index}', tags=['bus']) +async def set_bus_params(index: int, request: BusParams, voicemeeter=Depends(get_voicemeeter_client)): + """Set multiple parameters of a bus at once.""" + bus = voicemeeter.bus[index] + for key, value in request.model_dump(exclude_unset=True).items(): + setattr(bus, key, value) + + return {key: getattr(bus, key) for key in request.model_dump(exclude_unset=True)} + + +@router.get('/{index}/gain', tags=['bus']) +async def get_gain(index: int, voicemeeter=Depends(get_voicemeeter_client)): + """Get the current gain value for the specified bus index.""" + return {'gain': voicemeeter.bus[index].gain} + + +@router.put('/{index}/gain', tags=['bus']) +async def set_gain( + index: int, + gain: float = Body(..., embed=True), + voicemeeter=Depends(get_voicemeeter_client), +): + """Set the gain value for the specified bus index.""" + voicemeeter.bus[index].gain = gain + return {'gain': voicemeeter.bus[index].gain} + + +@router.get('/{index}/mute', tags=['bus']) +async def get_mute(index: int, voicemeeter=Depends(get_voicemeeter_client)): + """Get the current mute status for the specified bus index.""" + return {'mute': voicemeeter.bus[index].mute} + + +@router.put('/{index}/mute', tags=['bus']) +async def set_mute( + index: int, + mute: bool = Body(..., embed=True), + voicemeeter=Depends(get_voicemeeter_client), +): + """Set the mute status for the specified bus index.""" + voicemeeter.bus[index].mute = mute + return {'mute': voicemeeter.bus[index].mute} diff --git a/src/vmr_http/web/busmode.py b/src/vmr_http/web/busmode.py new file mode 100644 index 0000000..066108d --- /dev/null +++ b/src/vmr_http/web/busmode.py @@ -0,0 +1,45 @@ +"""module for bus mode related endpoints.""" + +from fastapi import APIRouter, Body, Depends, HTTPException + +from vmr_http.dependencies import get_voicemeeter_client + +router = APIRouter() + +_readable_busmodes = { + 'normal': 'Normal', + 'amix': 'Mix Down A', + 'bmix': 'Mix Down B', + 'repeat': 'Stereo Repeat', + 'composite': 'Composite', + 'tvmix': 'Up Mix TV', + 'upmix21': 'Up Mix 2.1', + 'upmix41': 'Up Mix 4.1', + 'upmix61': 'Up Mix 6.1', + 'centeronly': 'Center Only', + 'lfeonly': 'Low Frequency Effect Only', + 'rearonly': 'Rear Only', +} +_reversed_busmodes = {v: k for k, v in _readable_busmodes.items()} + + +@router.get('/{index}') +async def get_bus_mode(index: int, voicemeeter=Depends(get_voicemeeter_client)): + """Get the current bus mode for the specified bus index.""" + return {'mode': _readable_busmodes.get(voicemeeter.bus[index].mode.get(), 'Unknown')} + + +@router.put('/{index}') +async def set_bus_mode( + index: int, + mode: str = Body(..., embed=True), + voicemeeter=Depends(get_voicemeeter_client), +): + """Set the bus mode for the specified bus index.""" + if mode not in _reversed_busmodes: + raise HTTPException( + status_code=400, detail=f'Invalid mode. Valid modes are: {", ".join(_reversed_busmodes.keys())}' + ) + + setattr(voicemeeter.bus[index].mode, _reversed_busmodes[mode], True) + return {'mode': _readable_busmodes[_reversed_busmodes[mode]]} diff --git a/src/vmr_http/web/strip.py b/src/vmr_http/web/strip.py index d4c0af3..0161994 100644 --- a/src/vmr_http/web/strip.py +++ b/src/vmr_http/web/strip.py @@ -1,10 +1,227 @@ -from fastapi import APIRouter, Depends +"""module for strip-related endpoints.""" + +from fastapi import APIRouter, Body, Depends from vmr_http.dependencies import get_voicemeeter_client +from vmr_http.models.strip import StripParams + +from . import stripcomp, stripdenoiser, stripgate router = APIRouter() +router.include_router(stripcomp.router, prefix='/comp', tags=['strip comp']) +router.include_router(stripgate.router, prefix='/gate', tags=['strip gate']) +router.include_router(stripdenoiser.router, prefix='/denoiser', tags=['strip denoiser']) -@router.get("/{index}/gain") +@router.put('/{index}', tags=['strip']) +async def set_strip_params(index: int, request: StripParams, voicemeeter=Depends(get_voicemeeter_client)): + """Set the parameters for the specified strip index.""" + strip = voicemeeter.strip[index] + for key, value in request.model_dump(exclude_unset=True).items(): + setattr(strip, key, value) + + return {key: getattr(strip, key) for key in request.model_dump(exclude_unset=True)} + + +@router.get('/{index}/gain', tags=['strip']) async def get_gain(index: int, voicemeeter=Depends(get_voicemeeter_client)): - return {"gain": voicemeeter.strip[index].gain} + """Get the current gain value for the specified strip index.""" + return {'gain': voicemeeter.strip[index].gain} + + +@router.put('/{index}/gain', tags=['strip']) +async def set_gain( + index: int, + gain: float = Body(..., embed=True), + voicemeeter=Depends(get_voicemeeter_client), +): + """Set the gain value for the specified strip index.""" + voicemeeter.strip[index].gain = gain + return {'gain': voicemeeter.strip[index].gain} + + +@router.get('/{index}/mute', tags=['strip']) +async def get_mute(index: int, voicemeeter=Depends(get_voicemeeter_client)): + """Get the current mute status for the specified strip index.""" + return {'mute': voicemeeter.strip[index].mute} + + +@router.put('/{index}/mute', tags=['strip']) +async def set_mute( + index: int, + mute: bool = Body(..., embed=True), + voicemeeter=Depends(get_voicemeeter_client), +): + """Set the mute status for the specified strip index.""" + voicemeeter.strip[index].mute = mute + return {'mute': voicemeeter.strip[index].mute} + + +@router.get('/{index}/mono', tags=['strip']) +async def get_mono(index: int, voicemeeter=Depends(get_voicemeeter_client)): + """Get the current mono status for the specified strip index.""" + return {'mono': voicemeeter.strip[index].mono} + + +@router.put('/{index}/mono', tags=['strip']) +async def set_mono( + index: int, + mono: bool = Body(..., embed=True), + voicemeeter=Depends(get_voicemeeter_client), +): + """Set the mono status for the specified strip index.""" + voicemeeter.strip[index].mono = mono + return {'mono': voicemeeter.strip[index].mono} + + +@router.get('/{index}/solo', tags=['strip']) +async def get_solo(index: int, voicemeeter=Depends(get_voicemeeter_client)): + """Get the current solo status for the specified strip index.""" + return {'solo': voicemeeter.strip[index].solo} + + +@router.put('/{index}/solo', tags=['strip']) +async def set_solo( + index: int, + solo: bool = Body(..., embed=True), + voicemeeter=Depends(get_voicemeeter_client), +): + """Set the solo status for the specified strip index.""" + voicemeeter.strip[index].solo = solo + return {'solo': voicemeeter.strip[index].solo} + + +@router.get('/{index}/A1', tags=['strip']) +async def get_A1(index: int, voicemeeter=Depends(get_voicemeeter_client)): + """Get the current A1 output status for the specified strip index.""" + return {'A1': voicemeeter.strip[index].A1} + + +@router.put('/{index}/A1', tags=['strip']) +async def set_A1( + index: int, + A1: bool = Body(..., embed=True), + voicemeeter=Depends(get_voicemeeter_client), +): + """Set the A1 output status for the specified strip index.""" + voicemeeter.strip[index].A1 = A1 + return {'A1': voicemeeter.strip[index].A1} + + +@router.get('/{index}/A2', tags=['strip']) +async def get_A2(index: int, voicemeeter=Depends(get_voicemeeter_client)): + """Get the current A2 output status for the specified strip index.""" + return {'A2': voicemeeter.strip[index].A2} + + +@router.put('/{index}/A2', tags=['strip']) +async def set_A2( + index: int, + A2: bool = Body(..., embed=True), + voicemeeter=Depends(get_voicemeeter_client), +): + """Set the A2 output status for the specified strip index.""" + voicemeeter.strip[index].A2 = A2 + return {'A2': voicemeeter.strip[index].A2} + + +@router.get('/{index}/A3', tags=['strip']) +async def get_A3(index: int, voicemeeter=Depends(get_voicemeeter_client)): + """Get the current A3 output status for the specified strip index.""" + return {'A3': voicemeeter.strip[index].A3} + + +@router.put('/{index}/A3', tags=['strip']) +async def set_A3( + index: int, + A3: bool = Body(..., embed=True), + voicemeeter=Depends(get_voicemeeter_client), +): + """Set the A3 output status for the specified strip index.""" + voicemeeter.strip[index].A3 = A3 + return {'A3': voicemeeter.strip[index].A3} + + +@router.get('/{index}/A4', tags=['strip']) +async def get_A4(index: int, voicemeeter=Depends(get_voicemeeter_client)): + """Get the current A4 output status for the specified strip index.""" + return {'A4': voicemeeter.strip[index].A4} + + +@router.put('/{index}/A4', tags=['strip']) +async def set_A4( + index: int, + A4: bool = Body(..., embed=True), + voicemeeter=Depends(get_voicemeeter_client), +): + """Set the A4 output status for the specified strip index.""" + voicemeeter.strip[index].A4 = A4 + return {'A4': voicemeeter.strip[index].A4} + + +@router.get('/{index}/A5', tags=['strip']) +async def get_A5(index: int, voicemeeter=Depends(get_voicemeeter_client)): + """Get the current A5 output status for the specified strip index.""" + return {'A5': voicemeeter.strip[index].A5} + + +@router.put('/{index}/A5', tags=['strip']) +async def set_A5( + index: int, + A5: bool = Body(..., embed=True), + voicemeeter=Depends(get_voicemeeter_client), +): + """Set the A5 output status for the specified strip index.""" + voicemeeter.strip[index].A5 = A5 + return {'A5': voicemeeter.strip[index].A5} + + +@router.get('/{index}/B1', tags=['strip']) +async def get_B1(index: int, voicemeeter=Depends(get_voicemeeter_client)): + """Get the current B1 output status for the specified strip index.""" + return {'B1': voicemeeter.strip[index].B1} + + +@router.put('/{index}/B1', tags=['strip']) +async def set_B1( + index: int, + B1: bool = Body(..., embed=True), + voicemeeter=Depends(get_voicemeeter_client), +): + """Set the B1 output status for the specified strip index.""" + voicemeeter.strip[index].B1 = B1 + return {'B1': voicemeeter.strip[index].B1} + + +@router.get('/{index}/B2', tags=['strip']) +async def get_B2(index: int, voicemeeter=Depends(get_voicemeeter_client)): + """Get the current B2 output status for the specified strip index.""" + return {'B2': voicemeeter.strip[index].B2} + + +@router.put('/{index}/B2', tags=['strip']) +async def set_B2( + index: int, + B2: bool = Body(..., embed=True), + voicemeeter=Depends(get_voicemeeter_client), +): + """Set the B2 output status for the specified strip index.""" + voicemeeter.strip[index].B2 = B2 + return {'B2': voicemeeter.strip[index].B2} + + +@router.get('/{index}/B3', tags=['strip']) +async def get_B3(index: int, voicemeeter=Depends(get_voicemeeter_client)): + """Get the current B3 output status for the specified strip index.""" + return {'B3': voicemeeter.strip[index].B3} + + +@router.put('/{index}/B3', tags=['strip']) +async def set_B3( + index: int, + B3: bool = Body(..., embed=True), + voicemeeter=Depends(get_voicemeeter_client), +): + """Set the B3 output status for the specified strip index.""" + voicemeeter.strip[index].B3 = B3 + return {'B3': voicemeeter.strip[index].B3} diff --git a/src/vmr_http/web/stripcomp.py b/src/vmr_http/web/stripcomp.py new file mode 100644 index 0000000..21e3814 --- /dev/null +++ b/src/vmr_http/web/stripcomp.py @@ -0,0 +1,37 @@ +"""module for strip compressor related endpoints.""" + +from fastapi import APIRouter, Body, Depends + +from vmr_http.dependencies import get_voicemeeter_client +from vmr_http.models.strip import StripCompParams + +router = APIRouter() + + +@router.put('/{index}/comp') +async def set_strip_comp_params(index: int, request: StripCompParams, voicemeeter=Depends(get_voicemeeter_client)): + """Set the compressor parameters for the specified strip index.""" + strip_comp = voicemeeter.strip[index].comp + for key, value in request.model_dump(exclude_unset=True).items(): + setattr(strip_comp, key, value) + + return {key: getattr(strip_comp, key) for key in request.model_dump(exclude_unset=True)} + + +@router.get('/{index}/comp/knob') +async def get_strip_comp_knob(index: int, voicemeeter=Depends(get_voicemeeter_client)): + """Get the current compressor knob value for the specified strip index.""" + strip_comp = voicemeeter.strip[index].comp + return {'knob': strip_comp.knob} + + +@router.put('/{index}/comp/knob') +async def set_strip_comp_knob( + index: int, + knob: float = Body(..., embed=True), + voicemeeter=Depends(get_voicemeeter_client), +): + """Set the compressor knob value for the specified strip index.""" + strip_comp = voicemeeter.strip[index].comp + strip_comp.knob = knob + return {'knob': strip_comp.knob} diff --git a/src/vmr_http/web/stripdenoiser.py b/src/vmr_http/web/stripdenoiser.py new file mode 100644 index 0000000..614f25d --- /dev/null +++ b/src/vmr_http/web/stripdenoiser.py @@ -0,0 +1,44 @@ +"""module for strip denoiser related endpoints.""" + +from fastapi import APIRouter, Body, Depends + +from vmr_http.dependencies import get_voicemeeter_client +from vmr_http.models.strip import StripDenoiserParams + +router = APIRouter() + + +@router.put('/{index}/denoiser') +async def set_strip_denoiser_params( + index: int, + request: StripDenoiserParams, + voicemeeter=Depends(get_voicemeeter_client), +): + """Set the denoiser parameters for the specified strip index.""" + strip_denoiser = voicemeeter.strip[index].denoiser + for key, value in request.model_dump(exclude_unset=True).items(): + setattr(strip_denoiser, key, value) + + return {key: getattr(strip_denoiser, key) for key in request.model_dump(exclude_unset=True)} + + +@router.get('/{index}/denoiser/knob') +async def get_strip_denoiser_knob( + index: int, + voicemeeter=Depends(get_voicemeeter_client), +): + """Get the denoiser knob value for the specified strip index.""" + strip_denoiser = voicemeeter.strip[index].denoiser + return {'knob': strip_denoiser.knob} + + +@router.put('/{index}/denoiser/knob') +async def set_strip_denoiser_knob( + index: int, + knob: float = Body(..., embed=True), + voicemeeter=Depends(get_voicemeeter_client), +): + """Set the denoiser knob value for the specified strip index.""" + strip_denoiser = voicemeeter.strip[index].denoiser + strip_denoiser.knob = knob + return {'knob': strip_denoiser.knob} diff --git a/src/vmr_http/web/stripgate.py b/src/vmr_http/web/stripgate.py new file mode 100644 index 0000000..0727b5b --- /dev/null +++ b/src/vmr_http/web/stripgate.py @@ -0,0 +1,37 @@ +"""module for strip gate related endpoints.""" + +from fastapi import APIRouter, Body, Depends + +from vmr_http.dependencies import get_voicemeeter_client +from vmr_http.models.strip import StripGateParams + +router = APIRouter() + + +@router.put('/{index}/gate') +async def set_strip_gate_params(index: int, request: StripGateParams, voicemeeter=Depends(get_voicemeeter_client)): + """Set the gate parameters for the specified strip index.""" + strip_gate = voicemeeter.strip[index].gate + for key, value in request.model_dump(exclude_unset=True).items(): + setattr(strip_gate, key, value) + + return {key: getattr(strip_gate, key) for key in request.model_dump(exclude_unset=True)} + + +@router.get('/{index}/gate/knob') +async def get_strip_gate_knob(index: int, voicemeeter=Depends(get_voicemeeter_client)): + """Get the current gate knob value for the specified strip index.""" + strip_gate = voicemeeter.strip[index].gate + return {'knob': strip_gate.knob} + + +@router.put('/{index}/gate/knob') +async def set_strip_gate_knob( + index: int, + knob: float = Body(..., embed=True), + voicemeeter=Depends(get_voicemeeter_client), +): + """Set the gate knob value for the specified strip index.""" + strip_gate = voicemeeter.strip[index].gate + strip_gate.knob = knob + return {'knob': strip_gate.knob}