362 lines
13 KiB
Python
362 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""CRT photobooth.
|
|
|
|
States: IDLE (background + text) -> PREVIEW (live feed, "get ready") ->
|
|
CAPTURING (countdowns, flashes, 3 photos per shoot). Every 4th shoot the
|
|
12 photos are composed onto a contact sheet and printed. Five minutes of
|
|
inactivity brings up the fish tank screensaver.
|
|
|
|
While running, the booth re-execs itself under systemd-inhibit so the
|
|
machine will not sleep or suspend (and, when the lock is available, ignores
|
|
the laptop lid) - it can run headless with the lid closed.
|
|
|
|
Run with --windowed for testing without taking over the display.
|
|
Esc or Q quits.
|
|
"""
|
|
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import tomllib
|
|
from pathlib import Path
|
|
|
|
# On a console-only machine (no X or Wayland session), tell SDL to draw
|
|
# straight to the screen via KMS/DRM, so no desktop environment is needed.
|
|
if not os.environ.get("DISPLAY") and not os.environ.get("WAYLAND_DISPLAY"):
|
|
os.environ.setdefault("SDL_VIDEODRIVER", "kmsdrm")
|
|
|
|
import pygame
|
|
|
|
import camera as cam
|
|
import contact_sheet
|
|
import falling_photos
|
|
import screensaver
|
|
|
|
ROOT = Path(__file__).resolve().parent
|
|
|
|
IDLE, PREVIEW, COUNTDOWN, FLASH, SCREENSAVER = range(5)
|
|
|
|
KEEP_AWAKE_ENV = "PHOTOBOOTH_INHIBITED"
|
|
KEEP_AWAKE_WHY = "Photobooth is running (keep awake, ignore lid)"
|
|
|
|
|
|
def _probe_inhibit_locks():
|
|
"""Largest lock set systemd-inhibit can hold on this machine.
|
|
|
|
Returns a --what value like "sleep:idle:handle-lid-switch", or None when
|
|
systemd-inhibit is missing or no lock can be taken. handle-lid-switch
|
|
allows only one block holder, so it is skipped when a desktop
|
|
environment already holds it. Each probe runs `true` under the lock, so
|
|
locks are taken and released instantly.
|
|
"""
|
|
inhibit = shutil.which("systemd-inhibit")
|
|
if not inhibit:
|
|
return None
|
|
for what in ("sleep:idle:handle-lid-switch", "sleep:idle", "sleep"):
|
|
probe = subprocess.run([inhibit, "--mode=block", f"--what={what}",
|
|
"--", "true"], capture_output=True)
|
|
if probe.returncode == 0:
|
|
return what
|
|
return None
|
|
|
|
|
|
def reexec_keep_awake():
|
|
"""Re-exec this process under systemd-inhibit (once) so the machine
|
|
stays awake - and ignores the lid when that lock is free - while the
|
|
booth runs. Locks are released when the program exits. Best-effort:
|
|
hosts without systemd-inhibit simply run without locks."""
|
|
if os.environ.get(KEEP_AWAKE_ENV):
|
|
return
|
|
what = _probe_inhibit_locks()
|
|
if not what:
|
|
return
|
|
os.environ[KEEP_AWAKE_ENV] = "1"
|
|
argv = [shutil.which("systemd-inhibit"), "--mode=block",
|
|
f"--what={what}", f"--why={KEEP_AWAKE_WHY}", "--",
|
|
sys.executable, str(ROOT / "photobooth.py"), *sys.argv[1:]]
|
|
# flush: execvp replaces the process without draining stdout's buffer.
|
|
print(f"Keep-awake: re-running under systemd-inhibit ({what})", flush=True)
|
|
try:
|
|
os.execvp(argv[0], argv)
|
|
except OSError as e:
|
|
print(f"Keep-awake: re-exec failed ({e}); running without locks")
|
|
|
|
|
|
def load_config():
|
|
with open(ROOT / "config.toml", "rb") as f:
|
|
return tomllib.load(f)
|
|
|
|
|
|
def load_fish_images(fish_dir):
|
|
images = []
|
|
for path in sorted(fish_dir.iterdir()):
|
|
if path.suffix.lower() in (".png", ".jpg", ".jpeg", ".gif", ".bmp"):
|
|
images.append(pygame.image.load(path).convert_alpha())
|
|
return images
|
|
|
|
|
|
class Booth:
|
|
def __init__(self, windowed=False):
|
|
self.cfg = load_config()
|
|
disp = self.cfg["display"]
|
|
self.size = (disp["width"], disp["height"])
|
|
|
|
pygame.init()
|
|
flags = 0 if windowed else pygame.FULLSCREEN
|
|
self.screen = pygame.display.set_mode(self.size, flags)
|
|
pygame.display.set_caption("Photobooth")
|
|
pygame.mouse.set_visible(False)
|
|
self.clock = pygame.time.Clock()
|
|
|
|
h = self.size[1]
|
|
self.font_big = pygame.font.Font(None, h // 3) # countdown digits
|
|
self.font_title = pygame.font.Font(None, h // 6)
|
|
self.font_med = pygame.font.Font(None, h // 12)
|
|
self.font_small = pygame.font.Font(None, h // 18)
|
|
|
|
bg_path = ROOT / "assets" / "background.png"
|
|
self.background = pygame.transform.smoothscale(
|
|
pygame.image.load(bg_path).convert(), self.size)
|
|
|
|
self.photos_dir = ROOT / "photos"
|
|
self.sheets_dir = ROOT / "contact_sheets"
|
|
self.photos_dir.mkdir(exist_ok=True)
|
|
self.sheets_dir.mkdir(exist_ok=True)
|
|
|
|
c = self.cfg["camera"]
|
|
self.camera = cam.Camera(c["device"], c["capture_width"], c["capture_height"])
|
|
self.mirror = c.get("mirror_preview", True)
|
|
|
|
self.fish_images = load_fish_images(ROOT / "fish")
|
|
|
|
self.falling = falling_photos.FallingPhotos(self.size, self.photos_dir)
|
|
|
|
booth = self.cfg["booth"]
|
|
self.photos_per_shoot = booth["photos_per_shoot"]
|
|
self.shoots_per_print = booth["shoots_per_print"]
|
|
|
|
self.state = IDLE
|
|
self.last_activity = time.monotonic()
|
|
self.tank = None
|
|
self.notice = None # (message, expiry_time)
|
|
self.photos_in_shoot = 0
|
|
self.countdown_end = 0.0
|
|
self.flash_end = 0.0
|
|
|
|
# Resume mid-cycle after a crash: existing photos count toward the
|
|
# current print cycle.
|
|
self.shoots_done = len(self.saved_photos()) // self.photos_per_shoot
|
|
|
|
# ---------- helpers ----------
|
|
|
|
def saved_photos(self):
|
|
return sorted(self.photos_dir.glob("*.jpg"))
|
|
|
|
def set_notice(self, message, seconds=6):
|
|
self.notice = (message, time.monotonic() + seconds)
|
|
|
|
def draw_text(self, text, font, center, color=(255, 255, 255)):
|
|
shadow = font.render(text, True, (0, 0, 0))
|
|
label = font.render(text, True, color)
|
|
rect = label.get_rect(center=center)
|
|
offset = max(2, font.get_height() // 20)
|
|
self.screen.blit(shadow, rect.move(offset, offset))
|
|
self.screen.blit(label, rect)
|
|
|
|
def draw_camera_feed(self):
|
|
frame = self.camera.get_frame()
|
|
if frame is None:
|
|
self.screen.fill((20, 20, 20))
|
|
self.draw_text("Warming up camera...", self.font_med,
|
|
(self.size[0] // 2, self.size[1] // 2))
|
|
else:
|
|
self.screen.blit(cam.frame_to_surface(frame, self.size, self.mirror), (0, 0))
|
|
if self.camera.looks_black():
|
|
self.draw_text("Camera problem: unplug it and plug it back in!",
|
|
self.font_small,
|
|
(self.size[0] // 2, self.size[1] // 8),
|
|
(255, 120, 120))
|
|
|
|
# ---------- states ----------
|
|
|
|
def press(self):
|
|
"""Handle a button (mouse) press in the current state."""
|
|
self.last_activity = time.monotonic()
|
|
if self.state == SCREENSAVER:
|
|
self.state = IDLE
|
|
self.tank = None
|
|
elif self.state == IDLE:
|
|
self.state = PREVIEW
|
|
elif self.state == PREVIEW:
|
|
self.photos_in_shoot = 0
|
|
self.start_countdown(self.cfg["booth"]["first_countdown_seconds"])
|
|
# COUNTDOWN / FLASH ignore presses
|
|
|
|
def start_countdown(self, seconds):
|
|
self.state = COUNTDOWN
|
|
self.countdown_end = time.monotonic() + seconds
|
|
|
|
def take_photo(self):
|
|
stamp = time.strftime("%Y%m%d_%H%M%S")
|
|
path = self.photos_dir / (
|
|
f"shoot{self.shoots_done + 1}_photo{self.photos_in_shoot + 1}_{stamp}.jpg")
|
|
if self.camera.save_photo(str(path)):
|
|
self.photos_in_shoot += 1
|
|
else:
|
|
self.set_notice("Camera error - photo not saved!")
|
|
self.state = FLASH
|
|
self.flash_end = time.monotonic() + self.cfg["booth"]["flash_duration_ms"] / 1000
|
|
|
|
def finish_shoot(self):
|
|
self.shoots_done += 1
|
|
if self.shoots_done >= self.shoots_per_print:
|
|
self.print_cycle()
|
|
self.state = IDLE
|
|
self.last_activity = time.monotonic()
|
|
|
|
def print_cycle(self):
|
|
# Blocking is fine here: composing + queueing takes a second or two,
|
|
# and the "Printing..." screen explains the pause.
|
|
self.screen.fill((0, 0, 40))
|
|
self.draw_text("Printing...", self.font_title,
|
|
(self.size[0] // 2, self.size[1] // 2))
|
|
pygame.display.flip()
|
|
|
|
photos = self.saved_photos()
|
|
stamp = time.strftime("%Y%m%d_%H%M%S")
|
|
sheet_path = self.sheets_dir / f"sheet_{stamp}.jpg"
|
|
cs = self.cfg.get("contact_sheet", {})
|
|
contact_sheet.build_sheet(
|
|
photos, sheet_path,
|
|
footer_cfg=self.cfg.get("footer"),
|
|
photos_per_shoot=self.photos_per_shoot,
|
|
shoots_per_print=self.shoots_per_print,
|
|
margin=cs.get("margin_px", 120),
|
|
photo_gutter=cs.get("photo_gutter_px", 30),
|
|
strip_gutter=cs.get("strip_gutter_px", 30),
|
|
auto_orient=cs.get("auto_orient", True),
|
|
)
|
|
|
|
printing = self.cfg["printing"]
|
|
if printing["enabled"]:
|
|
ok, message = contact_sheet.print_sheet(sheet_path, printing["printer"])
|
|
else:
|
|
ok, message = True, "printing disabled in config"
|
|
|
|
if ok:
|
|
for photo in photos:
|
|
photo.unlink()
|
|
self.falling.reload()
|
|
if printing["delete_contact_sheets"]:
|
|
sheet_path.unlink()
|
|
sheet_path.with_suffix(".pdf").unlink(missing_ok=True)
|
|
self.shoots_done = 0
|
|
self.set_notice("Sent to printer!")
|
|
else:
|
|
# Keep everything so nothing is lost; the sheet can be printed
|
|
# by hand and the counter stays full until it succeeds.
|
|
self.shoots_done = 0
|
|
self.set_notice(f"Print failed: {message}", seconds=15)
|
|
print(f"PRINT FAILED ({message}). Sheet saved at {sheet_path}",
|
|
file=sys.stderr)
|
|
|
|
# ---------- drawing ----------
|
|
|
|
def draw_idle(self):
|
|
self.screen.blit(self.background, (0, 0))
|
|
self.falling.draw(self.screen)
|
|
text = self.cfg["text"]
|
|
w, h = self.size
|
|
self.draw_text(text["idle_title"], self.font_title, (w // 2, h // 4))
|
|
self.draw_text(text["idle_subtitle"], self.font_med, (w // 2, h // 2))
|
|
left = self.shoots_per_print - self.shoots_done
|
|
self.draw_text(text["shoots_left_format"].format(n=left),
|
|
self.font_med, (w // 2, h * 3 // 4), (255, 230, 120))
|
|
if self.notice:
|
|
message, expiry = self.notice
|
|
if time.monotonic() > expiry:
|
|
self.notice = None
|
|
else:
|
|
self.draw_text(message, self.font_small, (w // 2, h * 7 // 8),
|
|
(255, 150, 150))
|
|
|
|
def draw_preview(self):
|
|
self.draw_camera_feed()
|
|
self.draw_text(self.cfg["text"]["get_ready"], self.font_med,
|
|
(self.size[0] // 2, self.size[1] * 7 // 8))
|
|
|
|
def draw_countdown(self):
|
|
self.draw_camera_feed()
|
|
remaining = self.countdown_end - time.monotonic()
|
|
if remaining <= 0:
|
|
self.take_photo()
|
|
return
|
|
self.draw_text(str(int(remaining) + 1), self.font_big,
|
|
(self.size[0] // 2, self.size[1] // 2))
|
|
|
|
def draw_flash(self):
|
|
self.screen.fill((255, 255, 255))
|
|
if time.monotonic() >= self.flash_end:
|
|
if self.photos_in_shoot >= self.photos_per_shoot:
|
|
self.finish_shoot()
|
|
else:
|
|
self.start_countdown(self.cfg["booth"]["next_countdown_seconds"])
|
|
|
|
# ---------- main loop ----------
|
|
|
|
def run(self):
|
|
timeout = self.cfg["screensaver"]["timeout_seconds"]
|
|
while True:
|
|
dt = self.clock.tick(30) / 1000
|
|
for event in pygame.event.get():
|
|
if event.type == pygame.QUIT:
|
|
return
|
|
if event.type == pygame.KEYDOWN:
|
|
if event.key in (pygame.K_ESCAPE, pygame.K_q):
|
|
return
|
|
self.last_activity = time.monotonic()
|
|
if event.type == pygame.MOUSEBUTTONDOWN:
|
|
self.press()
|
|
|
|
if (self.state in (IDLE, PREVIEW)
|
|
and time.monotonic() - self.last_activity > timeout):
|
|
self.state = SCREENSAVER
|
|
self.tank = screensaver.FishTank(
|
|
self.size, self.fish_images,
|
|
self.cfg["screensaver"]["num_fish"])
|
|
|
|
if self.state == IDLE:
|
|
self.falling.update(dt)
|
|
self.draw_idle()
|
|
elif self.state == PREVIEW:
|
|
self.draw_preview()
|
|
elif self.state == COUNTDOWN:
|
|
self.draw_countdown()
|
|
elif self.state == FLASH:
|
|
self.draw_flash()
|
|
elif self.state == SCREENSAVER:
|
|
self.tank.update(dt)
|
|
self.tank.draw(self.screen)
|
|
|
|
pygame.display.flip()
|
|
|
|
def close(self):
|
|
self.camera.close()
|
|
pygame.quit()
|
|
|
|
|
|
def main():
|
|
reexec_keep_awake()
|
|
windowed = "--windowed" in sys.argv
|
|
booth = Booth(windowed=windowed)
|
|
try:
|
|
booth.run()
|
|
finally:
|
|
booth.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|