darkmedia-x-api / engine /asset_processor.py
cybermedia's picture
Upload folder using huggingface_hub
343eed9 verified
import os
import subprocess
import shutil
from pathlib import Path
from typing import List, Dict, Optional
import numpy as np
class AssetProcessor:
"""Base class for all asset-level transformations (ImageMagick, Inkscape, etc.)."""
def __init__(self, name: str):
self.name = name
self.available = self._check_availability()
def _check_availability(self) -> bool:
return shutil.which(self.name.lower()) is not None
def process(self, input_path: Path, output_path: Path, options: Dict = None) -> bool:
"""Process a single asset."""
raise NotImplementedError("Subclasses must implement process()")
class ImageMagickProcessor(AssetProcessor):
def __init__(self):
# On Linux it's often 'magick' or 'convert'
name = "magick" if shutil.which("magick") else "convert"
super().__init__(name)
def process(self, input_path: Path, output_path: Path, options: Dict = None) -> bool:
if not self.available:
return False
# Example options: {"filters": "-sepia-tone 80% -charcoal 2"}
filters = options.get("filters", "").split() if options else []
cmd = [self.name, str(input_path)] + filters + [str(output_path)]
try:
subprocess.run(cmd, check=True, capture_output=True)
return True
except Exception as e:
print(f" [AssetProcessor] ImageMagick error: {e}")
return False
class InkscapeProcessor(AssetProcessor):
def __init__(self):
super().__init__("inkscape")
def process(self, input_path: Path, output_path: Path, options: Dict = None) -> bool:
if not self.available:
return False
# Export SVG to PNG
cmd = [self.name, "--export-type=png", "--export-filename=" + str(output_path), str(input_path)]
try:
subprocess.run(cmd, check=True, capture_output=True)
return True
except Exception as e:
print(f" [AssetProcessor] Inkscape error: {e}")
return False
class AsepriteProcessor(AssetProcessor):
def __init__(self):
super().__init__("aseprite")
def process(self, input_path: Path, output_path: Path, options: Dict = None) -> bool:
if not self.available:
return False
# Example: Pixel art conversion
# aseprite -b input.png --save-as output.png
cmd = [self.name, "-b", str(input_path), "--save-as", str(output_path)]
try:
subprocess.run(cmd, check=True, capture_output=True)
return True
except Exception as e:
print(f" [AssetProcessor] Aseprite error: {e}")
return False
class OpenCVAssetProcessor(AssetProcessor):
"""Bridge to VFXProcessor for static asset processing."""
def __init__(self):
super().__init__("python") # Uses python/cv2
try:
import cv2
self.available = True
except ImportError:
self.available = False
def process(self, input_path: Path, output_path: Path, options: Dict = None) -> bool:
if not self.available: return False
try:
import cv2
from vfx.opencv_effects import VFXProcessor
img = cv2.imread(str(input_path))
if img is None: return False
# Convert BGR to RGB for VFXProcessor
img_rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
profile = options.get("profile", "none")
vfx = VFXProcessor(profile)
processed_rgb = vfx.process_frame(img_rgb, 0.0)
# Convert back to BGR for saving
processed_bgr = cv2.cvtColor(processed_rgb, cv2.COLOR_RGB2BGR)
cv2.imwrite(str(output_path), processed_bgr)
return True
except Exception as e:
print(f" [AssetProcessor] OpenCV error: {e}")
return False
def get_processor(name: str) -> Optional[AssetProcessor]:
processors = {
"imagemagick": ImageMagickProcessor,
"inkscape": InkscapeProcessor,
"aseprite": AsepriteProcessor,
"opencv": OpenCVAssetProcessor
}
cls = processors.get(name.lower())
return cls() if cls else None
def apply_artistic_profile(story_dir: Path, profile_name: str):
"""
Applies an artistic profile to all images in a story directory.
Profiles: 'pixel_art', 'sketch', 'watercolor', 'vhs_static'
"""
img_dir = story_dir / "assets" / "images"
if not img_dir.exists(): return
output_dir = story_dir / "assets" / "images_processed"
if profile_name == "none":
if output_dir.exists():
print(f" 🧹 Profil 'none' demandé : suppression des images processées.")
shutil.rmtree(output_dir)
return
output_dir.mkdir(exist_ok=True)
# --- CACHE CHECK ---
cache_file = output_dir / ".last_profile"
if cache_file.exists():
try:
last_profile = cache_file.read_text().strip()
# If profile is same and some images exist, skip
if last_profile == profile_name and any(output_dir.glob("scene_*.png")):
print(f" ✨ Profil '{profile_name}' déjà appliqué. On saute le re-processing.")
return
except: pass
processor = None
options = {}
# ... (rest of the logic remains)
if profile_name == "pixel_art":
processor = get_processor("aseprite")
elif profile_name == "sketch":
processor = get_processor("imagemagick")
options = {"filters": "-colorspace gray -edge 1 -negate"}
elif profile_name == "oil_paint":
processor = get_processor("imagemagick")
options = {"filters": "-paint 3"}
elif profile_name == "charcoal":
processor = get_processor("imagemagick")
options = {"filters": "-charcoal 2"}
elif profile_name == "vintage":
processor = get_processor("imagemagick")
options = {"filters": "-sepia-tone 80%"}
elif profile_name == "night_vision":
processor = get_processor("imagemagick")
options = {"filters": "-colorspace gray -fill #00ff00 -tint 100"}
elif profile_name == "vhs_static":
processor = get_processor("opencv")
options = {"profile": "vhs_pro"}
if not processor or not processor.available:
print(f" [AssetProcessor] Profile '{profile_name}' not available (tool missing).")
return
print(f" 🎨 Applying Artistic Profile: {profile_name}...")
processed_count = 0
for img_path in img_dir.glob("scene_*.png"):
out_path = output_dir / img_path.name
if processor.process(img_path, out_path, options):
processed_count += 1
# Save to cache if at least one image was processed
if processed_count > 0:
cache_file.write_text(profile_name)