| 1 | """RenderBackend: config-driven factory for image and video generators. |
| 2 | |
| 3 | Reads the ``image_generator`` and ``video_generator`` sections from a |
| 4 | ViMax YAML config, instantiates the concrete classes via *class_path*, |
| 5 | and wires up rate limiters. |
| 6 | |
| 7 | Usage:: |
| 8 | |
| 9 | backend = RenderBackend.from_config(config) |
| 10 | image = await backend.image_generator.generate_single_image(...) |
| 11 | video = await backend.video_generator.generate_single_video(...) |
| 12 | """ |
| 13 | |
| 14 | import importlib |
| 15 | import logging |
| 16 | from dataclasses import dataclass |
| 17 | from typing import Any, Dict |
| 18 | |
| 19 | from utils.rate_limiter import RateLimiter |
| 20 | |
| 21 | |
| 22 | @dataclass |
| 23 | class RenderBackend: |
| 24 | """Bundles an image generator and a video generator.""" |
| 25 | |
| 26 | image_generator: Any |
| 27 | video_generator: Any |
| 28 | |
| 29 | @classmethod |
| 30 | def from_config(cls, config: Dict[str, Any]) -> "RenderBackend": |
| 31 | """Build a RenderBackend from a parsed YAML config dict. |
| 32 | |
| 33 | Rate limiters are created from ``max_requests_per_minute`` / |
| 34 | ``max_requests_per_day`` if present in each generator section. |
| 35 | """ |
| 36 | img_cfg = config["image_generator"] |
| 37 | vid_cfg = config["video_generator"] |
| 38 | |
| 39 | image_gen = _instantiate(img_cfg, _build_rate_limiter(img_cfg)) |
| 40 | video_gen = _instantiate(vid_cfg, _build_rate_limiter(vid_cfg)) |
| 41 | |
| 42 | logging.info("RenderBackend: image=%s, video=%s", |
| 43 | img_cfg["class_path"], vid_cfg["class_path"]) |
| 44 | |
| 45 | return cls(image_generator=image_gen, video_generator=video_gen) |
| 46 | |
| 47 | |
| 48 | def _build_rate_limiter(section: Dict[str, Any]) -> RateLimiter | None: |
| 49 | rpm = section.get("max_requests_per_minute") |
| 50 | rpd = section.get("max_requests_per_day") |
| 51 | if rpm or rpd: |
| 52 | return RateLimiter(max_requests_per_minute=rpm, max_requests_per_day=rpd) |
| 53 | return None |
| 54 | |
| 55 | |
| 56 | def _instantiate(section: Dict[str, Any], rate_limiter: RateLimiter | None) -> Any: |
| 57 | module_path, cls_name = section["class_path"].rsplit(".", 1) |
| 58 | cls = getattr(importlib.import_module(module_path), cls_name) |
| 59 | init_args = dict(section.get("init_args", {})) |
| 60 | if rate_limiter is not None: |
| 61 | init_args["rate_limiter"] = rate_limiter |
| 62 | return cls(**init_args) |
| 63 |