| """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) |
|
|
