| import json |
| import os |
| from dataclasses import dataclass |
|
|
|
|
| def read_string(name: str, default: str) -> str: |
| raw_value = os.environ.get(name) |
| if raw_value is None: |
| return default |
| stripped_value = raw_value.strip() |
| if stripped_value == "": |
| return default |
| return stripped_value |
|
|
|
|
| def read_integer(name: str, default: int) -> int: |
| raw_value = os.environ.get(name) |
| if raw_value is None or raw_value.strip() == "": |
| return default |
| return int(raw_value.strip()) |
|
|
|
|
| def read_float(name: str, default: float) -> float: |
| raw_value = os.environ.get(name) |
| if raw_value is None or raw_value.strip() == "": |
| return default |
| return float(raw_value.strip()) |
|
|
|
|
| def read_boolean(name: str, default: bool) -> bool: |
| raw_value = os.environ.get(name) |
| if raw_value is None or raw_value.strip() == "": |
| return default |
| return raw_value.strip().lower() in ("1", "true", "yes", "on") |
|
|
|
|
| def read_list(name: str, default: list) -> list: |
| raw_value = os.environ.get(name) |
| if raw_value is None or raw_value.strip() == "": |
| return default |
| items = [] |
| for piece in raw_value.split(","): |
| cleaned_piece = piece.strip() |
| if cleaned_piece != "": |
| items.append(cleaned_piece) |
| return items |
|
|
|
|
| @dataclass(frozen=True) |
| class Settings: |
| model_id: str |
| base_model_id: str |
| device: str |
| fps: float |
| seconds_per_word: float |
| minimum_seconds: float |
| maximum_seconds: float |
| maximum_words_per_sentence: int |
| use_retrieval: bool |
| do_sample: bool |
| temperature: float |
| top_k: int |
| seed: int |
| max_keywords: int |
| maximum_text_characters: int |
| maximum_sentences_per_request: int |
| cache_size: int |
| frames_per_message: int |
| queue_size: int |
| maximum_pending_requests_per_connection: int |
| maximum_connections: int |
| data_channel_high_water_mark: int |
| idle_peer_timeout_seconds: float |
| allowed_origins: list |
| api_token: str |
| ice_servers: list |
|
|
| @classmethod |
| def from_environment(cls) -> "Settings": |
| default_ice_servers = '[{"urls": ["stun:stun.l.google.com:19302"]}]' |
| raw_ice_servers = read_string("ICE_SERVERS_JSON", default_ice_servers) |
| parsed_ice_servers = json.loads(raw_ice_servers) |
|
|
| if not isinstance(parsed_ice_servers, list): |
| raise ValueError("ICE_SERVERS_JSON must be a JSON list") |
|
|
| for ice_server in parsed_ice_servers: |
| if not isinstance(ice_server, dict) or "urls" not in ice_server: |
| raise ValueError("Every ICE server entry needs a urls field") |
|
|
| return cls( |
| model_id=read_string("MODEL_ID", "Ivystream/RoboSTAR"), |
| base_model_id=read_string("BASE_MODEL_ID", "google/mt5-large"), |
| device=read_string("DEVICE", "cuda"), |
| fps=read_float("FPS", 20.0), |
| seconds_per_word=read_float("SECONDS_PER_WORD", 0.7), |
| minimum_seconds=read_float("MINIMUM_SECONDS", 2.0), |
| maximum_seconds=read_float("MAXIMUM_SECONDS", 10.0), |
| maximum_words_per_sentence=read_integer("MAXIMUM_WORDS_PER_SENTENCE", 18), |
| use_retrieval=read_boolean("USE_RETRIEVAL", False), |
| do_sample=read_boolean("DO_SAMPLE", False), |
| temperature=read_float("TEMPERATURE", 1.0), |
| top_k=read_integer("TOP_K", 0), |
| seed=read_integer("SEED", 20260703), |
| max_keywords=read_integer("MAX_KEYWORDS", 3), |
| maximum_text_characters=read_integer("MAXIMUM_TEXT_CHARACTERS", 400), |
| maximum_sentences_per_request=read_integer("MAXIMUM_SENTENCES_PER_REQUEST", 8), |
| cache_size=read_integer("CACHE_SIZE", 256), |
| frames_per_message=read_integer("FRAMES_PER_MESSAGE", 20), |
| queue_size=read_integer("QUEUE_SIZE", 64), |
| maximum_pending_requests_per_connection=read_integer("MAXIMUM_PENDING_REQUESTS", 4), |
| maximum_connections=read_integer("MAXIMUM_CONNECTIONS", 16), |
| data_channel_high_water_mark=read_integer("DATA_CHANNEL_HIGH_WATER_MARK", 1048576), |
| idle_peer_timeout_seconds=read_float("IDLE_PEER_TIMEOUT_SECONDS", 30.0), |
| allowed_origins=read_list("ALLOWED_ORIGINS", ["*"]), |
| api_token=read_string("API_TOKEN", ""), |
| ice_servers=parsed_ice_servers, |
| ) |
|
|
