|
@@ -11,6 +11,7 @@ from contextlib import asynccontextmanager
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from datetime import datetime, timedelta, timezone
|
|
|
from logging.handlers import RotatingFileHandler
|
|
from logging.handlers import RotatingFileHandler
|
|
|
from pathlib import Path, PurePosixPath
|
|
from pathlib import Path, PurePosixPath
|
|
|
|
|
+from typing import cast
|
|
|
from urllib.parse import urlparse
|
|
from urllib.parse import urlparse
|
|
|
|
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi import FastAPI
|
|
@@ -18,7 +19,8 @@ from fastapi.encoders import jsonable_encoder
|
|
|
from fastapi.exceptions import RequestValidationError
|
|
from fastapi.exceptions import RequestValidationError
|
|
|
from fastapi.responses import FileResponse, JSONResponse
|
|
from fastapi.responses import FileResponse, JSONResponse
|
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
-from sqlalchemy import delete, inspect as sa_inspect, or_, select, text, update
|
|
|
|
|
|
|
+from sqlalchemy import delete, func, inspect as sa_inspect, or_, select, text, update
|
|
|
|
|
+from sqlalchemy.engine import CursorResult
|
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
from sqlalchemy.orm.exc import StaleDataError
|
|
from sqlalchemy.orm.exc import StaleDataError
|
|
|
|
|
|
|
@@ -101,6 +103,7 @@ from backend.app.core.static_assets import AssetStaticFiles
|
|
|
from backend.app.core.tasks import spawn_background_task
|
|
from backend.app.core.tasks import spawn_background_task
|
|
|
from backend.app.core.websocket import ws_manager
|
|
from backend.app.core.websocket import ws_manager
|
|
|
from backend.app.services import kprofile_drift, print_dispatch_context, slot_unlink_grace
|
|
from backend.app.services import kprofile_drift, print_dispatch_context, slot_unlink_grace
|
|
|
|
|
+from backend.app.services.ams_assignment_lock import get_ams_assignment_lock
|
|
|
from backend.app.services.archive import ArchiveService, peek_plate_index_in_3mf, swap_plate_suffix
|
|
from backend.app.services.archive import ArchiveService, peek_plate_index_in_3mf, swap_plate_suffix
|
|
|
from backend.app.services.archive_cost_estimate import (
|
|
from backend.app.services.archive_cost_estimate import (
|
|
|
schedule_archive_cost_estimate as _schedule_archive_cost_estimate,
|
|
schedule_archive_cost_estimate as _schedule_archive_cost_estimate,
|
|
@@ -706,35 +709,6 @@ def derive_failure_reason(status: str, hms_errors: list[dict] | None) -> str | N
|
|
|
# {(printer_id, filename): created_by_id}
|
|
# {(printer_id, filename): created_by_id}
|
|
|
_expected_print_creators: dict[tuple[int, str], int] = {}
|
|
_expected_print_creators: dict[tuple[int, str], int] = {}
|
|
|
|
|
|
|
|
-# Per-printer lock that serialises the spool-assignment side of on_ams_change
|
|
|
|
|
-# (auto-unlink stale + auto-assign new) when MQTT bursts deliver multiple AMS
|
|
|
|
|
-# updates for the same printer in quick succession (~30 ms apart, observed in
|
|
|
|
|
-# the wild on H2D + dual AMS).
|
|
|
|
|
-#
|
|
|
|
|
-# Without this serialisation, two concurrent on_ams_change callbacks each read
|
|
|
|
|
-# "no assignment for (printer, ams, tray)", each call auto_assign_spool, and
|
|
|
|
|
-# the second commit hits
|
|
|
|
|
-# IntegrityError: duplicate key value violates unique constraint
|
|
|
|
|
-# "spool_assignment_printer_id_ams_id_tray_id_key"
|
|
|
|
|
-# SQLite's WAL serial-write semantics had been silently swallowing the race
|
|
|
|
|
-# until optional Postgres support landed (asyncpg allows true concurrent
|
|
|
|
|
-# transactions and surfaces the constraint violation).
|
|
|
|
|
-#
|
|
|
|
|
-# Scope is intentionally narrow: only the two DB-mutating blocks (unlink +
|
|
|
|
|
-# assign) are inside the lock. The Spoolman sync block further down stays
|
|
|
|
|
-# concurrent because it's network-bound and idempotent.
|
|
|
|
|
-_ams_assignment_locks: dict[int, asyncio.Lock] = {}
|
|
|
|
|
-
|
|
|
|
|
-
|
|
|
|
|
-def _get_ams_assignment_lock(printer_id: int) -> asyncio.Lock:
|
|
|
|
|
- """Return the per-printer assignment lock, creating it on first use."""
|
|
|
|
|
- lock = _ams_assignment_locks.get(printer_id)
|
|
|
|
|
- if lock is None:
|
|
|
|
|
- lock = asyncio.Lock()
|
|
|
|
|
- _ams_assignment_locks[printer_id] = lock
|
|
|
|
|
- return lock
|
|
|
|
|
-
|
|
|
|
|
-
|
|
|
|
|
# Per-printer dedup for unknown_tag WS broadcasts. Keyed by
|
|
# Per-printer dedup for unknown_tag WS broadcasts. Keyed by
|
|
|
# (ams_id, tray_id) -> (tag_uid, tray_uuid); we only re-broadcast when the
|
|
# (ams_id, tray_id) -> (tag_uid, tray_uuid); we only re-broadcast when the
|
|
|
# tag tuple changes for the slot. Cleared when the slot is reported empty
|
|
# tag tuple changes for the slot. Cleared when the slot is reported empty
|
|
@@ -1257,16 +1231,25 @@ def register_expected_print(
|
|
|
base = filename[:-4]
|
|
base = filename[:-4]
|
|
|
_expected_prints[(printer_id, base)] = archive_id
|
|
_expected_prints[(printer_id, base)] = archive_id
|
|
|
_expected_prints[(printer_id, f"{base}.gcode")] = archive_id
|
|
_expected_prints[(printer_id, f"{base}.gcode")] = archive_id
|
|
|
- # Store AMS mapping for usage tracking at print completion
|
|
|
|
|
|
|
+ # Store AMS mapping for usage tracking at print completion, and plate_id for
|
|
|
|
|
+ # a single-plate dispatch from a multi-plate 3MF — without it, the direct-Print
|
|
|
|
|
+ # path attributes the whole file's filament total to the spool instead of just
|
|
|
|
|
+ # the printed plate (#1697). They describe this dispatch: one that has none
|
|
|
|
|
+ # drops what an earlier dispatch of the archive left (sent, never started),
|
|
|
|
|
+ # unless a plate of the archive is printing now and still needs them.
|
|
|
|
|
+ archive_printing = archive_id in set(_active_prints.values())
|
|
|
if ams_mapping is not None:
|
|
if ams_mapping is not None:
|
|
|
_print_ams_mappings[archive_id] = ams_mapping
|
|
_print_ams_mappings[archive_id] = ams_mapping
|
|
|
|
|
+ elif not archive_printing:
|
|
|
|
|
+ _print_ams_mappings.pop(archive_id, None)
|
|
|
if cost_center_id is not None:
|
|
if cost_center_id is not None:
|
|
|
_print_cost_center_ids[archive_id] = cost_center_id
|
|
_print_cost_center_ids[archive_id] = cost_center_id
|
|
|
- # Store plate_id for usage tracking when this is a single-plate dispatch from
|
|
|
|
|
- # a multi-plate 3MF — without this, the direct-Print path attributes the whole
|
|
|
|
|
- # file's filament total to the spool instead of just the printed plate (#1697).
|
|
|
|
|
|
|
+ elif not archive_printing:
|
|
|
|
|
+ _print_cost_center_ids.pop(archive_id, None)
|
|
|
if plate_id is not None:
|
|
if plate_id is not None:
|
|
|
_print_plate_ids[archive_id] = plate_id
|
|
_print_plate_ids[archive_id] = plate_id
|
|
|
|
|
+ elif not archive_printing:
|
|
|
|
|
+ _print_plate_ids.pop(archive_id, None)
|
|
|
# Store created_by_id so the user start email can be sent even when the archive
|
|
# Store created_by_id so the user start email can be sent even when the archive
|
|
|
# itself has no created_by_id (e.g. library-file-based queue prints)
|
|
# itself has no created_by_id (e.g. library-file-based queue prints)
|
|
|
if created_by_id is not None:
|
|
if created_by_id is not None:
|
|
@@ -1319,7 +1302,9 @@ def unregister_expected_print(printer_id: int, filename: str, archive_id: int) -
|
|
|
_expected_print_creators.pop(key, None)
|
|
_expected_print_creators.pop(key, None)
|
|
|
_expected_print_registered_at.pop(key, None)
|
|
_expected_print_registered_at.pop(key, None)
|
|
|
|
|
|
|
|
- if archive_id not in set(_expected_prints.values()):
|
|
|
|
|
|
|
+ # Still needed by another expected print of the archive, or a plate of it
|
|
|
|
|
+ # that is printing, as in the sweep.
|
|
|
|
|
+ if archive_id not in set(_expected_prints.values()) | set(_active_prints.values()):
|
|
|
_print_ams_mappings.pop(archive_id, None)
|
|
_print_ams_mappings.pop(archive_id, None)
|
|
|
_print_cost_center_ids.pop(archive_id, None)
|
|
_print_cost_center_ids.pop(archive_id, None)
|
|
|
_print_plate_ids.pop(archive_id, None)
|
|
_print_plate_ids.pop(archive_id, None)
|
|
@@ -2442,7 +2427,7 @@ async def _unlink_stale_assignments(printer_id: int, ams_data: list, printing_no
|
|
|
# per-printer assignment lock since #3186: the held-removal re-check runs
|
|
# per-printer assignment lock since #3186: the held-removal re-check runs
|
|
|
# this outside any MQTT push, so it can now overlap one.
|
|
# this outside any MQTT push, so it can now overlap one.
|
|
|
try:
|
|
try:
|
|
|
- async with _get_ams_assignment_lock(printer_id), async_session() as db:
|
|
|
|
|
|
|
+ async with get_ams_assignment_lock(printer_id), async_session() as db:
|
|
|
from sqlalchemy.orm import selectinload
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
|
|
|
|
from backend.app.api.routes.inventory import _find_tray_in_ams_data
|
|
from backend.app.api.routes.inventory import _find_tray_in_ams_data
|
|
@@ -2594,12 +2579,18 @@ async def _unlink_stale_assignments(printer_id: int, ams_data: list, printing_no
|
|
|
cur_state == 11 or (cur_state not in (9, 10) and cur_type.strip())
|
|
cur_state == 11 or (cur_state not in (9, 10) and cur_type.strip())
|
|
|
)
|
|
)
|
|
|
if not fp_type.strip() and loaded and assignment.spool:
|
|
if not fp_type.strip() and loaded and assignment.spool:
|
|
|
|
|
+ # The fingerprint ends this replay. Not recorded when nothing
|
|
|
|
|
+ # could be sent (no MQTT client while the printer reconnects),
|
|
|
|
|
+ # so the next AMS change tries again. An attempt that raised is
|
|
|
|
|
+ # recorded as before: it may have sent part of the commands and
|
|
|
|
|
+ # would raise again on every change, remain% steps included.
|
|
|
|
|
+ configured: bool | None = False
|
|
|
try:
|
|
try:
|
|
|
from backend.app.api.routes.inventory import (
|
|
from backend.app.api.routes.inventory import (
|
|
|
apply_spool_to_slot_via_mqtt,
|
|
apply_spool_to_slot_via_mqtt,
|
|
|
)
|
|
)
|
|
|
|
|
|
|
|
- await apply_spool_to_slot_via_mqtt(
|
|
|
|
|
|
|
+ configured = await apply_spool_to_slot_via_mqtt(
|
|
|
db=db,
|
|
db=db,
|
|
|
current_user=None,
|
|
current_user=None,
|
|
|
spool=assignment.spool,
|
|
spool=assignment.spool,
|
|
@@ -2610,23 +2601,35 @@ async def _unlink_stale_assignments(printer_id: int, ams_data: list, printing_no
|
|
|
current_tray_type=cur_type,
|
|
current_tray_type=cur_type,
|
|
|
current_tray_sub_brands=current_tray.get("tray_sub_brands", ""),
|
|
current_tray_sub_brands=current_tray.get("tray_sub_brands", ""),
|
|
|
)
|
|
)
|
|
|
- logger.info(
|
|
|
|
|
- "SpoolBuddy pre-config applied on insert: spool %d → printer %d AMS%d-T%d",
|
|
|
|
|
|
|
+ except Exception:
|
|
|
|
|
+ logger.exception(
|
|
|
|
|
+ "Pre-config apply failed for spool %d on printer %d AMS%d-T%d",
|
|
|
assignment.spool_id,
|
|
assignment.spool_id,
|
|
|
printer_id,
|
|
printer_id,
|
|
|
assignment.ams_id,
|
|
assignment.ams_id,
|
|
|
assignment.tray_id,
|
|
assignment.tray_id,
|
|
|
)
|
|
)
|
|
|
- except Exception:
|
|
|
|
|
- logger.exception(
|
|
|
|
|
- "Pre-config apply failed for spool %d on printer %d AMS%d-T%d",
|
|
|
|
|
|
|
+ configured = None
|
|
|
|
|
+ if configured is False:
|
|
|
|
|
+ logger.warning(
|
|
|
|
|
+ "Pre-config not sent for spool %d on printer %d AMS%d-T%d (no MQTT client); "
|
|
|
|
|
+ "retrying on the next AMS change",
|
|
|
assignment.spool_id,
|
|
assignment.spool_id,
|
|
|
printer_id,
|
|
printer_id,
|
|
|
assignment.ams_id,
|
|
assignment.ams_id,
|
|
|
assignment.tray_id,
|
|
assignment.tray_id,
|
|
|
)
|
|
)
|
|
|
- assignment.fingerprint_color = cur_color
|
|
|
|
|
- assignment.fingerprint_type = cur_type
|
|
|
|
|
|
|
+ else:
|
|
|
|
|
+ if configured:
|
|
|
|
|
+ logger.info(
|
|
|
|
|
+ "SpoolBuddy pre-config applied on insert: spool %d → printer %d AMS%d-T%d",
|
|
|
|
|
+ assignment.spool_id,
|
|
|
|
|
+ printer_id,
|
|
|
|
|
+ assignment.ams_id,
|
|
|
|
|
+ assignment.tray_id,
|
|
|
|
|
+ )
|
|
|
|
|
+ assignment.fingerprint_color = cur_color
|
|
|
|
|
+ assignment.fingerprint_type = cur_type
|
|
|
continue
|
|
continue
|
|
|
|
|
|
|
|
if not _colors_similar(cur_color, fp_color) or cur_type.upper() != fp_type.upper():
|
|
if not _colors_similar(cur_color, fp_color) or cur_type.upper() != fp_type.upper():
|
|
@@ -2903,11 +2906,12 @@ async def on_ams_change(printer_id: int, ams_data: list):
|
|
|
"""Handle AMS data changes - sync to Spoolman if enabled and auto mode."""
|
|
"""Handle AMS data changes - sync to Spoolman if enabled and auto mode."""
|
|
|
logger = logging.getLogger(__name__)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
|
|
- # Snapshot BEFORE any await: if a print is active, skip weight sync later.
|
|
|
|
|
- # on_print_complete may pop _active_sessions during our awaits (#880).
|
|
|
|
|
- from backend.app.services.usage_tracker import _active_sessions
|
|
|
|
|
|
|
+ # Snapshot BEFORE any await: if a print is active, or a finished one is
|
|
|
|
|
+ # still being charged, skip weight sync later. on_print_complete may pop
|
|
|
|
|
+ # _active_sessions during our awaits (#880).
|
|
|
|
|
+ from backend.app.services.usage_tracker import printer_is_charging
|
|
|
|
|
|
|
|
- _print_active = printer_id in _active_sessions
|
|
|
|
|
|
|
+ _print_active = printer_is_charging(printer_id)
|
|
|
|
|
|
|
|
# A slot that reports empty while a print is running is a filament runout,
|
|
# A slot that reports empty while a print is running is a filament runout,
|
|
|
# not a spool swap: the spool is still physically in the AMS, just
|
|
# not a spool swap: the spool is still physically in the AMS, just
|
|
@@ -2950,14 +2954,14 @@ async def on_ams_change(printer_id: int, ams_data: list):
|
|
|
await _unlink_stale_assignments(printer_id, ams_data, printing_now)
|
|
await _unlink_stale_assignments(printer_id, ams_data, printing_now)
|
|
|
|
|
|
|
|
# Auto-manage inventory spools from AMS tray data (skip if Spoolman manages AMS).
|
|
# Auto-manage inventory spools from AMS tray data (skip if Spoolman manages AMS).
|
|
|
- # Serialised per-printer via _ams_assignment_locks: MQTT bursts can deliver
|
|
|
|
|
|
|
+ # Serialised per-printer via services/ams_assignment_lock: MQTT bursts can deliver
|
|
|
# two AMS pushes ~30 ms apart, and without the lock both callbacks read
|
|
# two AMS pushes ~30 ms apart, and without the lock both callbacks read
|
|
|
# "no existing assignment" for the same (printer, ams, tray) and race to
|
|
# "no existing assignment" for the same (printer, ams, tray) and race to
|
|
|
# INSERT, hitting the spool_assignment_printer_id_ams_id_tray_id_key
|
|
# INSERT, hitting the spool_assignment_printer_id_ams_id_tray_id_key
|
|
|
# unique constraint on Postgres. SQLite's WAL serialises writes so the
|
|
# unique constraint on Postgres. SQLite's WAL serialises writes so the
|
|
|
- # bug stayed latent there. See _ams_assignment_locks comment for details.
|
|
|
|
|
|
|
+ # bug stayed latent there. See services/ams_assignment_lock for details.
|
|
|
try:
|
|
try:
|
|
|
- async with _get_ams_assignment_lock(printer_id), async_session() as db:
|
|
|
|
|
|
|
+ async with get_ams_assignment_lock(printer_id), async_session() as db:
|
|
|
from sqlalchemy.orm import selectinload
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
|
|
|
|
from backend.app.api.routes.settings import get_setting
|
|
from backend.app.api.routes.settings import get_setting
|
|
@@ -3026,15 +3030,31 @@ async def on_ams_change(printer_id: int, ams_data: list):
|
|
|
new_used = round(lw * (100 - remain_val) / 100.0, 1)
|
|
new_used = round(lw * (100 - remain_val) / 100.0, 1)
|
|
|
current_used = existing_assignment.spool.weight_used or 0
|
|
current_used = existing_assignment.spool.weight_used or 0
|
|
|
if new_used > current_used + 1:
|
|
if new_used > current_used + 1:
|
|
|
- logger.info(
|
|
|
|
|
- "Weight sync: spool %d weight_used %s -> %s (remain=%d)",
|
|
|
|
|
- existing_assignment.spool_id,
|
|
|
|
|
- current_used,
|
|
|
|
|
- new_used,
|
|
|
|
|
- remain_val,
|
|
|
|
|
|
|
+ # Rechecked in the statement: the spool was loaded
|
|
|
|
|
+ # before this, and a weight set by hand (or weighed)
|
|
|
|
|
+ # and locked since then must stay (WL-6).
|
|
|
|
|
+ result = cast(
|
|
|
|
|
+ CursorResult,
|
|
|
|
|
+ await db.execute(
|
|
|
|
|
+ update(Spool)
|
|
|
|
|
+ .where(
|
|
|
|
|
+ Spool.id == existing_assignment.spool_id,
|
|
|
|
|
+ Spool.weight_locked.is_not(True),
|
|
|
|
|
+ func.coalesce(Spool.weight_used, 0) < new_used - 1,
|
|
|
|
|
+ )
|
|
|
|
|
+ .values(weight_used=new_used)
|
|
|
|
|
+ ),
|
|
|
)
|
|
)
|
|
|
- existing_assignment.spool.weight_used = new_used
|
|
|
|
|
await db.commit()
|
|
await db.commit()
|
|
|
|
|
+ if result.rowcount:
|
|
|
|
|
+ logger.info(
|
|
|
|
|
+ "Weight sync: spool %d weight_used %s -> %s (remain=%d)",
|
|
|
|
|
+ existing_assignment.spool_id,
|
|
|
|
|
+ current_used,
|
|
|
|
|
+ new_used,
|
|
|
|
|
+ remain_val,
|
|
|
|
|
+ )
|
|
|
|
|
+ await ws_manager.broadcast({"type": "inventory_changed"})
|
|
|
|
|
|
|
|
# A K-profile the slot lost (the default K after a reset or
|
|
# A K-profile the slot lost (the default K after a reset or
|
|
|
# power cycle) is put back by kprofile_drift on the next idle
|
|
# power cycle) is put back by kprofile_drift on the next idle
|
|
@@ -6281,6 +6301,10 @@ async def _scan_for_timelapse_with_retries(
|
|
|
if not archive.printer_id:
|
|
if not archive.printer_id:
|
|
|
logger.warning("[TIMELAPSE] Archive %s has no printer, aborting", archive_id)
|
|
logger.warning("[TIMELAPSE] Archive %s has no printer, aborting", archive_id)
|
|
|
return
|
|
return
|
|
|
|
|
+ # The run this scan is for. The archive row passes to a newer run
|
|
|
|
|
+ # when it prints again (the next plate, a reprint), and that run's
|
|
|
|
|
+ # video is its own.
|
|
|
|
|
+ run_started_at = archive.started_at
|
|
|
|
|
|
|
|
if baseline_names is not None:
|
|
if baseline_names is not None:
|
|
|
logger.info(
|
|
logger.info(
|
|
@@ -6361,6 +6385,11 @@ async def _scan_for_timelapse_with_retries(
|
|
|
if archive.timelapse_path:
|
|
if archive.timelapse_path:
|
|
|
logger.info("[TIMELAPSE] Archive %s already has timelapse attached, stopping poll", archive_id)
|
|
logger.info("[TIMELAPSE] Archive %s already has timelapse attached, stopping poll", archive_id)
|
|
|
return
|
|
return
|
|
|
|
|
+ if archive.started_at != run_started_at:
|
|
|
|
|
+ logger.info(
|
|
|
|
|
+ "[TIMELAPSE] Archive %s is printing again, stopping the previous run's poll", archive_id
|
|
|
|
|
+ )
|
|
|
|
|
+ return
|
|
|
|
|
|
|
|
result = await db.execute(select(Printer).where(Printer.id == archive.printer_id))
|
|
result = await db.execute(select(Printer).where(Printer.id == archive.printer_id))
|
|
|
printer = result.scalar_one_or_none()
|
|
printer = result.scalar_one_or_none()
|
|
@@ -6398,6 +6427,7 @@ async def _scan_for_timelapse_with_retries(
|
|
|
quiet=not changed,
|
|
quiet=not changed,
|
|
|
require_unambiguous=not baseline_trusted,
|
|
require_unambiguous=not baseline_trusted,
|
|
|
plate_id=plate_id,
|
|
plate_id=plate_id,
|
|
|
|
|
+ run_started_at=run_started_at,
|
|
|
)
|
|
)
|
|
|
if attached:
|
|
if attached:
|
|
|
return
|
|
return
|
|
@@ -6421,6 +6451,55 @@ async def _scan_for_timelapse_with_retries(
|
|
|
)
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
+async def _attach_layer_timelapse(
|
|
|
|
|
+ archive_id: int, timelapse_path: Path, plate_id: int | None, run_started_at: datetime | None
|
|
|
|
|
+) -> None:
|
|
|
|
|
+ """Attach the external camera's stitched timelapse to the archive.
|
|
|
|
|
+
|
|
|
|
|
+ The camera's video is the one the user set up in Bambuddy, so it wins
|
|
|
|
|
+ over the printer's own when both are on: it replaces the printer's video
|
|
|
|
|
+ (usually there first; stitching takes minutes) and that file is deleted,
|
|
|
|
|
+ leaving nothing in the archive folder unreferenced. ``run_started_at`` is
|
|
|
|
|
+ the run the frames were taken for: when the archive is printing again by
|
|
|
|
|
+ the time the stitch is done, the video is not the new run's and is
|
|
|
|
|
+ dropped. The stitched file is a temporary either way.
|
|
|
|
|
+ """
|
|
|
|
|
+ from backend.app.utils.safe_path import PathTraversalError, safe_join_under
|
|
|
|
|
+
|
|
|
|
|
+ logger = logging.getLogger(__name__)
|
|
|
|
|
+ try:
|
|
|
|
|
+ async with async_session() as db:
|
|
|
|
|
+ service = ArchiveService(db)
|
|
|
|
|
+ archive = await service.get_archive(archive_id)
|
|
|
|
|
+ if archive is None:
|
|
|
|
|
+ return
|
|
|
|
|
+ if archive.started_at != run_started_at:
|
|
|
|
|
+ logger.info(
|
|
|
|
|
+ "[LAYER-TL] Archive %s is printing again; dropping the previous run's layer timelapse", archive_id
|
|
|
|
|
+ )
|
|
|
|
|
+ return
|
|
|
|
|
+ replaced = archive.timelapse_path
|
|
|
|
|
+ logger.info("[LAYER-TL] Attaching timelapse %s to archive %s", timelapse_path, archive_id)
|
|
|
|
|
+ timelapse_data = await asyncio.to_thread(timelapse_path.read_bytes)
|
|
|
|
|
+ if not await service.attach_timelapse(archive_id, timelapse_data, "layer_timelapse.mp4", plate_id=plate_id):
|
|
|
|
|
+ return
|
|
|
|
|
+ attached = archive.timelapse_path
|
|
|
|
|
+ logger.info("[LAYER-TL] Layer timelapse attached successfully")
|
|
|
|
|
+ if replaced and replaced != attached:
|
|
|
|
|
+ try:
|
|
|
|
|
+ old = safe_join_under(app_settings.base_dir, replaced, http=False)
|
|
|
|
|
+ except PathTraversalError:
|
|
|
|
|
+ return
|
|
|
|
|
+ await asyncio.to_thread(old.unlink, missing_ok=True)
|
|
|
|
|
+ logger.info("[LAYER-TL] Replaced the printer's timelapse %s on archive %s", replaced, archive_id)
|
|
|
|
|
+ finally:
|
|
|
|
|
+ await asyncio.to_thread(timelapse_path.unlink, missing_ok=True)
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+# ``run_started_at`` not given: attach whichever run the archive is on.
|
|
|
|
|
+_ANY_RUN = object()
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
async def _attach_first_unclaimed_timelapse(
|
|
async def _attach_first_unclaimed_timelapse(
|
|
|
archive_id: int,
|
|
archive_id: int,
|
|
|
printer,
|
|
printer,
|
|
@@ -6433,6 +6512,7 @@ async def _attach_first_unclaimed_timelapse(
|
|
|
quiet: bool = False,
|
|
quiet: bool = False,
|
|
|
require_unambiguous: bool = False,
|
|
require_unambiguous: bool = False,
|
|
|
plate_id: int | None = None,
|
|
plate_id: int | None = None,
|
|
|
|
|
+ run_started_at: datetime | None | object = _ANY_RUN,
|
|
|
) -> bool:
|
|
) -> bool:
|
|
|
"""Download and attach the one video that belongs to this print.
|
|
"""Download and attach the one video that belongs to this print.
|
|
|
|
|
|
|
@@ -6495,7 +6575,7 @@ async def _attach_first_unclaimed_timelapse(
|
|
|
)
|
|
)
|
|
|
|
|
|
|
|
target = candidates[0]
|
|
target = candidates[0]
|
|
|
- file_name = target.get("name")
|
|
|
|
|
|
|
+ file_name = str(target.get("name") or "") # every listed entry has one
|
|
|
remote_path = target.get("path") or f"/timelapse/{file_name}"
|
|
remote_path = target.get("path") or f"/timelapse/{file_name}"
|
|
|
logger.info(
|
|
logger.info(
|
|
|
"[TIMELAPSE] Attempt %s: New file detected: %s (downloading for archive %s)",
|
|
"[TIMELAPSE] Attempt %s: New file detected: %s (downloading for archive %s)",
|
|
@@ -6540,15 +6620,28 @@ async def _attach_first_unclaimed_timelapse(
|
|
|
):
|
|
):
|
|
|
return False
|
|
return False
|
|
|
|
|
|
|
|
- # Write phase: attach in a fresh short-lived session.
|
|
|
|
|
|
|
+ # Write phase: attach in a fresh short-lived session. The download took a
|
|
|
|
|
+ # while: by now the archive may have its video, or be printing again for a
|
|
|
|
|
+ # newer run, whose video this is not. Then nothing is attached or deleted,
|
|
|
|
|
+ # and the poll ends (True).
|
|
|
async with async_session() as db:
|
|
async with async_session() as db:
|
|
|
- success = await ArchiveService(db).attach_timelapse(archive_id, timelapse_data, file_name, plate_id=plate_id)
|
|
|
|
|
|
|
+ service = ArchiveService(db)
|
|
|
|
|
+ if run_started_at is not _ANY_RUN:
|
|
|
|
|
+ current = await service.get_archive(archive_id)
|
|
|
|
|
+ if current is None or current.timelapse_path or current.started_at != run_started_at:
|
|
|
|
|
+ logger.info(
|
|
|
|
|
+ "[TIMELAPSE] Archive %s moved on during the download (video attached or printing again); "
|
|
|
|
|
+ "%s stays on the printer",
|
|
|
|
|
+ archive_id,
|
|
|
|
|
+ file_name,
|
|
|
|
|
+ )
|
|
|
|
|
+ return True
|
|
|
|
|
+ success = await service.attach_timelapse(archive_id, timelapse_data, file_name, plate_id=plate_id)
|
|
|
if not success:
|
|
if not success:
|
|
|
logger.warning("[TIMELAPSE] Failed to attach timelapse to archive %s", archive_id)
|
|
logger.warning("[TIMELAPSE] Failed to attach timelapse to archive %s", archive_id)
|
|
|
return False
|
|
return False
|
|
|
|
|
|
|
|
logger.info("[TIMELAPSE] Successfully attached timelapse to archive %s", archive_id)
|
|
logger.info("[TIMELAPSE] Successfully attached timelapse to archive %s", archive_id)
|
|
|
- await ws_manager.send_archive_updated({"id": archive_id, "timelapse_attached": True})
|
|
|
|
|
|
|
|
|
|
await delete_archived_timelapse(
|
|
await delete_archived_timelapse(
|
|
|
printer.ip_address,
|
|
printer.ip_address,
|
|
@@ -7800,6 +7893,10 @@ async def _recover_fallback_from_cache_before_eviction(printer_id: int, data: di
|
|
|
logger.debug("[RECOVER] Pre-eviction recovery for %s failed: %s", name, e)
|
|
logger.debug("[RECOVER] Pre-eviction recovery for %s failed: %s", name, e)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
+# Longest the print's end waits for its file to be deleted from the SD card
|
|
|
|
|
+_SD_CLEANUP_BUDGET_SECONDS = 45.0
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
async def _cleanup_sd_card_after_print(
|
|
async def _cleanup_sd_card_after_print(
|
|
|
printer_id: int, subtask_name: str | None, archive_id: int | None, logger
|
|
printer_id: int, subtask_name: str | None, archive_id: int | None, logger
|
|
|
) -> None:
|
|
) -> None:
|
|
@@ -7916,7 +8013,22 @@ async def _cleanup_sd_card_after_print(
|
|
|
|
|
|
|
|
|
|
|
|
|
async def on_print_complete(printer_id: int, data: dict):
|
|
async def on_print_complete(printer_id: int, data: dict):
|
|
|
- """Handle print completion - update the archive status."""
|
|
|
|
|
|
|
+ """Handle print completion - update the archive status.
|
|
|
|
|
+
|
|
|
|
|
+ The printer counts as printing until the print's usage is booked (or this
|
|
|
|
|
+ ends), so the AMS remain% weight sync can't charge it a second time while
|
|
|
|
|
+ the usage tracker works (WU-3). Marked before the first await.
|
|
|
|
|
+ """
|
|
|
|
|
+ from backend.app.services.usage_tracker import done_completing, mark_completing
|
|
|
|
|
+
|
|
|
|
|
+ mark_completing(printer_id)
|
|
|
|
|
+ try:
|
|
|
|
|
+ await _close_out_print(printer_id, data)
|
|
|
|
|
+ finally:
|
|
|
|
|
+ done_completing(printer_id)
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+async def _close_out_print(printer_id: int, data: dict):
|
|
|
import time
|
|
import time
|
|
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
logger = logging.getLogger(__name__)
|
|
@@ -8025,7 +8137,19 @@ async def on_print_complete(printer_id: int, data: dict):
|
|
|
subtask_name = data.get("subtask_name", "")
|
|
subtask_name = data.get("subtask_name", "")
|
|
|
|
|
|
|
|
if not filename and not subtask_name:
|
|
if not filename and not subtask_name:
|
|
|
|
|
+ # Which print ended can't be told: the printer reported no name since
|
|
|
|
|
+ # Bambuddy started (a print that failed while preparing, a calibration).
|
|
|
|
|
+ # Charge nothing, but close the print session, which otherwise kept the
|
|
|
|
|
+ # AMS weight sync off until the next print (AS-5). A print archived as
|
|
|
|
|
+ # running is closed and charged by the reconnect reconcile.
|
|
|
logger.warning("Print complete without filename or subtask_name")
|
|
logger.warning("Print complete without filename or subtask_name")
|
|
|
|
|
+ try:
|
|
|
|
|
+ from backend.app.services.usage_tracker import discard_session
|
|
|
|
|
+
|
|
|
|
|
+ async with async_session() as db:
|
|
|
|
|
+ await discard_session(db, printer_id)
|
|
|
|
|
+ except Exception as e:
|
|
|
|
|
+ logger.warning("Failed to clear print session for printer %s: %s", printer_id, e)
|
|
|
return
|
|
return
|
|
|
|
|
|
|
|
logger.info("Print complete - filename: %s, subtask: %s, status: %s", filename, subtask_name, data.get("status"))
|
|
logger.info("Print complete - filename: %s, subtask: %s, status: %s", filename, subtask_name, data.get("status"))
|
|
@@ -8104,7 +8228,25 @@ async def on_print_complete(printer_id: int, data: dict):
|
|
|
if archive:
|
|
if archive:
|
|
|
archive_id = archive.id
|
|
archive_id = archive.id
|
|
|
|
|
|
|
|
- await _cleanup_sd_card_after_print(printer_id, subtask_name, archive_id, logger)
|
|
|
|
|
|
|
+ # Bounded: a printer whose FTP hangs on DELE took up to 3 candidates x 3
|
|
|
|
|
+ # attempts x a 60 s timeout, about nine minutes, before the queue item was
|
|
|
|
|
+ # closed below. The stranded-job closer gives up after five and closed it
|
|
|
|
|
+ # first, so the job's auto-off and its billing never ran. Inline rather
|
|
|
|
|
+ # than in the background, so the next job of the same file isn't released
|
|
|
|
|
+ # (and uploaded under the name being deleted) while the cleanup still
|
|
|
|
|
+ # works on it. A delete already sent may still finish in its FTP thread
|
|
|
|
|
+ # after the cutoff, as one past delete_file_async's own timeout always did.
|
|
|
|
|
+ try:
|
|
|
|
|
+ await asyncio.wait_for(
|
|
|
|
|
+ _cleanup_sd_card_after_print(printer_id, subtask_name, archive_id, logger),
|
|
|
|
|
+ timeout=_SD_CLEANUP_BUDGET_SECONDS,
|
|
|
|
|
+ )
|
|
|
|
|
+ except TimeoutError:
|
|
|
|
|
+ logger.warning(
|
|
|
|
|
+ "SD card cleanup on printer %s took over %ss; leaving the file there and closing the print",
|
|
|
|
|
+ printer_id,
|
|
|
|
|
+ _SD_CLEANUP_BUDGET_SECONDS,
|
|
|
|
|
+ )
|
|
|
|
|
|
|
|
log_timing("SD card cleanup")
|
|
log_timing("SD card cleanup")
|
|
|
|
|
|
|
@@ -8298,9 +8440,12 @@ async def on_print_complete(printer_id: int, data: dict):
|
|
|
usage_results: list[dict] = []
|
|
usage_results: list[dict] = []
|
|
|
# Prefer ams_mapping captured from MQTT request topic (works for all print sources)
|
|
# Prefer ams_mapping captured from MQTT request topic (works for all print sources)
|
|
|
stored_ams_mapping = data.get("ams_mapping")
|
|
stored_ams_mapping = data.get("ams_mapping")
|
|
|
- # Fallback to _print_ams_mappings for queue/reprint (set before print starts)
|
|
|
|
|
- if not stored_ams_mapping and archive_id:
|
|
|
|
|
- stored_ams_mapping = _print_ams_mappings.pop(archive_id, None)
|
|
|
|
|
|
|
+ # Fallback to _print_ams_mappings for queue/reprint (set before print starts).
|
|
|
|
|
+ # Drained either way, like the plate below: left over, it would be read by
|
|
|
|
|
+ # the archive's next run, dispatched without a mapping.
|
|
|
|
|
+ registered_ams_mapping = _print_ams_mappings.pop(archive_id, None) if archive_id else None
|
|
|
|
|
+ if not stored_ams_mapping:
|
|
|
|
|
+ stored_ams_mapping = registered_ams_mapping
|
|
|
|
|
|
|
|
# Always drain the plate_id register on completion — the session already
|
|
# Always drain the plate_id register on completion — the session already
|
|
|
# consumed it at print-start injection; leaving it would leak into the next
|
|
# consumed it at print-start injection; leaving it would leak into the next
|
|
@@ -9353,18 +9498,16 @@ async def on_print_complete(printer_id: int, data: dict):
|
|
|
try:
|
|
try:
|
|
|
if print_status == "completed":
|
|
if print_status == "completed":
|
|
|
logger.info("[LAYER-TL] Stitching layer timelapse for printer %s", printer_id)
|
|
logger.info("[LAYER-TL] Stitching layer timelapse for printer %s", printer_id)
|
|
|
|
|
+ # The run the frames belong to, read before the stitch: it can
|
|
|
|
|
+ # take long enough for the archive to start printing again.
|
|
|
|
|
+ run_started_at = None
|
|
|
|
|
+ if archive_id:
|
|
|
|
|
+ async with async_session() as db:
|
|
|
|
|
+ run_archive = await ArchiveService(db).get_archive(archive_id)
|
|
|
|
|
+ run_started_at = run_archive.started_at if run_archive else None
|
|
|
timelapse_path = await tl_complete(printer_id)
|
|
timelapse_path = await tl_complete(printer_id)
|
|
|
if timelapse_path and archive_id:
|
|
if timelapse_path and archive_id:
|
|
|
- logger.info("[LAYER-TL] Attaching timelapse %s to archive %s", timelapse_path, archive_id)
|
|
|
|
|
- async with async_session() as db:
|
|
|
|
|
- service = ArchiveService(db)
|
|
|
|
|
- timelapse_data = await asyncio.to_thread(timelapse_path.read_bytes)
|
|
|
|
|
- await service.attach_timelapse(
|
|
|
|
|
- archive_id, timelapse_data, "layer_timelapse.mp4", plate_id=notify_plate_id
|
|
|
|
|
- )
|
|
|
|
|
- # Clean up the temp file
|
|
|
|
|
- await asyncio.to_thread(timelapse_path.unlink, missing_ok=True)
|
|
|
|
|
- logger.info("[LAYER-TL] Layer timelapse attached successfully")
|
|
|
|
|
|
|
+ await _attach_layer_timelapse(archive_id, timelapse_path, notify_plate_id, run_started_at)
|
|
|
elif timelapse_path:
|
|
elif timelapse_path:
|
|
|
# Timelapse created but no archive - just clean up
|
|
# Timelapse created but no archive - just clean up
|
|
|
await asyncio.to_thread(timelapse_path.unlink, missing_ok=True)
|
|
await asyncio.to_thread(timelapse_path.unlink, missing_ok=True)
|