mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 03:01:21 +02:00
592 lines
22 KiB
Python
592 lines
22 KiB
Python
"""Batch order planning: per-plate targets, progress, and staged dispatch (#342).
|
|
|
|
A batch stores *intent* in :class:`PrintBatchPlate` rows — "this order wants 3
|
|
of plate 2" — while its queue items record what was actually dispatched.
|
|
Everything here derives one from the other.
|
|
|
|
The distinction matters for exactly one reason, and it is the reason the
|
|
feature exists: a failed or cancelled run does not count towards the target, so
|
|
``remaining`` goes back up and the order still says it owes a print. A design
|
|
that only counted the items it created could not tell "the user cancelled this
|
|
deliberately" apart from "this one burned and needs reprinting".
|
|
|
|
Batches created before targets existed have no plate rows. They still report
|
|
progress — the plate breakdown is derived from their queue items and every
|
|
target simply equals the number of items dispatched, so ``remaining`` is zero
|
|
and the dispatch endpoint has nothing to do. ``has_targets`` tells callers
|
|
which kind of batch they are looking at.
|
|
"""
|
|
|
|
import logging
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import func, select, text
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from backend.app.models.print_batch import PrintBatch, PrintBatchPlate
|
|
from backend.app.models.print_log import PrintLogEntry
|
|
from backend.app.models.print_queue import PrintQueueItem, PrintQueueVariant
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Statuses that consume a unit of the target. "printing" counts because the
|
|
# run is in flight — re-dispatching it would double-print. "failed",
|
|
# "cancelled" and "skipped" deliberately do not.
|
|
CONSUMING_STATUSES = ("pending", "printing", "completed")
|
|
|
|
# Queue statuses the roll-up has a counter for. Anything else is ignored rather
|
|
# than crashing the page — the queue's status vocabulary is allowed to grow
|
|
# without this module having to be updated in lockstep.
|
|
COUNTED_STATUSES = ("pending", "printing", "completed", "failed", "cancelled", "skipped")
|
|
|
|
# Columns copied onto a clone when dispatching more of a plate. This is the
|
|
# print *configuration* the user already chose and the API already validated —
|
|
# copying the row is what keeps a second dispatch identical to the first
|
|
# without re-serialising twenty fields through a template blob that would drift
|
|
# from the model the first time someone adds a column.
|
|
CLONED_SETTING_COLUMNS = (
|
|
"printer_id",
|
|
"target_model",
|
|
"target_location",
|
|
"required_filament_types",
|
|
"archive_id",
|
|
"library_file_id",
|
|
"project_id",
|
|
"batch_id",
|
|
"ams_mapping",
|
|
"filament_overrides",
|
|
"plate_id",
|
|
"print_time_seconds",
|
|
"gcode_injection",
|
|
"nozzle_mapping",
|
|
"nozzle_rack_choice",
|
|
"require_previous_success",
|
|
"auto_off_after",
|
|
"manual_start",
|
|
"bed_levelling",
|
|
"flow_cali",
|
|
"vibration_cali",
|
|
"layer_inspect",
|
|
"timelapse",
|
|
"use_ams",
|
|
"nozzle_offset_cali",
|
|
"confirm_outcome",
|
|
"preheat_override",
|
|
"preheat_chamber_target_override",
|
|
"skip_filament_check",
|
|
)
|
|
|
|
CLONED_VARIANT_COLUMNS = (
|
|
"position",
|
|
"library_file_id",
|
|
"target_model",
|
|
"plate_id",
|
|
"ams_mapping",
|
|
"nozzle_mapping",
|
|
"nozzle_rack_choice",
|
|
"filament_overrides",
|
|
"required_filament_types",
|
|
"print_time_seconds",
|
|
)
|
|
|
|
|
|
class BatchDispatchError(Exception):
|
|
"""Raised when more runs are owed but nothing can be cloned to produce them."""
|
|
|
|
|
|
@dataclass
|
|
class PlateProgress:
|
|
"""Per-plate roll-up for one batch."""
|
|
|
|
plate_id: int | None
|
|
plate_name: str | None
|
|
quantity_target: int
|
|
sort_order: int = 0
|
|
pending: int = 0
|
|
printing: int = 0
|
|
completed: int = 0
|
|
failed: int = 0
|
|
cancelled: int = 0
|
|
skipped: int = 0
|
|
# Actual material + energy cost of this plate's finished runs. None when no
|
|
# run has produced a cost yet — reported as "unknown", never as zero.
|
|
actual_cost: float | None = None
|
|
filament_used_grams: float | None = None
|
|
print_time_seconds: int = 0
|
|
# Whether any queue item for this plate still exists, in any status.
|
|
# Dispatch clones one to inherit the print configuration, so a plate with
|
|
# none cannot be queued however many runs it still owes.
|
|
has_source: bool = False
|
|
|
|
@property
|
|
def dispatched(self) -> int:
|
|
return self.pending + self.printing + self.completed
|
|
|
|
@property
|
|
def remaining(self) -> int:
|
|
return max(0, self.quantity_target - self.dispatched)
|
|
|
|
@property
|
|
def can_dispatch(self) -> bool:
|
|
"""True when this plate owes runs *and* something can produce them."""
|
|
return self.remaining > 0 and self.has_source
|
|
|
|
@property
|
|
def cost_per_run(self) -> float | None:
|
|
"""Observed mean cost of this plate's completed runs, or None.
|
|
|
|
Deliberately measured rather than estimated from the file: the file's
|
|
estimate ignores what the run actually consumed, and a plate that has
|
|
never completed has no honest number to show.
|
|
"""
|
|
if self.completed <= 0 or self.actual_cost is None:
|
|
return None
|
|
return self.actual_cost / self.completed
|
|
|
|
@property
|
|
def estimated_remaining_cost(self) -> float | None:
|
|
per_run = self.cost_per_run
|
|
if per_run is None:
|
|
return None
|
|
return per_run * self.remaining
|
|
|
|
|
|
@dataclass
|
|
class BatchProgress:
|
|
"""Whole-order roll-up, plus the per-plate breakdown it was derived from."""
|
|
|
|
plates: list[PlateProgress] = field(default_factory=list)
|
|
has_targets: bool = False
|
|
|
|
def _sum(self, attr: str) -> int:
|
|
return sum(getattr(p, attr) for p in self.plates)
|
|
|
|
@property
|
|
def pending(self) -> int:
|
|
return self._sum("pending")
|
|
|
|
@property
|
|
def printing(self) -> int:
|
|
return self._sum("printing")
|
|
|
|
@property
|
|
def completed(self) -> int:
|
|
return self._sum("completed")
|
|
|
|
@property
|
|
def failed(self) -> int:
|
|
return self._sum("failed")
|
|
|
|
@property
|
|
def cancelled(self) -> int:
|
|
return self._sum("cancelled")
|
|
|
|
@property
|
|
def skipped(self) -> int:
|
|
return self._sum("skipped")
|
|
|
|
@property
|
|
def target(self) -> int:
|
|
return self._sum("quantity_target")
|
|
|
|
@property
|
|
def remaining(self) -> int:
|
|
return self._sum("remaining")
|
|
|
|
@property
|
|
def dispatchable_remaining(self) -> int:
|
|
"""Of the runs still owed, how many can actually be queued.
|
|
|
|
Lower than ``remaining`` when a plate's last queue item was deleted:
|
|
the order still owes the run, but nothing survives to clone its
|
|
printer target, AMS mapping and print options from.
|
|
"""
|
|
return sum(p.remaining for p in self.plates if p.has_source)
|
|
|
|
@property
|
|
def actual_cost(self) -> float | None:
|
|
costs = [p.actual_cost for p in self.plates if p.actual_cost is not None]
|
|
return sum(costs) if costs else None
|
|
|
|
@property
|
|
def estimated_remaining_cost(self) -> float | None:
|
|
estimates = [p.estimated_remaining_cost for p in self.plates if p.estimated_remaining_cost is not None]
|
|
return sum(estimates) if estimates else None
|
|
|
|
@property
|
|
def filament_used_grams(self) -> float | None:
|
|
grams = [p.filament_used_grams for p in self.plates if p.filament_used_grams is not None]
|
|
return sum(grams) if grams else None
|
|
|
|
@property
|
|
def print_time_seconds(self) -> int:
|
|
return self._sum("print_time_seconds")
|
|
|
|
@property
|
|
def is_fulfilled(self) -> bool:
|
|
"""True when every target is met and nothing is still in flight.
|
|
|
|
A zero total target is never "fulfilled". Without that guard a legacy
|
|
batch whose items were all cancelled one by one would report itself
|
|
completed — its derived target counts only pending/printing/completed
|
|
items, so cancelling the lot leaves a target of zero that trivially
|
|
satisfies ``remaining == 0``.
|
|
"""
|
|
return self.target > 0 and self.remaining == 0 and self.pending == 0 and self.printing == 0
|
|
|
|
|
|
async def load_progress(db: AsyncSession, batch: PrintBatch) -> BatchProgress:
|
|
"""Build the per-plate progress roll-up for *batch*.
|
|
|
|
Two queries plus one for costs, regardless of how many plates the order
|
|
has — this runs once per batch in the list endpoint.
|
|
"""
|
|
plate_rows = (await db.execute(select(PrintBatchPlate).where(PrintBatchPlate.batch_id == batch.id))).scalars().all()
|
|
|
|
# (plate_id, status) -> count, plus the time/weight actually recorded.
|
|
item_rows = (
|
|
await db.execute(
|
|
select(
|
|
PrintQueueItem.plate_id,
|
|
PrintQueueItem.status,
|
|
func.count(PrintQueueItem.id),
|
|
func.sum(PrintQueueItem.print_time_seconds),
|
|
)
|
|
.where(PrintQueueItem.batch_id == batch.id)
|
|
.group_by(PrintQueueItem.plate_id, PrintQueueItem.status)
|
|
)
|
|
).all()
|
|
|
|
# Per-run actuals, attributed through the queue item that produced them.
|
|
# PrintLogEntry is the authoritative per-run record (#1378) and is already
|
|
# scoped to the printed plate (#2614), so a multi-plate order gets each
|
|
# plate's own cost rather than the whole file's.
|
|
cost_rows = (
|
|
await db.execute(
|
|
select(
|
|
PrintQueueItem.plate_id,
|
|
func.sum(func.coalesce(PrintLogEntry.cost, 0.0) + func.coalesce(PrintLogEntry.energy_cost, 0.0)),
|
|
func.sum(PrintLogEntry.filament_used_grams),
|
|
)
|
|
.select_from(PrintLogEntry)
|
|
.join(PrintQueueItem, PrintLogEntry.queue_item_id == PrintQueueItem.id)
|
|
.where(PrintQueueItem.batch_id == batch.id)
|
|
.group_by(PrintQueueItem.plate_id)
|
|
)
|
|
).all()
|
|
costs = {row[0]: (row[1], row[2]) for row in cost_rows}
|
|
|
|
progress = BatchProgress(has_targets=bool(plate_rows))
|
|
by_plate: dict[int | None, PlateProgress] = {}
|
|
|
|
for row in plate_rows:
|
|
by_plate[row.plate_id] = PlateProgress(
|
|
plate_id=row.plate_id,
|
|
plate_name=row.plate_name,
|
|
quantity_target=row.quantity_target,
|
|
sort_order=row.sort_order,
|
|
)
|
|
|
|
for plate_id, status, count, time_sum in item_rows:
|
|
plate = by_plate.get(plate_id)
|
|
if plate is None:
|
|
# A queue item for a plate the order has no target row for: either
|
|
# a legacy batch, or an item grouped in by hand after the fact.
|
|
# Its own dispatched count becomes its target so it reads as
|
|
# complete rather than as owing work nobody asked for.
|
|
plate = PlateProgress(plate_id=plate_id, plate_name=None, quantity_target=0, sort_order=plate_id or 0)
|
|
by_plate[plate_id] = plate
|
|
if status in CONSUMING_STATUSES:
|
|
plate.quantity_target += count
|
|
elif not progress.has_targets and status in CONSUMING_STATUSES:
|
|
plate.quantity_target += count
|
|
# Any surviving row is a clone source, whatever its status — a
|
|
# cancelled or failed run still carries the configuration a re-queue
|
|
# needs. Set before the status filter below so a status this module
|
|
# has no counter for still marks the plate dispatchable.
|
|
plate.has_source = True
|
|
if status in COUNTED_STATUSES:
|
|
setattr(plate, status, getattr(plate, status) + count)
|
|
else:
|
|
logger.debug("Batch %s: ignoring queue item status %r in progress roll-up", batch.id, status)
|
|
plate.print_time_seconds += int(time_sum or 0)
|
|
|
|
for plate_id, (cost_sum, gram_sum) in costs.items():
|
|
plate = by_plate.get(plate_id)
|
|
if plate is None:
|
|
continue
|
|
plate.actual_cost = float(cost_sum) if cost_sum else None
|
|
plate.filament_used_grams = float(gram_sum) if gram_sum else None
|
|
|
|
progress.plates = sorted(by_plate.values(), key=lambda p: (p.sort_order, p.plate_id or 0))
|
|
return progress
|
|
|
|
|
|
async def refresh_batch_status(db: AsyncSession, batch: PrintBatch) -> bool:
|
|
"""Flip an ``active`` batch to ``completed`` once its targets are met.
|
|
|
|
Returns True when the status changed. A ``cancelled`` batch is never
|
|
resurrected, and a ``completed`` batch drops back to ``active`` if its
|
|
targets grow — raising a target on a finished order reopens it rather than
|
|
leaving a "completed" order that still owes prints.
|
|
"""
|
|
progress = await load_progress(db, batch)
|
|
|
|
if batch.status == "cancelled":
|
|
return False
|
|
|
|
if batch.status == "active" and progress.is_fulfilled:
|
|
batch.status = "completed"
|
|
batch.completed_at = datetime.now(timezone.utc)
|
|
logger.info("Batch %s fulfilled — marked completed", batch.id)
|
|
return True
|
|
|
|
# A grouping whose every item was cancelled one at a time is finished, but
|
|
# nothing was produced, so "completed" would be a lie and `is_fulfilled`
|
|
# rightly refuses it (its derived target is zero). Left alone it would sit
|
|
# on "active" forever. Cancelled is what it is, and matches what the
|
|
# batch-level Cancel action would have set had it been used.
|
|
#
|
|
# Deliberately not applied to orders: an order states its intent
|
|
# independently of its runs, so cancelling every run still leaves it owing
|
|
# work and offering to re-queue it. A grouping has no such statement — it
|
|
# was only ever the sum of its items.
|
|
if batch.status == "active" and not progress.has_targets and progress.completed == 0:
|
|
settled = progress.pending == 0 and progress.printing == 0
|
|
if settled and progress.cancelled > 0 and progress.failed == 0 and progress.skipped == 0:
|
|
batch.status = "cancelled"
|
|
logger.info("Batch %s had every item cancelled — marked cancelled", batch.id)
|
|
return True
|
|
|
|
if batch.status == "completed" and not progress.is_fulfilled:
|
|
batch.status = "active"
|
|
batch.completed_at = None
|
|
logger.info("Batch %s reopened — targets no longer met", batch.id)
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
async def backfill_batch_statuses(db: AsyncSession) -> int:
|
|
"""Close out ``active`` batches that finished before the status existed.
|
|
|
|
``completed`` only became reachable with #342. Every batch created since
|
|
the feature shipped in April 2026 is therefore still marked ``active``,
|
|
however long ago its last run finished — so without this pass the Batches
|
|
tab opens on months of accumulated history.
|
|
|
|
Runs on every startup rather than once behind a marker: it is cheap (only
|
|
batches with nothing in flight are even considered), it is idempotent, and
|
|
repeating it also closes out any order whose last run landed while the
|
|
process was down.
|
|
|
|
Returns the number of batches whose status changed.
|
|
"""
|
|
candidates = (
|
|
(
|
|
await db.execute(
|
|
select(PrintBatch)
|
|
.where(PrintBatch.status == "active")
|
|
# Anything still queued or printing is by definition unfinished,
|
|
# and re-deriving its progress would change nothing.
|
|
.where(
|
|
~select(PrintQueueItem.id)
|
|
.where(PrintQueueItem.batch_id == PrintBatch.id)
|
|
.where(PrintQueueItem.status.in_(("pending", "printing")))
|
|
.exists()
|
|
)
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
|
|
changed = 0
|
|
for batch in candidates:
|
|
if await refresh_batch_status(db, batch):
|
|
changed += 1
|
|
|
|
if changed:
|
|
await db.commit()
|
|
logger.info("Marked %d finished batch(es) as completed at startup (#342)", changed)
|
|
return changed
|
|
|
|
|
|
async def refresh_batch_status_for_item(db: AsyncSession, queue_item_id: int) -> None:
|
|
"""Re-evaluate the batch owning *queue_item_id*, if it has one.
|
|
|
|
Called from the print-completion path so a finished order reports itself
|
|
complete the moment its last run lands, rather than whenever someone next
|
|
opens the page.
|
|
"""
|
|
batch_id = (
|
|
await db.execute(select(PrintQueueItem.batch_id).where(PrintQueueItem.id == queue_item_id))
|
|
).scalar_one_or_none()
|
|
if batch_id is None:
|
|
return
|
|
batch = (await db.execute(select(PrintBatch).where(PrintBatch.id == batch_id))).scalar_one_or_none()
|
|
if batch is None:
|
|
return
|
|
await refresh_batch_status(db, batch)
|
|
|
|
|
|
async def _next_position(db: AsyncSession, printer_id: int | None) -> int:
|
|
"""Next free queue position in the scope a clone will land in.
|
|
|
|
Positions are per-queue, not global: one sequence per printer plus one
|
|
shared sequence for unassigned / model-based items, matching the scope the
|
|
add-to-queue route uses. Taking a global MAX here would drop every clone
|
|
at the end of whichever printer's queue happens to be longest and scramble
|
|
the order the user sees.
|
|
"""
|
|
# Same advisory lock the add-to-queue route takes (#1625-followup): two
|
|
# concurrent inserts into an empty scope would otherwise both read
|
|
# MAX(position) as 0 and land on position 1. SQLite serialises writes
|
|
# implicitly and needs no equivalent.
|
|
bind = db.get_bind()
|
|
if bind.dialect.name == "postgresql":
|
|
await db.execute(
|
|
text("SELECT pg_advisory_xact_lock(1625, :k)"), {"k": printer_id if printer_id is not None else 0}
|
|
)
|
|
|
|
scope = PrintQueueItem.printer_id == printer_id if printer_id is not None else PrintQueueItem.printer_id.is_(None)
|
|
max_pos = (
|
|
await db.execute(
|
|
select(func.max(PrintQueueItem.position)).where(scope).where(PrintQueueItem.status == "pending")
|
|
)
|
|
).scalar() or 0
|
|
return max_pos + 1
|
|
|
|
|
|
def _clone_queue_item(source: PrintQueueItem, *, position: int, created_by_id: int | None) -> PrintQueueItem:
|
|
"""Copy *source*'s print configuration into a fresh pending item.
|
|
|
|
Lifecycle state (status, timestamps, retry counters, scheduler flags) is
|
|
deliberately not copied — the clone is a new run, not a resurrection.
|
|
|
|
``scheduled_time`` is dropped too: dispatching more of a plate is a
|
|
"queue this now" action, and replaying the original's scheduled time would
|
|
either fire immediately (it is in the past) or silently park the new run
|
|
until a moment the user chose for a different print.
|
|
|
|
``cleanup_library_after_dispatch`` is forced off. It only ever comes from
|
|
the Printers-page direct-print flow, where it deletes the transient library
|
|
row after dispatch — replaying that on a clone would delete the source file
|
|
out from under the rest of the order.
|
|
"""
|
|
clone = PrintQueueItem(
|
|
status="pending",
|
|
position=position,
|
|
created_by_id=created_by_id if created_by_id is not None else source.created_by_id,
|
|
cleanup_library_after_dispatch=False,
|
|
)
|
|
for column in CLONED_SETTING_COLUMNS:
|
|
setattr(clone, column, getattr(source, column))
|
|
return clone
|
|
|
|
|
|
def _plate_label(plate: PlateProgress) -> str:
|
|
"""How a plate is named in an error the user reads.
|
|
|
|
Prefers the plate name the order stored, because that is what the Batches
|
|
tab shows; falls back to the plate number for orders created before names
|
|
were recorded.
|
|
"""
|
|
if plate.plate_name:
|
|
return plate.plate_name
|
|
return f"Plate {plate.plate_id if plate.plate_id is not None else 1}"
|
|
|
|
|
|
async def dispatch_remaining(
|
|
db: AsyncSession,
|
|
batch: PrintBatch,
|
|
*,
|
|
plate_id: int | None = None,
|
|
only_plate: bool = False,
|
|
limit: int | None = None,
|
|
created_by_id: int | None = None,
|
|
) -> list[PrintQueueItem]:
|
|
"""Create queue items for the runs *batch* still owes.
|
|
|
|
``only_plate`` restricts the dispatch to the single plate named by
|
|
``plate_id`` (which may legitimately be ``None`` for a single-plate file);
|
|
otherwise every plate with work outstanding is dispatched in plate order.
|
|
``limit`` caps the total number of items created across all plates.
|
|
|
|
Raises :class:`BatchDispatchError` when nothing at all can be produced
|
|
because no plate that owes runs has an item to clone — the order can
|
|
describe work whose every run has since been deleted, and there is no
|
|
configuration to copy in that case. A plate in that state is skipped
|
|
rather than aborting the whole order: one unrecoverable plate must not
|
|
block the plates that are still perfectly dispatchable.
|
|
"""
|
|
progress = await load_progress(db, batch)
|
|
if not progress.has_targets:
|
|
return []
|
|
|
|
targets = [p for p in progress.plates if p.remaining > 0]
|
|
if only_plate:
|
|
targets = [p for p in targets if p.plate_id == plate_id]
|
|
|
|
created: list[PrintQueueItem] = []
|
|
stranded: list[PlateProgress] = []
|
|
|
|
for plate in targets:
|
|
if limit is not None and len(created) >= limit:
|
|
break
|
|
|
|
source = (
|
|
await db.execute(
|
|
select(PrintQueueItem)
|
|
.options(selectinload(PrintQueueItem.variants))
|
|
.where(PrintQueueItem.batch_id == batch.id)
|
|
.where(PrintQueueItem.plate_id == plate.plate_id)
|
|
.order_by(PrintQueueItem.id.desc())
|
|
.limit(1)
|
|
)
|
|
).scalar_one_or_none()
|
|
|
|
if source is None:
|
|
stranded.append(plate)
|
|
continue
|
|
|
|
wanted = plate.remaining
|
|
if limit is not None:
|
|
wanted = min(wanted, limit - len(created))
|
|
|
|
# One scope per source printer; clones for this plate all land in it,
|
|
# appended after whatever is already queued there.
|
|
position = await _next_position(db, source.printer_id)
|
|
|
|
for _ in range(wanted):
|
|
clone = _clone_queue_item(source, position=position, created_by_id=created_by_id)
|
|
position += 1
|
|
db.add(clone)
|
|
await db.flush()
|
|
for variant in source.variants:
|
|
cloned_variant = PrintQueueVariant(queue_item_id=clone.id)
|
|
for column in CLONED_VARIANT_COLUMNS:
|
|
setattr(cloned_variant, column, getattr(variant, column))
|
|
db.add(cloned_variant)
|
|
created.append(clone)
|
|
|
|
if not created and stranded:
|
|
names = ", ".join(_plate_label(p) for p in stranded)
|
|
raise BatchDispatchError(
|
|
f"{names} {'have' if len(stranded) > 1 else 'has'} no queued or finished run to copy settings from. "
|
|
"Queue the plate once from the file, then dispatch the rest from here."
|
|
)
|
|
|
|
if created:
|
|
# Dispatching more work can only ever un-fulfil an order, but run the
|
|
# check anyway so a reopened batch flips back from completed.
|
|
await db.flush()
|
|
await refresh_batch_status(db, batch)
|
|
|
|
if stranded:
|
|
logger.info("Batch %s: skipped %d plate(s) with no item to clone", batch.id, len(stranded))
|
|
logger.info("Dispatched %d item(s) for batch %s", len(created), batch.id)
|
|
return created
|