File size: 3,075 Bytes
e8b6587 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 | """Process-local filesystem, logging, token, and request-path support for the app."""
from __future__ import annotations
import logging
import os
import sys
import tempfile
import time
import uuid
from pathlib import Path
import PIL.Image
from .build_info import CANDIDATE_ID
from . import runtime_utils
APP_ROOT = Path(__file__).resolve().parents[1]
RUNTIME_EXAMPLE_DIR = APP_ROOT / "runtime_examples"
RUNTIME_IC_EXAMPLE_DIR = RUNTIME_EXAMPLE_DIR / "ic_colorizer"
RUNTIME_IC_PIXEL_UPSCALER_EXAMPLE_DIR = RUNTIME_EXAMPLE_DIR / "ic_pixel_upscaler"
RUNTIME_IC_INOUTPAINT_EXAMPLE_DIR = RUNTIME_EXAMPLE_DIR / "ic_inoutpaint"
RUNTIME_A2V_EXAMPLE_DIR = RUNTIME_EXAMPLE_DIR / "audio_to_video"
HF_TOKEN = os.environ.get("HF_TOKEN") or os.environ.get("HF_ACCESS_TOKEN")
SPACE_DEBUG = os.environ.get("SPACE_DEBUG") == "1"
LOG_FORMAT = "%(asctime)s.%(msecs)03dZ %(levelname)s %(name)s %(message)s"
LOG_DATEFMT = "%Y-%m-%dT%H:%M:%S"
LOG_FORMATTER = logging.Formatter(LOG_FORMAT, datefmt=LOG_DATEFMT)
LOG_FORMATTER.converter = time.gmtime
APP_LOGGER = logging.getLogger("ltx25")
APP_LOGGER.setLevel(logging.DEBUG)
APP_LOGGER.propagate = False
if not APP_LOGGER.handlers:
_stream_handler = logging.StreamHandler(sys.stdout)
_stream_handler.setLevel(logging.INFO)
_stream_handler.setFormatter(LOG_FORMATTER)
APP_LOGGER.addHandler(_stream_handler)
WORKER_UUID = uuid.uuid4().hex
WORKER_ROOT = Path(tempfile.gettempdir()) / "ltx25_workers" / WORKER_UUID
WORKER_ROOT.mkdir(parents=True, mode=0o700, exist_ok=False)
CIVITAI_LORA_CACHE_ROOT = WORKER_ROOT / "civitai_lora_cache"
RUNTIME_CIVITAI_NO_PREVIEW_PATH = WORKER_ROOT / "civitai_no_preview.png"
PIL.Image.new("RGB", (256, 256), (38, 38, 38)).save(RUNTIME_CIVITAI_NO_PREVIEW_PATH, format="PNG")
WORKER_LOG_PATH = WORKER_ROOT / "ltx25_worker.log"
_worker_file_handler = logging.FileHandler(WORKER_LOG_PATH, mode="a", encoding="utf-8", delay=True)
_worker_file_handler.setLevel(logging.INFO)
_worker_file_handler.setFormatter(LOG_FORMATTER)
APP_LOGGER.addHandler(_worker_file_handler)
APP_LOGGER.info("worker.start candidate=%s worker_uuid=%s root=%s", CANDIDATE_ID, WORKER_UUID, WORKER_ROOT)
RequestPaths = runtime_utils.RequestPaths
is_uuid_hex = runtime_utils.is_uuid_hex
create_session_id = runtime_utils.create_session_id
normalize_session_id = runtime_utils.normalize_session_id
disk_state = runtime_utils.disk_state
gpu_state = runtime_utils.gpu_state
package_identity = runtime_utils.package_identity
runtime_environment_identity = runtime_utils.runtime_environment_identity
sha256_file = runtime_utils.sha256_file
def open_request_logger(log_path: Path, request_id: str):
return runtime_utils.open_request_logger(log_path, request_id, LOG_FORMATTER)
def close_request_logger(logger, handler) -> None:
runtime_utils.close_request_logger(logger, handler)
def cleanup_session_results(session_id) -> None:
runtime_utils.cleanup_session_results(WORKER_ROOT, session_id)
def create_request_paths(session_id) -> RequestPaths:
return runtime_utils.create_request_paths(WORKER_ROOT, session_id)
|