321 lines
11 KiB
Python
321 lines
11 KiB
Python
"""Capture one image per detected Z increase and render completed prints."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import http.server
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from urllib.parse import unquote, urlparse
|
|
|
|
|
|
logging.basicConfig(
|
|
level=os.getenv("LOG_LEVEL", "INFO").upper(),
|
|
format="%(asctime)s %(levelname)s %(message)s",
|
|
)
|
|
LOGGER = logging.getLogger("timelapse")
|
|
|
|
|
|
def env_float(name: str, default: float) -> float:
|
|
return float(os.getenv(name, str(default)))
|
|
|
|
|
|
MOONRAKER_URL = os.getenv("MOONRAKER_URL", "http://host.docker.internal:7125").rstrip("/")
|
|
CAMERA_SNAPSHOT_URL = os.getenv(
|
|
"CAMERA_SNAPSHOT_URL",
|
|
"http://host.docker.internal/webcam/?action=snapshot",
|
|
)
|
|
CAMERA_STREAM_URL = os.getenv(
|
|
"CAMERA_STREAM_URL",
|
|
"http://host.docker.internal/webcam/?action=stream",
|
|
)
|
|
POLL_INTERVAL = env_float("POLL_INTERVAL_SECONDS", 0.5)
|
|
MIN_LAYER_HEIGHT = env_float("MIN_LAYER_HEIGHT_MM", 0.05)
|
|
HTTP_TIMEOUT = env_float("HTTP_TIMEOUT_SECONDS", 10)
|
|
OUTPUT_DIR = Path(os.getenv("OUTPUT_DIR", "/timelapses"))
|
|
VIDEO_FPS = os.getenv("VIDEO_FPS", "12")
|
|
VIDEO_CODEC = os.getenv("VIDEO_CODEC", "libx264")
|
|
VIDEO_CRF = os.getenv("VIDEO_CRF", "18")
|
|
VIDEO_PRESET = os.getenv("VIDEO_PRESET", "medium")
|
|
VIDEO_PIXEL_FORMAT = os.getenv("VIDEO_PIXEL_FORMAT", "yuv420p")
|
|
VIDEO_DUPLICATE_END_FRAMES = int(os.getenv("VIDEO_DUPLICATE_END_FRAMES", "0"))
|
|
WEB_HOST = os.getenv("WEB_HOST", "0.0.0.0")
|
|
WEB_PORT = int(os.getenv("WEB_PORT", "8080"))
|
|
|
|
STATUS_LOCK = threading.Lock()
|
|
STATUS = {
|
|
"connected": False,
|
|
"state": "unknown",
|
|
"z_height": None,
|
|
"layer": None,
|
|
"print_name": None,
|
|
"frames": 0,
|
|
"session": None,
|
|
"last_error": None,
|
|
"updated_at": None,
|
|
}
|
|
LATEST_SNAPSHOT: Path | None = None
|
|
|
|
|
|
def update_status(**changes: object) -> None:
|
|
with STATUS_LOCK:
|
|
STATUS.update(changes)
|
|
STATUS["updated_at"] = dt.datetime.now(dt.timezone.utc).isoformat()
|
|
|
|
|
|
def status_snapshot() -> dict:
|
|
with STATUS_LOCK:
|
|
return dict(STATUS)
|
|
|
|
|
|
class PanelHandler(http.server.BaseHTTPRequestHandler):
|
|
def do_GET(self) -> None: # noqa: N802
|
|
request_path = urlparse(self.path).path
|
|
if request_path == "/":
|
|
self.send_file(Path(__file__).with_name("panel.html"), "text/html; charset=utf-8")
|
|
elif request_path == "/api/status":
|
|
self.send_json(status_snapshot())
|
|
elif request_path == "/api/videos":
|
|
self.send_json(list_videos())
|
|
elif request_path == "/api/snapshot":
|
|
self.send_snapshot()
|
|
elif request_path == "/api/camera":
|
|
self.send_camera_stream()
|
|
elif request_path.startswith("/media/"):
|
|
self.send_media(unquote(request_path.removeprefix("/media/")))
|
|
else:
|
|
self.send_error(404)
|
|
|
|
def send_json(self, payload: object) -> None:
|
|
body = json.dumps(payload).encode("utf-8")
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def send_file(self, path: Path, content_type: str) -> None:
|
|
try:
|
|
body = path.read_bytes()
|
|
except OSError:
|
|
self.send_error(404)
|
|
return
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", content_type)
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def send_media(self, relative_path: str) -> None:
|
|
path = (OUTPUT_DIR / relative_path).resolve()
|
|
try:
|
|
path.relative_to(OUTPUT_DIR.resolve())
|
|
except ValueError:
|
|
self.send_error(403)
|
|
return
|
|
if path.suffix != ".mp4" or not path.is_file():
|
|
self.send_error(404)
|
|
return
|
|
self.send_file(path, "video/mp4")
|
|
|
|
def send_snapshot(self) -> None:
|
|
if LATEST_SNAPSHOT is None or not LATEST_SNAPSHOT.is_file():
|
|
self.send_error(404)
|
|
return
|
|
self.send_file(LATEST_SNAPSHOT, "image/jpeg")
|
|
|
|
def send_camera_stream(self) -> None:
|
|
request = urllib.request.Request(CAMERA_STREAM_URL)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=HTTP_TIMEOUT) as response:
|
|
content_type = response.headers.get("Content-Type", "multipart/x-mixed-replace")
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", content_type)
|
|
self.send_header("Cache-Control", "no-cache")
|
|
self.end_headers()
|
|
while True:
|
|
chunk = response.read(64 * 1024)
|
|
if not chunk:
|
|
break
|
|
self.wfile.write(chunk)
|
|
self.wfile.flush()
|
|
except (OSError, urllib.error.URLError):
|
|
LOGGER.warning("Camera stream unavailable at %s", CAMERA_STREAM_URL)
|
|
|
|
def log_message(self, format: str, *args: object) -> None:
|
|
LOGGER.info("panel: " + format, *args)
|
|
|
|
|
|
def list_videos() -> list[dict[str, object]]:
|
|
videos = []
|
|
for path in sorted(OUTPUT_DIR.glob("*.mp4"), reverse=True):
|
|
stat = path.stat()
|
|
videos.append(
|
|
{
|
|
"name": path.name,
|
|
"url": f"/media/{path.name}",
|
|
"size_bytes": stat.st_size,
|
|
"created_at": dt.datetime.fromtimestamp(stat.st_mtime, dt.timezone.utc).isoformat(),
|
|
}
|
|
)
|
|
return videos
|
|
|
|
|
|
def start_panel() -> None:
|
|
server = http.server.ThreadingHTTPServer((WEB_HOST, WEB_PORT), PanelHandler)
|
|
threading.Thread(target=server.serve_forever, name="web-panel", daemon=True).start()
|
|
LOGGER.info("Web panel listening at http://%s:%s", WEB_HOST, WEB_PORT)
|
|
|
|
|
|
def get_json(path: str) -> dict:
|
|
request = urllib.request.Request(f"{MOONRAKER_URL}{path}")
|
|
with urllib.request.urlopen(request, timeout=HTTP_TIMEOUT) as response:
|
|
return json.loads(response.read())
|
|
|
|
|
|
def moonraker_status() -> tuple[str, float, str | None, int | None]:
|
|
result = get_json("/printer/objects/query?print_stats&gcode_move")
|
|
status = result["result"]["status"]
|
|
state = status["print_stats"]["state"]
|
|
z_height = float(status["gcode_move"]["gcode_position"][2])
|
|
filename = status["print_stats"].get("filename")
|
|
current_layer = status["print_stats"].get("info", {}).get("current_layer")
|
|
if current_layer is None:
|
|
current_layer = status["print_stats"].get("current_layer")
|
|
try:
|
|
current_layer = int(current_layer) if current_layer is not None else None
|
|
except (TypeError, ValueError):
|
|
current_layer = None
|
|
return state, z_height, filename or None, current_layer
|
|
|
|
|
|
def capture_snapshot(path: Path) -> None:
|
|
request = urllib.request.Request(CAMERA_SNAPSHOT_URL)
|
|
with urllib.request.urlopen(request, timeout=HTTP_TIMEOUT) as response:
|
|
path.write_bytes(response.read())
|
|
|
|
|
|
def render(session_dir: Path) -> None:
|
|
frames = session_dir / "frame-%06d.jpg"
|
|
output = OUTPUT_DIR / f"{session_dir.name}.mp4"
|
|
command = [
|
|
"ffmpeg",
|
|
"-hide_banner",
|
|
"-loglevel",
|
|
"error",
|
|
"-y",
|
|
"-framerate",
|
|
VIDEO_FPS,
|
|
"-i",
|
|
str(frames),
|
|
]
|
|
if VIDEO_DUPLICATE_END_FRAMES > 0:
|
|
command.extend(
|
|
[
|
|
"-vf",
|
|
f"tpad=stop_mode=clone:stop_frames={VIDEO_DUPLICATE_END_FRAMES}",
|
|
]
|
|
)
|
|
command.extend(
|
|
[
|
|
"-c:v",
|
|
VIDEO_CODEC,
|
|
"-preset",
|
|
VIDEO_PRESET,
|
|
"-crf",
|
|
VIDEO_CRF,
|
|
"-pix_fmt",
|
|
VIDEO_PIXEL_FORMAT,
|
|
str(output),
|
|
]
|
|
)
|
|
subprocess.run(command, check=True)
|
|
LOGGER.info("Rendered %s", output)
|
|
|
|
|
|
def safe_print_name(filename: str | None) -> str:
|
|
name = Path(filename or "unnamed-print").stem
|
|
name = re.sub(r"[^A-Za-z0-9._-]+", "-", name).strip(".-_")
|
|
return name[:100] or "unnamed-print"
|
|
|
|
|
|
def new_session(filename: str | None) -> tuple[Path, float]:
|
|
timestamp = dt.datetime.now(dt.timezone.utc).strftime("%Y%m%d-%H%M%S")
|
|
session_name = f"{safe_print_name(filename)}_{timestamp}"
|
|
session_dir = OUTPUT_DIR / session_name
|
|
session_dir.mkdir(parents=True, exist_ok=True)
|
|
LOGGER.info("Started print session %s", session_name)
|
|
return session_dir, -1.0
|
|
|
|
|
|
def main() -> None:
|
|
global LATEST_SNAPSHOT
|
|
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
|
|
start_panel()
|
|
session_dir: Path | None = None
|
|
last_state = "unknown"
|
|
last_z = -1.0
|
|
last_layer: int | None = None
|
|
frame_number = 0
|
|
rendered = False
|
|
|
|
while True:
|
|
try:
|
|
state, z_height, print_name, current_layer = moonraker_status()
|
|
update_status(
|
|
connected=True,
|
|
state=state,
|
|
z_height=z_height,
|
|
layer=current_layer,
|
|
print_name=print_name,
|
|
last_error=None,
|
|
)
|
|
|
|
if state == "printing" and last_state != "printing":
|
|
session_dir, last_z = new_session(print_name)
|
|
frame_number = 0
|
|
last_layer = None
|
|
rendered = False
|
|
update_status(session=session_dir.name, frames=0)
|
|
|
|
layer_changed = (
|
|
current_layer is not None
|
|
and (last_layer is None or current_layer > last_layer)
|
|
)
|
|
z_changed = current_layer is None and z_height - last_z >= MIN_LAYER_HEIGHT
|
|
if state == "printing" and session_dir and (layer_changed or z_changed):
|
|
frame_path = session_dir / f"frame-{frame_number:06d}.jpg"
|
|
capture_snapshot(frame_path)
|
|
layer_label = f"layer={current_layer}" if current_layer is not None else f"Z={z_height:.3f}"
|
|
LOGGER.info("Captured %s (%s)", layer_label, frame_path.name)
|
|
frame_number += 1
|
|
last_z = z_height
|
|
last_layer = current_layer
|
|
LATEST_SNAPSHOT = frame_path
|
|
update_status(frames=frame_number)
|
|
|
|
if state == "complete" and last_state == "printing" and session_dir and not rendered:
|
|
render(session_dir)
|
|
rendered = True
|
|
|
|
if state in {"cancelled", "error"}:
|
|
LOGGER.warning("Print ended in state %s; no video rendered", state)
|
|
|
|
last_state = state
|
|
except (OSError, KeyError, TypeError, ValueError, urllib.error.URLError, subprocess.CalledProcessError) as error:
|
|
LOGGER.warning("Poll/capture failed: %s", error)
|
|
update_status(connected=False, last_error=str(error))
|
|
|
|
time.sleep(POLL_INTERVAL)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |