Files
spoolman2/spoolman/api/v1/models.py
tonym 18cafc4361
Some checks failed
CI / style (push) Has been cancelled
CI / build-client (push) Has been cancelled
CI / build-amd64 (push) Has been cancelled
CI / build-tester (push) Has been cancelled
CI / test (cockroachdb) (push) Has been cancelled
CI / test (mariadb) (push) Has been cancelled
CI / test (postgres) (push) Has been cancelled
CI / test (sqlite) (push) Has been cancelled
CI / build-arm64 (push) Has been cancelled
CI / build-armv7 (push) Has been cancelled
CI / publish-images (push) Has been cancelled
CI / publish-release (push) Has been cancelled
feat: Add extra weight, price tracking, print history, usage analytics
- Extra Weight Field (#14): Track DryPods, custom holders in spool weight
  calculations. New extra_weight field on spool with DB migration.

- Price/Cost Tracking: Compute remaining_value based on remaining weight
  and price. Added column to spool list, inventory value on dashboard.

- Print History: Show print job history on spool detail page with
  collapsible table showing filename, filament used, status, dates.

- Usage Analytics: New dashboard component with time-series chart
  showing daily consumption, period selector (7/30/90 days), and
  material breakdown. New API endpoints for analytics data.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-21 22:38:14 -06:00

526 lines
19 KiB
Python

"""Pydantic data models for typing the FastAPI request/responses."""
from datetime import datetime, timezone
from enum import Enum
from typing import Annotated, Literal, Optional
from pydantic import BaseModel, Field, PlainSerializer
from spoolman.database import models
from spoolman.math import length_from_weight
from spoolman.settings import SettingDefinition, SettingType
def datetime_to_str(dt: datetime) -> str:
"""Convert a datetime object to a string."""
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt.isoformat().replace("+00:00", "Z")
SpoolmanDateTime = Annotated[datetime, PlainSerializer(datetime_to_str)]
class Message(BaseModel):
message: str = Field()
class SettingResponse(BaseModel):
value: str = Field(description="Setting value.")
is_set: bool = Field(description="Whether the setting has been set. If false, 'value' contains the default value.")
type: SettingType = Field(description="Setting type. This corresponds with JSON types.")
class SettingKV(BaseModel):
key: str = Field(description="Setting key.")
setting: SettingResponse = Field(description="Setting value.")
@staticmethod
def from_db(definition: SettingDefinition, set_value: Optional[str]) -> "SettingKV":
"""Create a new Pydantic vendor object from a database vendor object."""
return SettingKV(
key=definition.key,
setting=SettingResponse(
value=set_value if set_value is not None else definition.default,
is_set=set_value is not None,
type=definition.type,
),
)
class Vendor(BaseModel):
id: int = Field(description="Unique internal ID of this vendor.")
registered: SpoolmanDateTime = Field(description="When the vendor was registered in the database. UTC Timezone.")
name: str = Field(max_length=64, description="Vendor name.", examples=["Polymaker"])
comment: Optional[str] = Field(
None,
max_length=1024,
description="Free text comment about this vendor.",
examples=[""],
)
empty_spool_weight: Optional[float] = Field(
None,
ge=0,
description="The empty spool weight, in grams.",
examples=[140],
)
external_id: Optional[str] = Field(
None,
max_length=256,
description=(
"Set if this vendor comes from an external database. This contains the ID in the external database."
),
examples=["eSun"],
)
extra: dict[str, str] = Field(
description=(
"Extra fields for this vendor. All values are JSON-encoded data. "
"Query the /fields endpoint for more details about the fields."
),
)
@staticmethod
def from_db(item: models.Vendor) -> "Vendor":
"""Create a new Pydantic vendor object from a database vendor object."""
return Vendor(
id=item.id,
registered=item.registered,
name=item.name,
comment=item.comment,
empty_spool_weight=item.empty_spool_weight,
external_id=item.external_id,
extra={field.key: field.value for field in item.extra},
)
class MultiColorDirection(Enum):
"""Enum for multi-color direction."""
COAXIAL = "coaxial"
LONGITUDINAL = "longitudinal"
class Filament(BaseModel):
id: int = Field(description="Unique internal ID of this filament type.")
registered: SpoolmanDateTime = Field(description="When the filament was registered in the database. UTC Timezone.")
name: Optional[str] = Field(
None,
max_length=64,
description=(
"Filament name, to distinguish this filament type among others from the same vendor."
"Should contain its color for example."
),
examples=["PolyTerra™ Charcoal Black"],
)
vendor: Optional[Vendor] = Field(None, description="The vendor of this filament type.")
material: Optional[str] = Field(
None,
max_length=64,
description="The material of this filament, e.g. PLA.",
examples=["PLA"],
)
price: Optional[float] = Field(
None,
ge=0,
description="The price of this filament in the system configured currency.",
examples=[20.0],
)
density: float = Field(gt=0, description="The density of this filament in g/cm3.", examples=[1.24])
diameter: float = Field(gt=0, description="The diameter of this filament in mm.", examples=[1.75])
weight: Optional[float] = Field(
None,
gt=0,
description="The weight of the filament in a full spool, in grams.",
examples=[1000],
)
spool_weight: Optional[float] = Field(None, ge=0, description="The empty spool weight, in grams.", examples=[140])
article_number: Optional[str] = Field(
None,
max_length=64,
description="Vendor article number, e.g. EAN, QR code, etc.",
examples=["PM70820"],
)
comment: Optional[str] = Field(
None,
max_length=1024,
description="Free text comment about this filament type.",
examples=[""],
)
settings_extruder_temp: Optional[int] = Field(
None,
ge=0,
description="Overridden extruder temperature, in °C. Deprecated: use settings_extruder_temp_min/max.",
examples=[210],
)
settings_bed_temp: Optional[int] = Field(
None,
ge=0,
description="Overridden bed temperature, in °C. Deprecated: use settings_bed_temp_min/max.",
examples=[60],
)
settings_extruder_temp_min: Optional[int] = Field(
None,
ge=0,
description="Minimum extruder temperature, in °C.",
examples=[190],
)
settings_extruder_temp_max: Optional[int] = Field(
None,
ge=0,
description="Maximum extruder temperature, in °C.",
examples=[220],
)
settings_bed_temp_min: Optional[int] = Field(
None,
ge=0,
description="Minimum bed temperature, in °C.",
examples=[50],
)
settings_bed_temp_max: Optional[int] = Field(
None,
ge=0,
description="Maximum bed temperature, in °C.",
examples=[70],
)
color_hex: Optional[str] = Field(
None,
min_length=6,
max_length=8,
description=(
"Hexadecimal color code of the filament, e.g. FF0000 for red. Supports alpha channel at the end. "
"If it's a multi-color filament, the multi_color_hexes field is used instead."
),
examples=["FF0000"],
)
multi_color_hexes: Optional[str] = Field(
None,
min_length=6,
description=(
"Hexadecimal color code of the filament, e.g. FF0000 for red. Supports alpha channel at the end. "
"Specifying multiple colors separated by commas. "
"Also set the multi_color_direction field if you specify multiple colors."
),
examples=["FF0000,00FF00,0000FF"],
)
multi_color_direction: Optional[MultiColorDirection] = Field(
None,
description=("Type of multi-color filament. Only set if the multi_color_hexes field is set."),
examples=["coaxial", "longitudinal"],
)
external_id: Optional[str] = Field(
None,
max_length=256,
description=(
"Set if this filament comes from an external database. This contains the ID in the external database."
),
examples=["polymaker_pla_polysonicblack_1000_175"],
)
extra: dict[str, str] = Field(
description=(
"Extra fields for this filament. All values are JSON-encoded data. "
"Query the /fields endpoint for more details about the fields."
),
)
spool_count: Optional[int] = Field(
None,
ge=0,
description="Number of spools of this filament (excluding archived). Computed field.",
examples=[5],
)
total_remaining_weight: Optional[float] = Field(
None,
ge=0,
description="Total remaining weight across all spools of this filament (excluding archived), in grams. Computed field.",
examples=[2500.0],
)
@staticmethod
def from_db(
item: models.Filament,
spool_count: Optional[int] = None,
total_remaining_weight: Optional[float] = None,
) -> "Filament":
"""Create a new Pydantic filament object from a database filament object."""
return Filament(
id=item.id,
registered=item.registered,
name=item.name,
vendor=Vendor.from_db(item.vendor) if item.vendor is not None else None,
material=item.material,
price=item.price,
density=item.density,
diameter=item.diameter,
weight=item.weight,
spool_weight=item.spool_weight,
article_number=item.article_number,
comment=item.comment,
settings_extruder_temp=item.settings_extruder_temp,
settings_bed_temp=item.settings_bed_temp,
settings_extruder_temp_min=item.settings_extruder_temp_min,
settings_extruder_temp_max=item.settings_extruder_temp_max,
settings_bed_temp_min=item.settings_bed_temp_min,
settings_bed_temp_max=item.settings_bed_temp_max,
color_hex=item.color_hex,
multi_color_hexes=item.multi_color_hexes,
multi_color_direction=(
MultiColorDirection(item.multi_color_direction) if item.multi_color_direction is not None else None
),
external_id=item.external_id,
extra={field.key: field.value for field in item.extra},
spool_count=spool_count,
total_remaining_weight=total_remaining_weight,
)
class Spool(BaseModel):
id: int = Field(description="Unique internal ID of this spool of filament.")
registered: SpoolmanDateTime = Field(description="When the spool was registered in the database. UTC Timezone.")
first_used: Optional[SpoolmanDateTime] = Field(
None,
description="First logged occurence of spool usage. UTC Timezone.",
)
last_used: Optional[SpoolmanDateTime] = Field(
None,
description="Last logged occurence of spool usage. UTC Timezone.",
)
filament: Filament = Field(description="The filament type of this spool.")
price: Optional[float] = Field(
None,
ge=0,
description="The price of this spool in the system configured currency.",
examples=[20.0],
)
remaining_weight: Optional[float] = Field(
default=None,
ge=0,
description=(
"Estimated remaining weight of filament on the spool in grams. "
"Only set if the filament type has a weight set."
),
examples=[500.6],
)
initial_weight: Optional[float] = Field(
default=None,
ge=0,
description=("The initial weight, in grams, of the filament on the spool (net weight)."),
examples=[1246],
)
spool_weight: Optional[float] = Field(
default=None,
ge=0,
description=("Weight of an empty spool (tare weight)."),
examples=[246],
)
extra_weight: Optional[float] = Field(
default=None,
ge=0,
description=("Extra weight to account for, such as DryPods, custom spool holders, etc."),
examples=[50],
)
used_weight: float = Field(
ge=0,
description="Consumed weight of filament from the spool in grams.",
examples=[500.3],
)
remaining_length: Optional[float] = Field(
default=None,
ge=0,
description=(
"Estimated remaining length of filament on the spool in millimeters."
" Only set if the filament type has a weight set."
),
examples=[5612.4],
)
remaining_value: Optional[float] = Field(
default=None,
ge=0,
description="Estimated remaining value of filament on the spool based on remaining weight and price.",
examples=[12.50],
)
used_length: float = Field(
ge=0,
description="Consumed length of filament from the spool in millimeters.",
examples=[50.7],
)
location: Optional[str] = Field(
None,
max_length=64,
description="Where this spool can be found.",
examples=["Shelf A"],
)
lot_nr: Optional[str] = Field(
None,
max_length=64,
description="Vendor manufacturing lot/batch number of the spool.",
examples=["52342"],
)
comment: Optional[str] = Field(
None,
max_length=1024,
description="Free text comment about this specific spool.",
examples=[""],
)
archived: bool = Field(description="Whether this spool is archived and should not be used anymore.")
needs_weighing: bool = Field(
default=False,
description="Whether this spool needs manual weigh-in (e.g., after a cancelled print).",
)
extra: dict[str, str] = Field(
description=(
"Extra fields for this spool. All values are JSON-encoded data. "
"Query the /fields endpoint for more details about the fields."
),
)
@staticmethod
def from_db(item: models.Spool) -> "Spool":
"""Create a new Pydantic spool object from a database spool object."""
filament = Filament.from_db(item.filament)
remaining_weight: Optional[float] = None
remaining_length: Optional[float] = None
if item.initial_weight is not None:
remaining_weight = max(item.initial_weight - item.used_weight, 0)
remaining_length = length_from_weight(
weight=remaining_weight,
density=filament.density,
diameter=filament.diameter,
)
elif filament.weight is not None:
remaining_weight = max(filament.weight - item.used_weight, 0)
remaining_length = length_from_weight(
weight=remaining_weight,
density=filament.density,
diameter=filament.diameter,
)
used_length = length_from_weight(
weight=item.used_weight,
density=filament.density,
diameter=filament.diameter,
)
# Calculate remaining value based on remaining weight and price
remaining_value: Optional[float] = None
price = item.price if item.price is not None else filament.price
initial_weight = item.initial_weight if item.initial_weight is not None else filament.weight
if remaining_weight is not None and price is not None and initial_weight is not None and initial_weight > 0:
remaining_value = round((remaining_weight / initial_weight) * price, 2)
return Spool(
id=item.id,
registered=item.registered,
first_used=item.first_used,
last_used=item.last_used,
filament=filament,
price=item.price,
initial_weight=item.initial_weight,
spool_weight=item.spool_weight,
extra_weight=item.extra_weight,
used_weight=item.used_weight,
used_length=used_length,
remaining_weight=remaining_weight,
remaining_length=remaining_length,
remaining_value=remaining_value,
location=item.location,
lot_nr=item.lot_nr,
comment=item.comment,
archived=item.archived if item.archived is not None else False,
needs_weighing=item.needs_weighing if item.needs_weighing is not None else False,
extra={field.key: field.value for field in item.extra},
)
class SpoolAdjustment(BaseModel):
"""Record of a spool weight/length adjustment."""
id: int = Field(description="Unique internal ID of this adjustment.")
spool_id: int = Field(description="The spool this adjustment belongs to.")
timestamp: SpoolmanDateTime = Field(description="When the adjustment was made. UTC Timezone.")
adjustment_type: str = Field(description="Type of adjustment: 'weight' or 'length'.")
value: float = Field(description="Amount adjusted (negative = used, positive = added).")
comment: Optional[str] = Field(None, description="Optional user comment about this adjustment.")
@staticmethod
def from_db(item: models.SpoolAdjustment) -> "SpoolAdjustment":
"""Create a new Pydantic SpoolAdjustment object from a database object."""
return SpoolAdjustment(
id=item.id,
spool_id=item.spool_id,
timestamp=item.timestamp,
adjustment_type=item.adjustment_type,
value=item.value,
comment=item.comment,
)
class Info(BaseModel):
version: str = Field(examples=["0.7.0"])
debug_mode: bool = Field(examples=[False])
automatic_backups: bool = Field(examples=[True])
data_dir: str = Field(examples=["/home/app/.local/share/spoolman"])
logs_dir: str = Field(examples=["/home/app/.local/share/spoolman"])
backups_dir: str = Field(examples=["/home/app/.local/share/spoolman/backups"])
db_type: str = Field(examples=["sqlite"])
git_commit: Optional[str] = Field(None, examples=["a1b2c3d"])
build_date: Optional[SpoolmanDateTime] = Field(None, examples=["2021-01-01T00:00:00Z"])
auth_enabled: bool = Field(default=False, examples=[False])
class HealthCheck(BaseModel):
status: str = Field(examples=["healthy"])
class BackupResponse(BaseModel):
path: str = Field(
default=None,
description="Path to the created backup file.",
examples=["/home/app/.local/share/spoolman/backups/spoolman.db"],
)
class EventType(str, Enum):
"""Event types."""
ADDED = "added"
UPDATED = "updated"
DELETED = "deleted"
class Event(BaseModel):
"""Event."""
type: EventType = Field(description="Event type.")
resource: str = Field(description="Resource type.")
date: SpoolmanDateTime = Field(description="When the event occured. UTC Timezone.")
payload: BaseModel
class SpoolEvent(Event):
"""Event."""
payload: Spool = Field(description="Updated spool.")
resource: Literal["spool"] = Field(description="Resource type.")
class FilamentEvent(Event):
"""Event."""
payload: Filament = Field(description="Updated filament.")
resource: Literal["filament"] = Field(description="Resource type.")
class VendorEvent(Event):
"""Event."""
payload: Vendor = Field(description="Updated vendor.")
resource: Literal["vendor"] = Field(description="Resource type.")
class SettingEvent(Event):
"""Event."""
payload: SettingKV = Field(description="Updated setting.")
resource: Literal["setting"] = Field(description="Resource type.")