initial POC
This commit is contained in:
Dark98
@@ -0,0 +1,321 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user