spoolman2/spoolman/api/v1/other.py

"""Filament related endpoints."""

import logging
from datetime import datetime, timedelta
from typing import Annotated, Optional

import sqlalchemy
from fastapi import APIRouter, Depends, Query
from pydantic import BaseModel, Field, RootModel
from sqlalchemy import func
from sqlalchemy.ext.asyncio import AsyncSession

from spoolman.database import filament, models, spool
from spoolman.database.database import get_db_session

logger = logging.getLogger(__name__)

router = APIRouter(
    prefix="",
    tags=["other"],
)

# ruff: noqa: D103, B008


@router.get(
    "/material",
    name="Find materials",
    description="Get a list of all filament materials.",
    response_model_exclude_none=True,
    responses={
        200: {
            "description": "A list of all filament materials.",
            "content": {
                "application/json": {
                    "example": [
                        "PLA",
                        "ABS",
                        "PETG",
                    ],
                },
            },
        },
    },
)
async def find_materials(
    *,
    db: Annotated[AsyncSession, Depends(get_db_session)],
) -> list[str]:
    return await filament.find_materials(db=db)


@router.get(
    "/color",
    name="Find colors",
    description="Get a list of all filament colors (hex codes).",
    response_model_exclude_none=True,
    responses={
        200: {
            "description": "A list of all filament color hex codes.",
            "content": {
                "application/json": {
                    "example": [
                        "FF0000",
                        "00FF00",
                        "0000FF",
                    ],
                },
            },
        },
    },
)
async def find_colors(
    *,
    db: Annotated[AsyncSession, Depends(get_db_session)],
) -> list[str]:
    return await filament.find_colors(db=db)


@router.get(
    "/article-number",
    name="Find article numbers",
    description="Get a list of all article numbers.",
    response_model_exclude_none=True,
    responses={
        200: {
            "description": "A list of all article numbers.",
            "content": {
                "application/json": {
                    "example": [
                        "123456",
                        "987654",
                    ],
                },
            },
        },
    },
)
async def find_article_numbers(
    *,
    db: Annotated[AsyncSession, Depends(get_db_session)],
) -> list[str]:
    return await filament.find_article_numbers(db=db)


@router.get(
    "/lot-number",
    name="Find lot numbers",
    description="Get a list of all lot numbers.",
    response_model_exclude_none=True,
    responses={
        200: {
            "description": "A list of all lot numbers.",
            "content": {
                "application/json": {
                    "example": [
                        "123456",
                        "987654",
                    ],
                },
            },
        },
    },
)
async def find_lot_numbers(
    *,
    db: Annotated[AsyncSession, Depends(get_db_session)],
) -> list[str]:
    return await spool.find_lot_numbers(db=db)


@router.get(
    "/location",
    name="Find locations",
    description="Get a list of all spool locations.",
    response_model_exclude_none=True,
    responses={
        200: {
            "description": "A list of all spool locations.",
            "content": {
                "application/json": {
                    "example": [
                        "Printer 1",
                        "Printer 2",
                        "Storage Shelf A",
                    ],
                },
            },
        },
    },
)
async def find_locations(
    *,
    db: Annotated[AsyncSession, Depends(get_db_session)],
) -> list[str]:
    return await spool.find_locations(db=db)


class RenameLocationBody(BaseModel):
    name: str = Field(description="The new name of the location.", min_length=1)


@router.patch(
    "/location/{location}",
    name="Rename location",
    description="Rename a spool location. All spools in this location will be moved to the new location.",
    response_model_exclude_none=True,
    response_model=RootModel[str],
)
async def rename_location(
    location: str,
    *,
    db: Annotated[AsyncSession, Depends(get_db_session)],
    body: RenameLocationBody,
) -> str:
    logger.info("Renaming location %s to %s", location, body.name)
    await spool.rename_location(db=db, current_name=location, new_name=body.name)
    return body.name


class UsageDataPoint(BaseModel):
    """A single data point for usage analytics."""

    date: str = Field(description="Date in YYYY-MM-DD format.")
    weight_used: float = Field(description="Total weight used in grams (positive value).")


@router.get(
    "/analytics/usage",
    name="Get usage analytics",
    description="Get filament usage data aggregated by day for the specified period.",
    response_model=list[UsageDataPoint],
)
async def get_usage_analytics(
    *,
    db: Annotated[AsyncSession, Depends(get_db_session)],
    days: Annotated[
        int,
        Query(description="Number of days to look back.", ge=1, le=365),
    ] = 30,
) -> list[UsageDataPoint]:
    """Get usage analytics aggregated by day."""
    # Calculate the start date
    start_date = datetime.utcnow() - timedelta(days=days)

    # Query adjustments grouped by date
    # Only include negative values (filament used, not added)
    stmt = (
        sqlalchemy.select(
            func.date(models.SpoolAdjustment.timestamp).label("date"),
            func.sum(
                sqlalchemy.case(
                    (models.SpoolAdjustment.value < 0, -models.SpoolAdjustment.value),
                    else_=0,
                )
            ).label("weight_used"),
        )
        .where(models.SpoolAdjustment.timestamp >= start_date)
        .where(models.SpoolAdjustment.adjustment_type == "weight")
        .group_by(func.date(models.SpoolAdjustment.timestamp))
        .order_by(func.date(models.SpoolAdjustment.timestamp))
    )

    result = await db.execute(stmt)
    rows = result.all()

    # Fill in missing dates with 0
    data_by_date = {str(row.date): row.weight_used for row in rows}
    all_dates = []
    current_date = start_date.date()
    end_date = datetime.utcnow().date()

    while current_date <= end_date:
        date_str = current_date.strftime("%Y-%m-%d")
        all_dates.append(
            UsageDataPoint(
                date=date_str,
                weight_used=data_by_date.get(date_str, 0),
            )
        )
        current_date += timedelta(days=1)

    return all_dates


class MaterialUsage(BaseModel):
    """Usage breakdown by material."""

    material: str = Field(description="Material name.")
    weight_used: float = Field(description="Total weight used in grams.")


@router.get(
    "/analytics/by-material",
    name="Get usage by material",
    description="Get filament usage breakdown by material type.",
    response_model=list[MaterialUsage],
)
async def get_usage_by_material(
    *,
    db: Annotated[AsyncSession, Depends(get_db_session)],
    days: Annotated[
        Optional[int],
        Query(description="Number of days to look back. If not set, returns all-time usage.", ge=1, le=365),
    ] = None,
) -> list[MaterialUsage]:
    """Get usage breakdown by material."""
    # Join adjustments with spools and filaments to get material
    stmt = (
        sqlalchemy.select(
            func.coalesce(models.Filament.material, "Unknown").label("material"),
            func.sum(
                sqlalchemy.case(
                    (models.SpoolAdjustment.value < 0, -models.SpoolAdjustment.value),
                    else_=0,
                )
            ).label("weight_used"),
        )
        .join(models.Spool, models.SpoolAdjustment.spool_id == models.Spool.id)
        .join(models.Filament, models.Spool.filament_id == models.Filament.id)
        .where(models.SpoolAdjustment.adjustment_type == "weight")
    )

    if days is not None:
        start_date = datetime.utcnow() - timedelta(days=days)
        stmt = stmt.where(models.SpoolAdjustment.timestamp >= start_date)

    stmt = stmt.group_by(models.Filament.material).order_by(func.sum(-models.SpoolAdjustment.value).desc())

    result = await db.execute(stmt)
    rows = result.all()

    return [MaterialUsage(material=row.material or "Unknown", weight_used=row.weight_used) for row in rows]