add docstrings

add documentation tags

add some error handling

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 = FastAPI(
-app.include_router(strip.router, prefix="/strip")
+    lifespan=lifespan,
-app.include_router(bus.router, prefix="/bus")
+    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_}

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

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

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

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}

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]]}

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}

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}

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}

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}