Spaces:
Running
on
Zero
Running
on
Zero
| """ | |
| Feature flags module for ComfyUI WebSocket protocol negotiation. | |
| This module handles capability negotiation between frontend and backend, | |
| allowing graceful protocol evolution while maintaining backward compatibility. | |
| """ | |
| from typing import Any, Dict | |
| from comfy.cli_args import args | |
| # Default server capabilities | |
| SERVER_FEATURE_FLAGS: Dict[str, Any] = { | |
| "supports_preview_metadata": True, | |
| "max_upload_size": args.max_upload_size * 1024 * 1024, # Convert MB to bytes | |
| } | |
| def get_connection_feature( | |
| sockets_metadata: Dict[str, Dict[str, Any]], | |
| sid: str, | |
| feature_name: str, | |
| default: Any = False | |
| ) -> Any: | |
| """ | |
| Get a feature flag value for a specific connection. | |
| Args: | |
| sockets_metadata: Dictionary of socket metadata | |
| sid: Session ID of the connection | |
| feature_name: Name of the feature to check | |
| default: Default value if feature not found | |
| Returns: | |
| Feature value or default if not found | |
| """ | |
| if sid not in sockets_metadata: | |
| return default | |
| return sockets_metadata[sid].get("feature_flags", {}).get(feature_name, default) | |
| def supports_feature( | |
| sockets_metadata: Dict[str, Dict[str, Any]], | |
| sid: str, | |
| feature_name: str | |
| ) -> bool: | |
| """ | |
| Check if a connection supports a specific feature. | |
| Args: | |
| sockets_metadata: Dictionary of socket metadata | |
| sid: Session ID of the connection | |
| feature_name: Name of the feature to check | |
| Returns: | |
| Boolean indicating if feature is supported | |
| """ | |
| return get_connection_feature(sockets_metadata, sid, feature_name, False) is True | |
| def get_server_features() -> Dict[str, Any]: | |
| """ | |
| Get the server's feature flags. | |
| Returns: | |
| Dictionary of server feature flags | |
| """ | |
| return SERVER_FEATURE_FLAGS.copy() | |