print_queue.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314
  1. """API routes for print queue management."""
  2. import json
  3. import logging
  4. from datetime import datetime
  5. from fastapi import APIRouter, Depends, HTTPException, Query
  6. from sqlalchemy import func, select
  7. from sqlalchemy.ext.asyncio import AsyncSession
  8. from sqlalchemy.orm import selectinload
  9. from backend.app.core.database import get_db
  10. from backend.app.models.archive import PrintArchive
  11. from backend.app.models.print_queue import PrintQueueItem
  12. from backend.app.models.printer import Printer
  13. from backend.app.schemas.print_queue import (
  14. PrintQueueItemCreate,
  15. PrintQueueItemResponse,
  16. PrintQueueItemUpdate,
  17. PrintQueueReorder,
  18. )
  19. logger = logging.getLogger(__name__)
  20. router = APIRouter(prefix="/queue", tags=["queue"])
  21. def _enrich_response(item: PrintQueueItem) -> PrintQueueItemResponse:
  22. """Add nested archive/printer info to response."""
  23. response = PrintQueueItemResponse.model_validate(item)
  24. # Parse ams_mapping from JSON string
  25. if item.ams_mapping:
  26. try:
  27. response.ams_mapping = json.loads(item.ams_mapping)
  28. except json.JSONDecodeError:
  29. response.ams_mapping = None
  30. if item.archive:
  31. response.archive_name = item.archive.print_name or item.archive.filename
  32. response.archive_thumbnail = item.archive.thumbnail_path
  33. response.print_time_seconds = item.archive.print_time_seconds
  34. if item.printer:
  35. response.printer_name = item.printer.name
  36. return response
  37. @router.get("/", response_model=list[PrintQueueItemResponse])
  38. async def list_queue(
  39. printer_id: int | None = Query(None, description="Filter by printer"),
  40. status: str | None = Query(None, description="Filter by status"),
  41. db: AsyncSession = Depends(get_db),
  42. ):
  43. """List all queue items, optionally filtered by printer or status."""
  44. query = (
  45. select(PrintQueueItem)
  46. .options(selectinload(PrintQueueItem.archive), selectinload(PrintQueueItem.printer))
  47. .order_by(PrintQueueItem.printer_id, PrintQueueItem.position)
  48. )
  49. if printer_id is not None:
  50. query = query.where(PrintQueueItem.printer_id == printer_id)
  51. if status:
  52. query = query.where(PrintQueueItem.status == status)
  53. result = await db.execute(query)
  54. items = result.scalars().all()
  55. return [_enrich_response(item) for item in items]
  56. @router.post("/", response_model=PrintQueueItemResponse)
  57. async def add_to_queue(
  58. data: PrintQueueItemCreate,
  59. db: AsyncSession = Depends(get_db),
  60. ):
  61. """Add an item to the print queue."""
  62. # Validate printer exists
  63. result = await db.execute(select(Printer).where(Printer.id == data.printer_id))
  64. if not result.scalar_one_or_none():
  65. raise HTTPException(400, "Printer not found")
  66. # Validate archive exists
  67. result = await db.execute(select(PrintArchive).where(PrintArchive.id == data.archive_id))
  68. if not result.scalar_one_or_none():
  69. raise HTTPException(400, "Archive not found")
  70. # Get next position for this printer
  71. result = await db.execute(
  72. select(func.max(PrintQueueItem.position))
  73. .where(PrintQueueItem.printer_id == data.printer_id)
  74. .where(PrintQueueItem.status == "pending")
  75. )
  76. max_pos = result.scalar() or 0
  77. item = PrintQueueItem(
  78. printer_id=data.printer_id,
  79. archive_id=data.archive_id,
  80. scheduled_time=data.scheduled_time,
  81. require_previous_success=data.require_previous_success,
  82. auto_off_after=data.auto_off_after,
  83. manual_start=data.manual_start,
  84. ams_mapping=json.dumps(data.ams_mapping) if data.ams_mapping else None,
  85. position=max_pos + 1,
  86. status="pending",
  87. )
  88. db.add(item)
  89. await db.commit()
  90. await db.refresh(item)
  91. # Load relationships for response
  92. await db.refresh(item, ["archive", "printer"])
  93. logger.info(f"Added archive {data.archive_id} to queue for printer {data.printer_id}")
  94. return _enrich_response(item)
  95. @router.get("/{item_id}", response_model=PrintQueueItemResponse)
  96. async def get_queue_item(item_id: int, db: AsyncSession = Depends(get_db)):
  97. """Get a specific queue item."""
  98. result = await db.execute(
  99. select(PrintQueueItem)
  100. .options(selectinload(PrintQueueItem.archive), selectinload(PrintQueueItem.printer))
  101. .where(PrintQueueItem.id == item_id)
  102. )
  103. item = result.scalar_one_or_none()
  104. if not item:
  105. raise HTTPException(404, "Queue item not found")
  106. return _enrich_response(item)
  107. @router.patch("/{item_id}", response_model=PrintQueueItemResponse)
  108. async def update_queue_item(
  109. item_id: int,
  110. data: PrintQueueItemUpdate,
  111. db: AsyncSession = Depends(get_db),
  112. ):
  113. """Update a queue item."""
  114. result = await db.execute(select(PrintQueueItem).where(PrintQueueItem.id == item_id))
  115. item = result.scalar_one_or_none()
  116. if not item:
  117. raise HTTPException(404, "Queue item not found")
  118. if item.status != "pending":
  119. raise HTTPException(400, "Can only update pending items")
  120. update_data = data.model_dump(exclude_unset=True)
  121. # Validate new printer_id if being changed
  122. if "printer_id" in update_data:
  123. result = await db.execute(select(Printer).where(Printer.id == update_data["printer_id"]))
  124. if not result.scalar_one_or_none():
  125. raise HTTPException(400, "Printer not found")
  126. for field, value in update_data.items():
  127. setattr(item, field, value)
  128. await db.commit()
  129. await db.refresh(item, ["archive", "printer"])
  130. logger.info(f"Updated queue item {item_id}")
  131. return _enrich_response(item)
  132. @router.delete("/{item_id}")
  133. async def delete_queue_item(item_id: int, db: AsyncSession = Depends(get_db)):
  134. """Remove an item from the queue."""
  135. result = await db.execute(select(PrintQueueItem).where(PrintQueueItem.id == item_id))
  136. item = result.scalar_one_or_none()
  137. if not item:
  138. raise HTTPException(404, "Queue item not found")
  139. if item.status == "printing":
  140. raise HTTPException(400, "Cannot delete item that is currently printing")
  141. await db.delete(item)
  142. await db.commit()
  143. logger.info(f"Deleted queue item {item_id}")
  144. return {"message": "Queue item deleted"}
  145. @router.post("/reorder")
  146. async def reorder_queue(
  147. data: PrintQueueReorder,
  148. db: AsyncSession = Depends(get_db),
  149. ):
  150. """Bulk update positions for queue items."""
  151. for reorder_item in data.items:
  152. result = await db.execute(select(PrintQueueItem).where(PrintQueueItem.id == reorder_item.id))
  153. item = result.scalar_one_or_none()
  154. if item and item.status == "pending":
  155. item.position = reorder_item.position
  156. await db.commit()
  157. logger.info(f"Reordered {len(data.items)} queue items")
  158. return {"message": f"Reordered {len(data.items)} items"}
  159. @router.post("/{item_id}/cancel")
  160. async def cancel_queue_item(item_id: int, db: AsyncSession = Depends(get_db)):
  161. """Cancel a pending queue item."""
  162. result = await db.execute(select(PrintQueueItem).where(PrintQueueItem.id == item_id))
  163. item = result.scalar_one_or_none()
  164. if not item:
  165. raise HTTPException(404, "Queue item not found")
  166. if item.status not in ("pending",):
  167. raise HTTPException(400, f"Cannot cancel item with status '{item.status}'")
  168. item.status = "cancelled"
  169. item.completed_at = datetime.now()
  170. await db.commit()
  171. logger.info(f"Cancelled queue item {item_id}")
  172. return {"message": "Queue item cancelled"}
  173. @router.post("/{item_id}/stop")
  174. async def stop_queue_item(
  175. item_id: int,
  176. db: AsyncSession = Depends(get_db),
  177. ):
  178. """Stop an actively printing queue item."""
  179. import asyncio
  180. from backend.app.models.smart_plug import SmartPlug
  181. from backend.app.services.printer_manager import printer_manager
  182. from backend.app.services.tasmota import tasmota_service
  183. result = await db.execute(select(PrintQueueItem).where(PrintQueueItem.id == item_id))
  184. item = result.scalar_one_or_none()
  185. if not item:
  186. raise HTTPException(404, "Queue item not found")
  187. if item.status != "printing":
  188. raise HTTPException(400, f"Can only stop items that are printing, current status: '{item.status}'")
  189. # Capture values we need for background task
  190. printer_id = item.printer_id
  191. auto_off_after = item.auto_off_after
  192. # Try to send stop command to printer
  193. stop_sent = False
  194. try:
  195. stop_sent = printer_manager.stop_print(printer_id)
  196. if not stop_sent:
  197. logger.warning(f"stop_print returned False for printer {printer_id} - printer may not be connected")
  198. except Exception as e:
  199. logger.error(f"Error sending stop command for queue item {item_id}: {e}")
  200. # Update queue item status regardless - if printer is off, print is already stopped
  201. item.status = "cancelled"
  202. item.completed_at = datetime.now()
  203. item.error_message = "Stopped by user" if stop_sent else "Stopped by user (printer was offline)"
  204. await db.commit()
  205. # Get smart plug info if auto-off is enabled
  206. plug_ip = None
  207. if auto_off_after:
  208. result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id))
  209. plug = result.scalar_one_or_none()
  210. if plug and plug.enabled:
  211. plug_ip = plug.ip_address
  212. logger.info(f"Stopped printing queue item {item_id} (stop command sent: {stop_sent})")
  213. # Schedule background task for cooldown + power off
  214. if plug_ip:
  215. async def cooldown_and_poweroff():
  216. logger.info(f"Auto-off: Waiting for printer {printer_id} to cool down before power off...")
  217. await printer_manager.wait_for_cooldown(printer_id, target_temp=50.0, timeout=600)
  218. # Re-fetch plug since we're in a new async context
  219. from backend.app.core.database import async_session
  220. async with async_session() as new_db:
  221. result = await new_db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id))
  222. plug = result.scalar_one_or_none()
  223. if plug and plug.enabled:
  224. logger.info(f"Auto-off: Powering off printer {printer_id}")
  225. await tasmota_service.turn_off(plug)
  226. asyncio.create_task(cooldown_and_poweroff())
  227. return {"message": "Print stopped" if stop_sent else "Queue item cancelled (printer was offline)"}
  228. @router.post("/{item_id}/start")
  229. async def start_queue_item(
  230. item_id: int,
  231. db: AsyncSession = Depends(get_db),
  232. ):
  233. """Manually start a staged (manual_start) queue item.
  234. This clears the manual_start flag so the scheduler will pick it up,
  235. or starts immediately if the printer is ready.
  236. """
  237. result = await db.execute(
  238. select(PrintQueueItem)
  239. .options(selectinload(PrintQueueItem.archive), selectinload(PrintQueueItem.printer))
  240. .where(PrintQueueItem.id == item_id)
  241. )
  242. item = result.scalar_one_or_none()
  243. if not item:
  244. raise HTTPException(404, "Queue item not found")
  245. if item.status != "pending":
  246. raise HTTPException(400, f"Can only start pending items, current status: '{item.status}'")
  247. # Clear manual_start flag so scheduler picks it up
  248. item.manual_start = False
  249. await db.commit()
  250. await db.refresh(item, ["archive", "printer"])
  251. logger.info(f"Manually started queue item {item_id} (cleared manual_start flag)")
  252. return _enrich_response(item)