xinyun/ComfyUI
1
0
Fork 0
mirror of https://git.datalinker.icu/comfyanonymous/ComfyUI synced 2026-07-15 18:06:59 +08:00
Code Issues Packages Projects Releases Wiki Activity

chore(api-nodes): applied ruff's pyupgrade(python3.10) to api-nodes client's to folder

Browse Source
This commit is contained in:
bigcat88 2025-12-04 13:37:43 +02:00
parent 6be85c7920
commit 81e2cb7fdb
No known key found for this signature in database
GPG Key ID: 1F0BF0EC3CF22721
7 changed files with 114 additions and 118 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
comfy_api_nodes/util/_helpers.py
View File

@ -2,8 +2,8 @@ import asyncio
import contextlib import contextlib
import os import os
import time import time
from collections.abc import Callable
from io import BytesIO from io import BytesIO
from typing import Callable, Optional, Union
from comfy.cli_args import args from comfy.cli_args import args
from comfy.model_management import processing_interrupted from comfy.model_management import processing_interrupted
@ -35,12 +35,12 @@ def default_base_url() -> str:
async def sleep_with_interrupt( async def sleep_with_interrupt(
seconds: float, seconds: float,
node_cls: Optional[type[IO.ComfyNode]], node_cls: type[IO.ComfyNode] | None,
label: Optional[str] = None, label: str | None = None,
start_ts: Optional[float] = None, start_ts: float | None = None,
estimated_total: Optional[int] = None, estimated_total: int | None = None,
*, *,
display_callback: Optional[Callable[[type[IO.ComfyNode], str, int, Optional[int]], None]] = None, display_callback: Callable[[type[IO.ComfyNode], str, int, int | None], None] | None = None,
): ):
""" """
Sleep in 1s slices while: Sleep in 1s slices while:
@ -65,7 +65,7 @@ def mimetype_to_extension(mime_type: str) -> str:
return mime_type.split("/")[-1].lower() return mime_type.split("/")[-1].lower()
def get_fs_object_size(path_or_object: Union[str, BytesIO]) -> int: def get_fs_object_size(path_or_object: str | BytesIO) -> int:
if isinstance(path_or_object, str): if isinstance(path_or_object, str):
return os.path.getsize(path_or_object) return os.path.getsize(path_or_object)
return len(path_or_object.getvalue()) return len(path_or_object.getvalue())

145
comfy_api_nodes/util/client.py
View File

@ -4,10 +4,11 @@ import json
import logging import logging
import time import time
import uuid import uuid
from collections.abc import Callable, Iterable
from dataclasses import dataclass from dataclasses import dataclass
from enum import Enum from enum import Enum
from io import BytesIO from io import BytesIO
from typing import Any, Callable, Iterable, Literal, Optional, Type, TypeVar, Union from typing import Any, Literal, TypeVar
from urllib.parse import urljoin, urlparse from urllib.parse import urljoin, urlparse
import aiohttp import aiohttp
@ -37,8 +38,8 @@ class ApiEndpoint:
path: str, path: str,
method: Literal["GET", "POST", "PUT", "DELETE", "PATCH"] = "GET", method: Literal["GET", "POST", "PUT", "DELETE", "PATCH"] = "GET",
*, *,
query_params: Optional[dict[str, Any]] = None, query_params: dict[str, Any] | None = None,
headers: Optional[dict[str, str]] = None, headers: dict[str, str] | None = None,
): ):
self.path = path self.path = path
self.method = method self.method = method
@ -52,18 +53,18 @@ class _RequestConfig:
endpoint: ApiEndpoint endpoint: ApiEndpoint
timeout: float timeout: float
content_type: str content_type: str
data: Optional[dict[str, Any]] data: dict[str, Any] | None
files: Optional[Union[dict[str, Any], list[tuple[str, Any]]]] files: dict[str, Any] | list[tuple[str, Any]] | None
multipart_parser: Optional[Callable] multipart_parser: Callable | None
max_retries: int max_retries: int
retry_delay: float retry_delay: float
retry_backoff: float retry_backoff: float
wait_label: str = "Waiting" wait_label: str = "Waiting"
monitor_progress: bool = True monitor_progress: bool = True
estimated_total: Optional[int] = None estimated_total: int | None = None
final_label_on_success: Optional[str] = "Completed" final_label_on_success: str | None = "Completed"
progress_origin_ts: Optional[float] = None progress_origin_ts: float | None = None
price_extractor: Optional[Callable[[dict[str, Any]], Optional[float]]] = None price_extractor: Callable[[dict[str, Any]], float | None] | None = None
@dataclass @dataclass
@ -71,10 +72,10 @@ class _PollUIState:
started: float started: float
status_label: str = "Queued" status_label: str = "Queued"
is_queued: bool = True is_queued: bool = True
price: Optional[float] = None price: float | None = None
estimated_duration: Optional[int] = None estimated_duration: int | None = None
base_processing_elapsed: float = 0.0 # sum of completed active intervals base_processing_elapsed: float = 0.0 # sum of completed active intervals
active_since: Optional[float] = None # start time of current active interval (None if queued) active_since: float | None = None # start time of current active interval (None if queued)
_RETRY_STATUS = {408, 429, 500, 502, 503, 504} _RETRY_STATUS = {408, 429, 500, 502, 503, 504}
@ -87,20 +88,20 @@ async def sync_op(
cls: type[IO.ComfyNode], cls: type[IO.ComfyNode],
endpoint: ApiEndpoint, endpoint: ApiEndpoint,
*, *,
response_model: Type[M], response_model: type[M],
price_extractor: Optional[Callable[[M], Optional[float]]] = None, price_extractor: Callable[[M], float | None] | None = None,
data: Optional[BaseModel] = None, data: BaseModel | None = None,
files: Optional[Union[dict[str, Any], list[tuple[str, Any]]]] = None, files: dict[str, Any] | list[tuple[str, Any]] | None = None,
content_type: str = "application/json", content_type: str = "application/json",
timeout: float = 3600.0, timeout: float = 3600.0,
multipart_parser: Optional[Callable] = None, multipart_parser: Callable | None = None,
max_retries: int = 3, max_retries: int = 3,
retry_delay: float = 1.0, retry_delay: float = 1.0,
retry_backoff: float = 2.0, retry_backoff: float = 2.0,
wait_label: str = "Waiting for server", wait_label: str = "Waiting for server",
estimated_duration: Optional[int] = None, estimated_duration: int | None = None,
final_label_on_success: Optional[str] = "Completed", final_label_on_success: str | None = "Completed",
progress_origin_ts: Optional[float] = None, progress_origin_ts: float | None = None,
monitor_progress: bool = True, monitor_progress: bool = True,
) -> M: ) -> M:
raw = await sync_op_raw( raw = await sync_op_raw(
@ -131,22 +132,22 @@ async def poll_op(
cls: type[IO.ComfyNode], cls: type[IO.ComfyNode],
poll_endpoint: ApiEndpoint, poll_endpoint: ApiEndpoint,
*, *,
response_model: Type[M], response_model: type[M],
status_extractor: Callable[[M], Optional[Union[str, int]]], status_extractor: Callable[[M], str | int | None],
progress_extractor: Optional[Callable[[M], Optional[int]]] = None, progress_extractor: Callable[[M], int | None] | None = None,
price_extractor: Optional[Callable[[M], Optional[float]]] = None, price_extractor: Callable[[M], float | None] | None = None,
completed_statuses: Optional[list[Union[str, int]]] = None, completed_statuses: list[str | int] | None = None,
failed_statuses: Optional[list[Union[str, int]]] = None, failed_statuses: list[str | int] | None = None,
queued_statuses: Optional[list[Union[str, int]]] = None, queued_statuses: list[str | int] | None = None,
data: Optional[BaseModel] = None, data: BaseModel | None = None,
poll_interval: float = 5.0, poll_interval: float = 5.0,
max_poll_attempts: int = 120, max_poll_attempts: int = 120,
timeout_per_poll: float = 120.0, timeout_per_poll: float = 120.0,
max_retries_per_poll: int = 3, max_retries_per_poll: int = 3,
retry_delay_per_poll: float = 1.0, retry_delay_per_poll: float = 1.0,
retry_backoff_per_poll: float = 2.0, retry_backoff_per_poll: float = 2.0,
estimated_duration: Optional[int] = None, estimated_duration: int | None = None,
cancel_endpoint: Optional[ApiEndpoint] = None, cancel_endpoint: ApiEndpoint | None = None,
cancel_timeout: float = 10.0, cancel_timeout: float = 10.0,
) -> M: ) -> M:
raw = await poll_op_raw( raw = await poll_op_raw(
@ -178,22 +179,22 @@ async def sync_op_raw(
cls: type[IO.ComfyNode], cls: type[IO.ComfyNode],
endpoint: ApiEndpoint, endpoint: ApiEndpoint,
*, *,
price_extractor: Optional[Callable[[dict[str, Any]], Optional[float]]] = None, price_extractor: Callable[[dict[str, Any]], float | None] | None = None,
data: Optional[Union[dict[str, Any], BaseModel]] = None, data: dict[str, Any] | BaseModel | None = None,
files: Optional[Union[dict[str, Any], list[tuple[str, Any]]]] = None, files: dict[str, Any] | list[tuple[str, Any]] | None = None,
content_type: str = "application/json", content_type: str = "application/json",
timeout: float = 3600.0, timeout: float = 3600.0,
multipart_parser: Optional[Callable] = None, multipart_parser: Callable | None = None,
max_retries: int = 3, max_retries: int = 3,
retry_delay: float = 1.0, retry_delay: float = 1.0,
retry_backoff: float = 2.0, retry_backoff: float = 2.0,
wait_label: str = "Waiting for server", wait_label: str = "Waiting for server",
estimated_duration: Optional[int] = None, estimated_duration: int | None = None,
as_binary: bool = False, as_binary: bool = False,
final_label_on_success: Optional[str] = "Completed", final_label_on_success: str | None = "Completed",
progress_origin_ts: Optional[float] = None, progress_origin_ts: float | None = None,
monitor_progress: bool = True, monitor_progress: bool = True,
) -> Union[dict[str, Any], bytes]: ) -> dict[str, Any] | bytes:
""" """
Make a single network request. Make a single network request.
- If as_binary=False (default): returns JSON dict (or {'_raw': '<text>'} if non-JSON). - If as_binary=False (default): returns JSON dict (or {'_raw': '<text>'} if non-JSON).
@ -229,21 +230,21 @@ async def poll_op_raw(
cls: type[IO.ComfyNode], cls: type[IO.ComfyNode],
poll_endpoint: ApiEndpoint, poll_endpoint: ApiEndpoint,
*, *,
status_extractor: Callable[[dict[str, Any]], Optional[Union[str, int]]], status_extractor: Callable[[dict[str, Any]], str | int | None],
progress_extractor: Optional[Callable[[dict[str, Any]], Optional[int]]] = None, progress_extractor: Callable[[dict[str, Any]], int | None] | None = None,
price_extractor: Optional[Callable[[dict[str, Any]], Optional[float]]] = None, price_extractor: Callable[[dict[str, Any]], float | None] | None = None,
completed_statuses: Optional[list[Union[str, int]]] = None, completed_statuses: list[str | int] | None = None,
failed_statuses: Optional[list[Union[str, int]]] = None, failed_statuses: list[str | int] | None = None,
queued_statuses: Optional[list[Union[str, int]]] = None, queued_statuses: list[str | int] | None = None,
data: Optional[Union[dict[str, Any], BaseModel]] = None, data: dict[str, Any] | BaseModel | None = None,
poll_interval: float = 5.0, poll_interval: float = 5.0,
max_poll_attempts: int = 120, max_poll_attempts: int = 120,
timeout_per_poll: float = 120.0, timeout_per_poll: float = 120.0,
max_retries_per_poll: int = 3, max_retries_per_poll: int = 3,
retry_delay_per_poll: float = 1.0, retry_delay_per_poll: float = 1.0,
retry_backoff_per_poll: float = 2.0, retry_backoff_per_poll: float = 2.0,
estimated_duration: Optional[int] = None, estimated_duration: int | None = None,
cancel_endpoint: Optional[ApiEndpoint] = None, cancel_endpoint: ApiEndpoint | None = None,
cancel_timeout: float = 10.0, cancel_timeout: float = 10.0,
) -> dict[str, Any]: ) -> dict[str, Any]:
""" """
@ -261,7 +262,7 @@ async def poll_op_raw(
consumed_attempts = 0 # counts only non-queued polls consumed_attempts = 0 # counts only non-queued polls
progress_bar = utils.ProgressBar(100) if progress_extractor else None progress_bar = utils.ProgressBar(100) if progress_extractor else None
last_progress: Optional[int] = None last_progress: int | None = None
state = _PollUIState(started=started, estimated_duration=estimated_duration) state = _PollUIState(started=started, estimated_duration=estimated_duration)
stop_ticker = asyncio.Event() stop_ticker = asyncio.Event()
@ -420,10 +421,10 @@ async def poll_op_raw(
def _display_text( def _display_text(
node_cls: type[IO.ComfyNode], node_cls: type[IO.ComfyNode],
text: Optional[str], text: str | None,
*, *,
status: Optional[Union[str, int]] = None, status: str | int | None = None,
price: Optional[float] = None, price: float | None = None,
) -> None: ) -> None:
display_lines: list[str] = [] display_lines: list[str] = []
if status: if status:
@ -440,13 +441,13 @@ def _display_text(
def _display_time_progress( def _display_time_progress(
node_cls: type[IO.ComfyNode], node_cls: type[IO.ComfyNode],
status: Optional[Union[str, int]], status: str | int | None,
elapsed_seconds: int, elapsed_seconds: int,
estimated_total: Optional[int] = None, estimated_total: int | None = None,
*, *,
price: Optional[float] = None, price: float | None = None,
is_queued: Optional[bool] = None, is_queued: bool | None = None,
processing_elapsed_seconds: Optional[int] = None, processing_elapsed_seconds: int | None = None,
) -> None: ) -> None:
if estimated_total is not None and estimated_total > 0 and is_queued is False: if estimated_total is not None and estimated_total > 0 and is_queued is False:
pe = processing_elapsed_seconds if processing_elapsed_seconds is not None else elapsed_seconds pe = processing_elapsed_seconds if processing_elapsed_seconds is not None else elapsed_seconds
@ -488,7 +489,7 @@ def _unpack_tuple(t: tuple) -> tuple[str, Any, str]:
raise ValueError("files tuple must be (filename, file[, content_type])") raise ValueError("files tuple must be (filename, file[, content_type])")
def _merge_params(endpoint_params: dict[str, Any], method: str, data: Optional[dict[str, Any]]) -> dict[str, Any]: def _merge_params(endpoint_params: dict[str, Any], method: str, data: dict[str, Any] | None) -> dict[str, Any]:
params = dict(endpoint_params or {}) params = dict(endpoint_params or {})
if method.upper() == "GET" and data: if method.upper() == "GET" and data:
for k, v in data.items(): for k, v in data.items():
@ -534,9 +535,9 @@ def _generate_operation_id(method: str, path: str, attempt: int) -> str:
def _snapshot_request_body_for_logging( def _snapshot_request_body_for_logging(
content_type: str, content_type: str,
method: str, method: str,
data: Optional[dict[str, Any]], data: dict[str, Any] | None,
files: Optional[Union[dict[str, Any], list[tuple[str, Any]]]], files: dict[str, Any] | list[tuple[str, Any]] | None,
) -> Optional[Union[dict[str, Any], str]]: ) -> dict[str, Any] | str | None:
if method.upper() == "GET": if method.upper() == "GET":
return None return None
if content_type == "multipart/form-data": if content_type == "multipart/form-data":
@ -586,13 +587,13 @@ async def _request_base(cfg: _RequestConfig, expect_binary: bool):
attempt = 0 attempt = 0
delay = cfg.retry_delay delay = cfg.retry_delay
operation_succeeded: bool = False operation_succeeded: bool = False
final_elapsed_seconds: Optional[int] = None final_elapsed_seconds: int | None = None
extracted_price: Optional[float] = None extracted_price: float | None = None
while True: while True:
attempt += 1 attempt += 1
stop_event = asyncio.Event() stop_event = asyncio.Event()
monitor_task: Optional[asyncio.Task] = None monitor_task: asyncio.Task | None = None
sess: Optional[aiohttp.ClientSession] = None sess: aiohttp.ClientSession | None = None
operation_id = _generate_operation_id(method, cfg.endpoint.path, attempt) operation_id = _generate_operation_id(method, cfg.endpoint.path, attempt)
logging.debug("[DEBUG] HTTP %s %s (attempt %d)", method, url, attempt) logging.debug("[DEBUG] HTTP %s %s (attempt %d)", method, url, attempt)
@ -887,7 +888,7 @@ async def _request_base(cfg: _RequestConfig, expect_binary: bool):
) )
def _validate_or_raise(response_model: Type[M], payload: Any) -> M: def _validate_or_raise(response_model: type[M], payload: Any) -> M:
try: try:
return response_model.model_validate(payload) return response_model.model_validate(payload)
except Exception as e: except Exception as e:
@ -902,9 +903,9 @@ def _validate_or_raise(response_model: Type[M], payload: Any) -> M:
def _wrap_model_extractor( def _wrap_model_extractor(
response_model: Type[M], response_model: type[M],
extractor: Optional[Callable[[M], Any]], extractor: Callable[[M], Any] | None,
) -> Optional[Callable[[dict[str, Any]], Any]]: ) -> Callable[[dict[str, Any]], Any] | None:
"""Wrap a typed extractor so it can be used by the dict-based poller. """Wrap a typed extractor so it can be used by the dict-based poller.
Validates the dict into `response_model` before invoking `extractor`. Validates the dict into `response_model` before invoking `extractor`.
Uses a small per-wrapper cache keyed by `id(dict)` to avoid re-validating Uses a small per-wrapper cache keyed by `id(dict)` to avoid re-validating
@ -929,10 +930,10 @@ def _wrap_model_extractor(
return _wrapped return _wrapped
def _normalize_statuses(values: Optional[Iterable[Union[str, int]]]) -> set[Union[str, int]]: def _normalize_statuses(values: Iterable[str | int] | None) -> set[str | int]:
if not values: if not values:
return set() return set()
out: set[Union[str, int]] = set() out: set[str | int] = set()
for v in values: for v in values:
nv = _normalize_status_value(v) nv = _normalize_status_value(v)
if nv is not None: if nv is not None:
@ -940,7 +941,7 @@ def _normalize_statuses(values: Optional[Iterable[Union[str, int]]]) -> set[Unio
return out return out
def _normalize_status_value(val: Union[str, int, None]) -> Union[str, int, None]: def _normalize_status_value(val: str | int | None) -> str | int | None:
if isinstance(val, str): if isinstance(val, str):
return val.strip().lower() return val.strip().lower()
return val return val

9
comfy_api_nodes/util/conversions.py
View File

@ -4,7 +4,6 @@ import math
import mimetypes import mimetypes
import uuid import uuid
from io import BytesIO from io import BytesIO
from typing import Optional
import av import av
import numpy as np import numpy as np
@ -57,7 +56,7 @@ def image_tensor_pair_to_batch(image1: torch.Tensor, image2: torch.Tensor) -> to
def tensor_to_bytesio( def tensor_to_bytesio(
image: torch.Tensor, image: torch.Tensor,
name: Optional[str] = None, name: str | None = None,
total_pixels: int = 2048 * 2048, total_pixels: int = 2048 * 2048,
mime_type: str = "image/png", mime_type: str = "image/png",
) -> BytesIO: ) -> BytesIO:
@ -178,7 +177,7 @@ def audio_to_base64_string(audio: Input.Audio, container_format: str = "mp4", co
def video_to_base64_string( def video_to_base64_string(
video: Input.Video, video: Input.Video,
container_format: VideoContainer = None, container_format: VideoContainer = None,
codec: VideoCodec = None codec: VideoCodec = None,
) -> str: ) -> str:
""" """
Converts a video input to a base64 string. Converts a video input to a base64 string.
@ -191,8 +190,8 @@ def video_to_base64_string(
video_bytes_io = BytesIO() video_bytes_io = BytesIO()
# Use provided format/codec if specified, otherwise use video's own if available # Use provided format/codec if specified, otherwise use video's own if available
format_to_use = container_format if container_format is not None else getattr(video, 'container', VideoContainer.MP4) format_to_use = container_format if container_format is not None else getattr(video, "container", VideoContainer.MP4)
codec_to_use = codec if codec is not None else getattr(video, 'codec', VideoCodec.H264) codec_to_use = codec if codec is not None else getattr(video, "codec", VideoCodec.H264)
video.save_to(video_bytes_io, format=format_to_use, codec=codec_to_use) video.save_to(video_bytes_io, format=format_to_use, codec=codec_to_use)
video_bytes_io.seek(0) video_bytes_io.seek(0)

14
comfy_api_nodes/util/download_helpers.py
View File

@ -3,7 +3,7 @@ import contextlib
import uuid import uuid
from io import BytesIO from io import BytesIO
from pathlib import Path from pathlib import Path
from typing import IO, Optional, Union from typing import IO
from urllib.parse import urljoin, urlparse from urllib.parse import urljoin, urlparse
import aiohttp import aiohttp
@ -29,9 +29,9 @@ _RETRY_STATUS = {408, 429, 500, 502, 503, 504}
async def download_url_to_bytesio( async def download_url_to_bytesio(
url: str, url: str,
dest: Optional[Union[BytesIO, IO[bytes], str, Path]], dest: BytesIO | IO[bytes] | str | Path | None,
*, *,
timeout: Optional[float] = None, timeout: float | None = None,
max_retries: int = 5, max_retries: int = 5,
retry_delay: float = 1.0, retry_delay: float = 1.0,
retry_backoff: float = 2.0, retry_backoff: float = 2.0,
@ -71,10 +71,10 @@ async def download_url_to_bytesio(
is_path_sink = isinstance(dest, (str, Path)) is_path_sink = isinstance(dest, (str, Path))
fhandle = None fhandle = None
session: Optional[aiohttp.ClientSession] = None session: aiohttp.ClientSession | None = None
stop_evt: Optional[asyncio.Event] = None stop_evt: asyncio.Event | None = None
monitor_task: Optional[asyncio.Task] = None monitor_task: asyncio.Task | None = None
req_task: Optional[asyncio.Task] = None req_task: asyncio.Task | None = None
try: try:
with contextlib.suppress(Exception): with contextlib.suppress(Exception):

2
comfy_api_nodes/util/request_logger.py
View File

@ -1,5 +1,3 @@
from __future__ import annotations
import datetime import datetime
import hashlib import hashlib
import json import json

7
comfy_api_nodes/util/upload_helpers.py
View File

@ -4,7 +4,6 @@ import logging
import time import time
import uuid import uuid
from io import BytesIO from io import BytesIO
from typing import Optional
from urllib.parse import urlparse from urllib.parse import urlparse
import aiohttp import aiohttp
@ -32,7 +31,7 @@ from .conversions import (
class UploadRequest(BaseModel): class UploadRequest(BaseModel):
file_name: str = Field(..., description="Filename to upload") file_name: str = Field(..., description="Filename to upload")
content_type: Optional[str] = Field( content_type: str | None = Field(
None, None,
description="Mime type of the file. For example: image/png, image/jpeg, video/mp4, etc.", description="Mime type of the file. For example: image/png, image/jpeg, video/mp4, etc.",
) )
@ -102,7 +101,7 @@ async def upload_video_to_comfyapi(
*, *,
container: VideoContainer = VideoContainer.MP4, container: VideoContainer = VideoContainer.MP4,
codec: VideoCodec = VideoCodec.H264, codec: VideoCodec = VideoCodec.H264,
max_duration: Optional[int] = None, max_duration: int | None = None,
wait_label: str | None = "Uploading", wait_label: str | None = "Uploading",
) -> str: ) -> str:
""" """
@ -220,7 +219,7 @@ async def upload_file(
return return
monitor_task = asyncio.create_task(_monitor()) monitor_task = asyncio.create_task(_monitor())
sess: Optional[aiohttp.ClientSession] = None sess: aiohttp.ClientSession | None = None
try: try:
try: try:
request_logger.log_request_response( request_logger.log_request_response(

41
comfy_api_nodes/util/validation_utils.py
View File

@ -1,5 +1,4 @@
import logging import logging
from typing import Optional
import torch import torch
@ -18,10 +17,10 @@ def get_image_dimensions(image: torch.Tensor) -> tuple[int, int]:
def validate_image_dimensions( def validate_image_dimensions(
image: torch.Tensor, image: torch.Tensor,
min_width: Optional[int] = None, min_width: int | None = None,
max_width: Optional[int] = None, max_width: int | None = None,
min_height: Optional[int] = None, min_height: int | None = None,
max_height: Optional[int] = None, max_height: int | None = None,
): ):
height, width = get_image_dimensions(image) height, width = get_image_dimensions(image)
@ -37,8 +36,8 @@ def validate_image_dimensions(
def validate_image_aspect_ratio( def validate_image_aspect_ratio(
image: torch.Tensor, image: torch.Tensor,
min_ratio: Optional[tuple[float, float]] = None, # e.g. (1, 4) min_ratio: tuple[float, float] | None = None, # e.g. (1, 4)
max_ratio: Optional[tuple[float, float]] = None, # e.g. (4, 1) max_ratio: tuple[float, float] | None = None, # e.g. (4, 1)
*, *,
strict: bool = True, # True -> (min, max); False -> [min, max] strict: bool = True, # True -> (min, max); False -> [min, max]
) -> float: ) -> float:
@ -54,8 +53,8 @@ def validate_image_aspect_ratio(
def validate_images_aspect_ratio_closeness( def validate_images_aspect_ratio_closeness(
first_image: torch.Tensor, first_image: torch.Tensor,
second_image: torch.Tensor, second_image: torch.Tensor,
min_rel: float, # e.g. 0.8 min_rel: float, # e.g. 0.8
max_rel: float, # e.g. 1.25 max_rel: float, # e.g. 1.25
*, *,
strict: bool = False, # True -> (min, max); False -> [min, max] strict: bool = False, # True -> (min, max); False -> [min, max]
) -> float: ) -> float:
@ -84,8 +83,8 @@ def validate_images_aspect_ratio_closeness(
def validate_aspect_ratio_string( def validate_aspect_ratio_string(
aspect_ratio: str, aspect_ratio: str,
min_ratio: Optional[tuple[float, float]] = None, # e.g. (1, 4) min_ratio: tuple[float, float] | None = None, # e.g. (1, 4)
max_ratio: Optional[tuple[float, float]] = None, # e.g. (4, 1) max_ratio: tuple[float, float] | None = None, # e.g. (4, 1)
*, *,
strict: bool = False, # True -> (min, max); False -> [min, max] strict: bool = False, # True -> (min, max); False -> [min, max]
) -> float: ) -> float:
@ -97,10 +96,10 @@ def validate_aspect_ratio_string(
def validate_video_dimensions( def validate_video_dimensions(
video: Input.Video, video: Input.Video,
min_width: Optional[int] = None, min_width: int | None = None,
max_width: Optional[int] = None, max_width: int | None = None,
min_height: Optional[int] = None, min_height: int | None = None,
max_height: Optional[int] = None, max_height: int | None = None,
): ):
try: try:
width, height = video.get_dimensions() width, height = video.get_dimensions()
@ -120,8 +119,8 @@ def validate_video_dimensions(
def validate_video_duration( def validate_video_duration(
video: Input.Video, video: Input.Video,
min_duration: Optional[float] = None, min_duration: float | None = None,
max_duration: Optional[float] = None, max_duration: float | None = None,
): ):
try: try:
duration = video.get_duration() duration = video.get_duration()
@ -144,8 +143,8 @@ def get_number_of_images(images):
def validate_audio_duration( def validate_audio_duration(
audio: Input.Audio, audio: Input.Audio,
min_duration: Optional[float] = None, min_duration: float | None = None,
max_duration: Optional[float] = None, max_duration: float | None = None,
) -> None: ) -> None:
sr = int(audio["sample_rate"]) sr = int(audio["sample_rate"])
dur = int(audio["waveform"].shape[-1]) / sr dur = int(audio["waveform"].shape[-1]) / sr
@ -194,8 +193,8 @@ def _ratio_from_tuple(r: tuple[float, float]) -> float:
def _assert_ratio_bounds( def _assert_ratio_bounds(
ar: float, ar: float,
*, *,
min_ratio: Optional[tuple[float, float]] = None, min_ratio: tuple[float, float] | None = None,
max_ratio: Optional[tuple[float, float]] = None, max_ratio: tuple[float, float] | None = None,
strict: bool = True, strict: bool = True,
) -> None: ) -> None:
"""Validate a numeric aspect ratio against optional min/max ratio bounds.""" """Validate a numeric aspect ratio against optional min/max ratio bounds."""