"""Service for writing independent print log entries. Log entries are written to a separate table and never touch archives or queue items. """ import logging import uuid from collections.abc import Set as AbstractSet from datetime import datetime, timezone from types import SimpleNamespace from sqlalchemy import delete, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from backend.app.models.print_log import INCLUDE_OPEN_RUNS, RUN_OPEN, PrintLogEntry from backend.app.utils.local_time import utcnow_naive logger = logging.getLogger(__name__) def wear_cost_for_run(duration_seconds: int | None, wear_cost_per_hour: float | None) -> float | None: """Printer wear for one run: its duration at the printer's hourly rate (#694). None when the printer has no rate or the run has no measured duration (a reconciled run stores 0 because its real end time is unknown). """ if not isinstance(wear_cost_per_hour, int | float) or not isinstance(duration_seconds, int): return None if wear_cost_per_hour <= 0 or duration_seconds <= 0: return None return round(duration_seconds / 3600 * wear_cost_per_hour, 3) async def write_log_entry( db: AsyncSession, *, status: str, archive_id: int | None = None, queue_item_id: int | None = None, print_name: str | None = None, plate_id: int | None = None, printer_name: str | None = None, printer_id: int | None = None, started_at: datetime | None = None, completed_at: datetime | None = None, filament_type: str | None = None, filament_color: str | None = None, filament_used_grams: float | None = None, cost: float | None = None, energy_kwh: float | None = None, energy_cost: float | None = None, wear_cost_per_hour: float | None = None, failure_reason: str | None = None, thumbnail_path: str | None = None, created_by_id: int | None = None, created_by_username: str | None = None, reconciled: bool = False, duration_unmeasured: bool = False, ended_offline: bool = False, replaces_offline_stop: bool = False, ) -> PrintLogEntry: """Write a print log entry. ``reconciled`` marks a synthetic completion written when a stale ``status="printing"`` archive is closed out at reconnect. Its real end time is unknown — the print stopped somewhere during the disconnect and ``completed_at`` is only the reconnect moment — so ``completed_at - started_at`` would bank the entire disconnect gap as print time, adding hundreds of fictitious hours across a farm of stale rows (#2592). For those entries we store an explicit ``0`` ("no measured runtime") rather than a fabricated duration; the stats total trusts a stored 0 instead of recomputing from the stale timestamps. ``duration_unmeasured`` marks a run closed at reconnect, whose times were not measured (#3310): ``completed_at`` is its start plus the slicer's estimate, or the reconnect. The time accuracy leaves it out. ``ended_offline`` marks a run logged by a stop that could not reach the printer. A print's end passes ``replaces_offline_stop``: it may then fill such a run of its printer, the printer's latest, which its archive's start (another plate's since, maybe) and its closed queue item no longer find. """ if replaces_offline_stop: entry = await run_being_ended(db, archive_id, printer_id, started_at) else: entry = await _entry_of_this_run(db, archive_id, printer_id, started_at) filling_open_run = entry is not None and entry.status == RUN_OPEN if entry is not None and entry.started_at is not None: # The run's own start. The archive's is its latest run's: another # plate of it may have started on another printer since. started_at = entry.started_at if reconciled: duration: int | None = 0 elif started_at and completed_at: duration = int((_utc(completed_at) - _utc(started_at)).total_seconds()) else: duration = None fields = { "archive_id": archive_id, "queue_item_id": queue_item_id, "print_name": print_name, "plate_id": plate_id, "printer_name": printer_name, "printer_id": printer_id, "status": status, "started_at": started_at, "completed_at": completed_at, "duration_seconds": duration, "filament_type": filament_type, "filament_color": filament_color, "filament_used_grams": filament_used_grams, "cost": cost, "energy_kwh": energy_kwh, "energy_cost": energy_cost, "wear_cost": wear_cost_for_run(duration, wear_cost_per_hour), "duration_unmeasured": True if duration_unmeasured else None, "failure_reason": failure_reason, "thumbnail_path": thumbnail_path, "created_by_id": created_by_id, "created_by_username": created_by_username, "ended_offline": True if ended_offline else None, } if filling_open_run: # When the run was logged, as a run written at its end always was: # the date ranges of statistics and the print log go by it. fields["created_at"] = utcnow_naive() if entry is None: if archive_id is not None and started_at is not None and status != RUN_OPEN: # Shadow check while the run's start moves over (archive redesign): # every print start that binds an archive opens its run, so a run # ending without one is a start path that does not yet. logger.info("[RUN] Archive %s: no run was opened when the print started; written at its end", archive_id) entry = PrintLogEntry(**fields) db.add(entry) elif entry.status != status: if entry.status == RUN_OPEN and None not in (entry.plate_id, plate_id) and entry.plate_id != plate_id: logger.warning( "[RUN] Archive %s: the run opened for plate %s ends as plate %s", archive_id, entry.plate_id, plate_id ) # The run ended differently from how it was first logged: a print # stopped while its printer was offline that finished after all. The # real outcome replaces the stop rather than counting the run twice. for name, value in fields.items(): if value is None and name in _KEPT_WHEN_NOT_REPORTED: continue setattr(entry, name, value) # The same outcome reported again (a repeated completion after a reconnect) # changes nothing: the repeat no longer has the run's tracked usage, and the # entry may have been edited or given its energy since. await db.flush() if archive_id is not None: await mirror_latest_run(db, archive_id) return entry # What the archive card shows of its prints, its latest run's. _MIRRORED_FROM_RUN = ("status", "started_at", "completed_at", "failure_reason") async def mirror_latest_run(db: AsyncSession, archive_id: int) -> None: """Give the archive its latest run's outcome, by start: status, times, failure reason, plate and verdict. Only once none of its runs is printing: until the print's start, end, reconcile and timelapse look their print up by its run, they still go by the archive's printer, status and start, which a print running elsewhere must not move under them. While one prints, the archive keeps what its prints' starts and ends wrote. An archive without runs keeps its own (uploaded, or printed before runs were logged). Does not commit.""" from backend.app.models.archive import PrintArchive archive = await db.get(PrintArchive, archive_id) if archive is None: return runs = select(PrintLogEntry).where(PrintLogEntry.archive_id == archive_id) still_printing = await db.scalar( runs.with_only_columns(PrintLogEntry.id) .where(PrintLogEntry.status == RUN_OPEN) .limit(1) .execution_options(**{INCLUDE_OPEN_RUNS: True}) ) if still_printing is not None: return latest = await db.scalar( runs.order_by(PrintLogEntry.started_at.desc().nulls_last(), PrintLogEntry.id.desc()).limit(1) ) if latest is None: return for name in _MIRRORED_FROM_RUN: setattr(archive, name, getattr(latest, name)) if latest.plate_id is not None: archive.plate_id = latest.plate_id # Its outcome verdict too (#1898), once the run holds its own: it has one, # or was opened at its start with its opt-in. A run logged before then # may never have been given the archive's, which stays. Runs logged # before they kept the verdict's provenance have only its value: the # archive's own provenance stays with that same verdict. owns_verdict = latest.user_verdict is not None or latest.confirm_requested is not None if owns_verdict and (latest.user_verdict != archive.user_verdict or latest.user_verdict_source is not None): archive.user_verdict = latest.user_verdict archive.user_verdict_source = latest.user_verdict_source archive.user_verdict_at = latest.user_verdict_at async def mirror_archive_cost( db: AsyncSession, archive, previous_cost: float | None, *, skip_runs: AbstractSet[int] = frozenset() ) -> None: """Give the archive's newest run the archive's new cost, if it had the old one. Statistics, project totals and batch sums read a run's cost. A completed run without its own spool price took the archive's (or none), so a new archive cost belongs to it too. A run its spools priced keeps that price, and a failed or cancelled run keeps its own: it printed only part of it. A run in ``skip_runs`` was given its own new cost already. """ entry = await db.scalar( select(PrintLogEntry).where(PrintLogEntry.archive_id == archive.id).order_by(PrintLogEntry.id.desc()).limit(1) ) if entry is None or entry.id in skip_runs: return if entry.status == "completed" and entry.cost in (None, previous_cost): entry.cost = archive.cost async def other_runs_of(db: AsyncSession, archive) -> int: """How many log entries *archive* has from runs other than its current one. The archive card keeps its first run's figures; a run whose own entry is already written (its stop, logged while the printer was offline) is still the first run. """ # Without a start time or printer on the archive, ``!= None`` reads as # ``IS NOT NULL``: every entry is another run's. query = select(func.count(PrintLogEntry.id)).where( PrintLogEntry.archive_id == archive.id, or_( PrintLogEntry.started_at.is_(None), PrintLogEntry.started_at != archive.started_at, PrintLogEntry.printer_id.is_(None), PrintLogEntry.printer_id != archive.printer_id, ), ) return await db.scalar(query) or 0 # Who and what the run was, which a repeated report may no longer know: a # completion that finds the queue item already closed reports none. Its # outcome (status, reason, grams, cost) is replaced as reported. _KEPT_WHEN_NOT_REPORTED = frozenset( { "queue_item_id", "print_name", "plate_id", "printer_name", "thumbnail_path", "created_by_id", "created_by_username", } ) def _utc(moment: datetime) -> datetime: """Naive UTC, as the columns store it; an aware time is converted first.""" return moment.astimezone(timezone.utc).replace(tzinfo=None) if moment.tzinfo else moment async def _entry_of_this_run( db: AsyncSession, archive_id: int | None, printer_id: int | None, started_at: datetime | None, *, exact: bool = False, ) -> PrintLogEntry | None: """The entry already written for this run, if any. A run is its archive, printer and start time together: plates of one archive printing on two printers share the archive row (and can share its start time), and a reprint starts at a new time. The queue item is not part of it: a repeated completion finds the item already closed and reports none. Without an archive or printer a run can't be told apart, so nothing is matched. A run opened at its start is also found as the archive's run open on the printer, unless ``exact``: the archive's start is its latest run's, and another plate of it may have started on another printer since. """ if archive_id is None or printer_id is None: return None if started_at is None: return None if exact else await _open_run_on(db, archive_id, printer_id) query = ( select(PrintLogEntry) .where( PrintLogEntry.archive_id == archive_id, PrintLogEntry.printer_id == printer_id, PrintLogEntry.started_at == started_at, ) # The run opened when the print started is the one this fills in. .execution_options(**{INCLUDE_OPEN_RUNS: True}) ) entry = (await db.scalars(query.order_by(PrintLogEntry.id.desc()).limit(1))).first() # The archive's start is its latest run's. With another plate of it # started on another printer since, this printer's run is found as the # one open there. if entry is not None or exact: return entry return await _open_run_on(db, archive_id, printer_id) async def run_being_ended( db: AsyncSession, archive_id: int | None, printer_id: int | None, started_at: datetime | None ) -> PrintLogEntry | None: """The run a print's end fills in (``write_log_entry`` with ``replaces_offline_stop``): the one written for it, else the one a stop logged while the printer was offline. Billing charges the same run.""" entry = await _entry_of_this_run(db, archive_id, printer_id, started_at) if entry is None and archive_id is not None and printer_id is not None: entry = await _stopped_while_offline(db, archive_id, printer_id) return entry async def _stopped_while_offline(db: AsyncSession, archive_id: int, printer_id: int) -> PrintLogEntry | None: """The run a print's real end replaces after a stop logged while the printer was offline: the printer's latest run, when it is of this archive and was ended that way. A printer prints one thing at a time, so any later run there, of any file, means that print is long over.""" latest = await db.scalar( select(PrintLogEntry) .where(PrintLogEntry.printer_id == printer_id) .order_by(PrintLogEntry.started_at.desc().nulls_last(), PrintLogEntry.id.desc()) .limit(1) .execution_options(**{INCLUDE_OPEN_RUNS: True}) ) if latest is not None and latest.archive_id == archive_id and latest.ended_offline: return latest return None async def _open_run_on(db: AsyncSession, archive_id: int, printer_id: int) -> PrintLogEntry | None: """The run of the archive open on the printer: there is one at most, a printer prints one thing at a time (``open_run``).""" return await db.scalar( select(PrintLogEntry) .where( PrintLogEntry.archive_id == archive_id, PrintLogEntry.printer_id == printer_id, PrintLogEntry.status == RUN_OPEN, ) .order_by(PrintLogEntry.id.desc()) .limit(1) .execution_options(**{INCLUDE_OPEN_RUNS: True}) ) async def open_run_on_printer( db: AsyncSession, printer_id: int, *, subtask_id: str | None = None ) -> PrintLogEntry | None: """The run open on the printer: the print it is running now, which Bambuddy saw start. One at most (``open_run``). With ``subtask_id``, only a run of that job.""" query = select(PrintLogEntry).where(PrintLogEntry.printer_id == printer_id, PrintLogEntry.status == RUN_OPEN) if subtask_id is not None: query = query.where(PrintLogEntry.subtask_id == subtask_id) return await db.scalar( query.order_by(PrintLogEntry.id.desc()).limit(1).execution_options(**{INCLUDE_OPEN_RUNS: True}) ) async def discard_open_runs(db: AsyncSession, printer_id: int) -> None: """Drop the runs left open on the printer by prints that ended without their end reaching a run: cancelled at the next start as stale, or whose archive was deleted. As before runs were opened at the start, such a print leaves no run. Does not commit.""" await db.execute( delete(PrintLogEntry) .where(PrintLogEntry.printer_id == printer_id, PrintLogEntry.status == RUN_OPEN) .execution_options(**{INCLUDE_OPEN_RUNS: True}) ) async def open_run( db: AsyncSession, archive, *, printer_id: int, printer_name: str | None = None, plate_id: int | None = None, subtask_id: str | None = None, ) -> PrintLogEntry | None: """The run of the print that just started on ``archive``, written now with the open status; its end fills it in (``write_log_entry``). Idempotent: a print start seen again (a restart resuming the print, a repeated report) finds the run it opened: by the printer's job (``subtask_id``) when it has one, else keyed as its end will key it, by archive, printer and start time. The queue item running it, if any, gives the dispatch's own billing run and cost centre; a run without one gets its own billing run id, so each print of the archive is charged as its own. Does not commit. """ from backend.app.models.print_queue import PrintQueueItem if archive.started_at is None: return None if subtask_id: # The same job seen again: its run, though another plate of the # archive may have started on another printer since and moved the # archive's start. running = await open_run_on_printer(db, printer_id, subtask_id=subtask_id) if running is not None and running.archive_id == archive.id: return running # Only the run of this very start: one still open on the printer from an # earlier start of the archive is of a print that ended unseen. entry = await _entry_of_this_run(db, archive.id, printer_id, archive.started_at, exact=True) if entry is not None: return entry # A printer prints one thing at a time: a run still open on it is of a # print that ended without its end reaching it. await discard_open_runs(db, printer_id) item = await db.scalar( select(PrintQueueItem) .where( PrintQueueItem.archive_id == archive.id, PrintQueueItem.printer_id == printer_id, PrintQueueItem.status == "printing", ) .order_by(PrintQueueItem.id.desc()) .limit(1) ) entry = PrintLogEntry( status=RUN_OPEN, archive_id=archive.id, queue_item_id=item.id if item is not None else None, print_name=archive.print_name, plate_id=plate_id if plate_id is not None else archive.plate_id, printer_name=printer_name, printer_id=printer_id, started_at=archive.started_at, filament_type=archive.filament_type, filament_color=archive.filament_color, thumbnail_path=archive.thumbnail_path, created_by_id=archive.created_by_id, subtask_id=archive.subtask_id, # Not the archive's: that is the latest dispatch's, or the id an # earlier print of the archive was charged under. billing_run_id=(item.billing_run_id if item is not None else None) or str(uuid.uuid4()), cost_center_id=(item.cost_center_id if item is not None else None) or archive.cost_center_id, confirm_requested=archive.confirm_requested, quantity=archive.quantity, ) db.add(entry) await db.flush() return entry async def mirror_energy_start(db: AsyncSession, archive, *, printer_id: int) -> None: """Copy the archive's energy start reading onto the run it was taken for. Does not commit.""" entry = await _entry_of_this_run(db, archive.id, printer_id, archive.started_at) if entry is None or entry.status != RUN_OPEN: return for name in _ENERGY_START: setattr(entry, name, getattr(archive, name)) _ENERGY_START = ("energy_start_kwh", "energy_start_at", "energy_start_plug_id", "energy_start_price") def _reading_taken_for(archive, entry: PrintLogEntry) -> bool: """Whether the archive's energy start reading was taken for this run: the archive's latest start is this run's (another plate of it started since takes the reading), and it was read at or after it (an earlier print's is older). A reading without its time (taken before #1251) goes by the start alone.""" if archive.energy_start_kwh is None: return False if entry.started_at is None or archive.started_at is None: return False if _utc(archive.started_at) != _utc(entry.started_at): return False return archive.energy_start_at is None or _utc(archive.energy_start_at) >= _utc(entry.started_at) async def run_needs_energy_start(db: AsyncSession, archive, *, printer_id: int) -> bool: """Whether the print running on the archive still lacks its energy start reading, at a resume. Its run's, or the archive's taken for it (copied onto the run then), counts; the archive's of an earlier print does not. Without a run, the archive's, as before. Does not commit.""" entry = await _entry_of_this_run(db, archive.id, printer_id, archive.started_at) if entry is None or entry.status != RUN_OPEN: return archive.energy_start_kwh is None if entry.energy_start_kwh is not None: return False if _reading_taken_for(archive, entry): for name in _ENERGY_START: setattr(entry, name, getattr(archive, name)) return False return True def energy_start_of(entry: PrintLogEntry | None, archive) -> SimpleNamespace | None: """The energy start reading of the run that ended: its own. Without one, the archive's only when taken for this run (a run written only at its end, or none written): another plate of the archive may have started on another printer since and taken it.""" source = None if entry is not None and entry.energy_start_kwh is not None: source = entry elif entry is None or _reading_taken_for(archive, entry): source = archive if source is None or source.energy_start_kwh is None: return None return SimpleNamespace( kwh=source.energy_start_kwh, at=source.energy_start_at, plug_id=source.energy_start_plug_id, price=source.energy_start_price, ) async def record_archive_wear(db: AsyncSession, archive, entry: PrintLogEntry) -> None: """Copy a run's wear cost onto its archive when it is the archive's first run. Same rule as energy (#1378): the archive shows its first print, and a reprint's wear stays on its own log entry (#694). ``entry`` must already be flushed, so it counts as one of the archive's runs. """ if entry.wear_cost is None: return runs = await db.scalar(select(func.count(PrintLogEntry.id)).where(PrintLogEntry.archive_id == archive.id)) if (runs or 0) <= 1: archive.wear_cost = entry.wear_cost