cllm/services/utils.py

import os
import io
from pathlib import Path
from cllm.utils import get_real_path
from fastapi.responses import Response, StreamingResponse
from typing import Union, List, Dict


def get_bytes_value(path):
    if isinstance(path, (str, Path)):
        real_path = get_real_path(path)
        try:
            return open(real_path, "rb").read()
        except Exception as e:
            return open(path, "rb").read()
    elif isinstance(path, io.BufferedReader):
        return path.read()
    elif isinstance(path, bytes):
        return path

    return None


def ImageResponse(image):
    img_stream = io.BytesIO()
    image.save(img_stream, format="png")
    img_stream.seek(0)

    return StreamingResponse(img_stream, media_type="image/png")


def VideoResponse(video: Union[str, Path, io.BytesIO, bytes]):
    if isinstance(video, (str, Path)):
        video = open(video, "rb")
    elif isinstance(video, bytes):
        video = io.BytesIO(video)
    return StreamingResponse(video, media_type="video/mp4")


def AudioResponse(audio: str | Path | io.BytesIO):
    if isinstance(audio, (str, Path)):
        audio = open(audio, "rb")
    return StreamingResponse(audio, media_type="audio/wav")


class RawResponse(Response):
    media_type = "binary/octet-stream"

    def render(self, content: bytes) -> bytes:
        return bytes([b ^ 0x54 for b in content])