|
""" |
|
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 |
|
|
|
|
|
SERVER_FEATURE_FLAGS: Dict[str, Any] = { |
|
"supports_preview_metadata": True, |
|
"max_upload_size": args.max_upload_size * 1024 * 1024, |
|
} |
|
|
|
|
|
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() |
|
|