summary | shortlog | log | commit | commitdiff | tree
raw | patch | inline | side by side (parent: 6a61f63)
raw | patch | inline | side by side (parent: 6a61f63)
| author | sbkelley <sb24kelley@gmail.com> | |
| Sun, 7 Dec 2025 17:22:19 +0000 (12:22 -0500) | ||
| committer | sbkelley <sb24kelley@gmail.com> | |
| Sun, 7 Dec 2025 17:22:27 +0000 (12:22 -0500) |
diff --git a/src/obs_ctl/app.py b/src/obs_ctl/app.py
index 040a664a590416ba451c68c95e980dc27e5c986e..79819d9193905f2af8b6e45caf1fdcb75d984216 100644 (file)
--- a/src/obs_ctl/app.py
+++ b/src/obs_ctl/app.py
from textual.widgets import Footer, Header
# need to import log to initalize logging
-from . import data, log # type: ignore
+from . import data, const, log # type: ignore
from .interface import RecordControls, StatsDisplay, SettingsBlock, obs, OBSWebSocketPane, OBSSettingsPane
logger = getLogger(__name__)
def compose(self):
yield Header()
yield Footer()
- yield obs.API(1, id="api")
+ yield obs.API(1, ws_host=const.HOST, ws_port=const.PORT, ws_password=const.PASSWORD, ws_timeout=const.TIMEOUT, id="api")
logger.debug("Composing OBSCtlApp")
with ScrollableContainer(classes="scrollbox"):
with VerticalGroup(classes="content"):
def on_mount(self, message: events.Mount):
- self.on_obs_ws_connected(None)
+ obsws_pane = self.get_websocket_pane()
+ api = self.get_api()
+ obsws_pane.host = api.ws_host
+ obsws_pane.port = api.ws_port
+ obsws_pane.password = api.ws_password
+ obsws_pane.timeout = api.ws_timeout
@on(obs.API.Report, "#api")
diff --git a/src/obs_ctl/const.py b/src/obs_ctl/const.py
index fd992a6228e4c2335031d20cc625fc3852d0db44..67b860664749f49a8536603e16f224c53a3d4969 100644 (file)
--- a/src/obs_ctl/const.py
+++ b/src/obs_ctl/const.py
POSSIBLE_CONFIG: list[str] = [USERDIRS.user_config_dir]
CURRENT_CONFIG = CONFIG.read(POSSIBLE_CONFIG)
+
+
+HOST = "localhost"
+PORT = 4455
+PASSWORD = ""
+TIMEOUT = 1.0
\ No newline at end of file
diff --git a/src/obs_ctl/interface/molecule/settings_block/pane.py b/src/obs_ctl/interface/molecule/settings_block/pane.py
index 8855c6bbde1ad2675a5b8fe268b0535b95012ecc..1f4cc964555409f079c24c61f10c26dff374050c 100644 (file)
from enum import Enum
-from functools import partial
from itertools import zip_longest
from logging import getLogger
-from typing import Any, Mapping, Sequence, overload
+from typing import Any, Mapping, Sequence
+
from textual import on
from textual.app import ComposeResult
from textual.containers import ScrollableContainer, Grid
from textual.reactive import reactive
+from textual.widget import Widget
from textual.widgets import TabPane, Input, Label, Select
+
from ...message import MSGBase
-from textual.widget import Widget
logger = getLogger(__name__)
+
class SettingsMSGBase(MSGBase):
- # value: Any
def __init__(self, control: Widget, value: Any, *, _log_me: bool=True):
super().__init__(control, _log_me=_log_me)
self.value = value
input.value = str(value)
return watch_func
-
def make_on_control_message(
listener_message: SettingsWidgetMsg,
listener_id: str,
reactive: reactive[Any],
post_message: type[SettingsMSGBase]):
def on_func(self: PaneBase, event: SettingsWidgetMsg):
- self.set_reactive(reactive, event.value)
+ reactive = event.value
self.post_message(post_message(self, event.value))
return on(listener_message, listener_id)(on_func)
-
def make_compose():
def compose(self: PaneBase) -> ComposeResult:
with ScrollableContainer(classes="--container"):
id=ctrl_id)
return compose
-
new_dict: dict[str, Any] = dict()
for setting_name, setting_type, setting_default, _ in loop_list:
- control_setting_id: str = f"{setting_type.widget.__name__.lower()}_{setting_name}"
+ setting_control_id: str = f"{setting_type.widget.__name__.lower()}_{setting_name}"
SettingChanged_name: str = f"{_snake_to_camel(setting_name)}Changed"
watch_setting_name: str = f"watch_{setting_name}"
on_setting_changed_name: str = f"on_{setting_name}_changed"
# Set up a record_path reactive
new_dict[setting_name] = reactive(setting_type.type(setting_default))
# Set up a watch_record_path function
- new_dict[watch_setting_name] = make_watch_reactive(f"#{control_setting_id}")
+ new_dict[watch_setting_name] = make_watch_reactive(f"#{setting_control_id}")
# Set up a RecordPathChanged message for when user interacts with control
# (see SettingTypes enum)
new_dict[SettingChanged_name] = type(SettingChanged_name, (SettingsMSGBase, ), {})
# and post RecordPathChanged
new_dict[on_setting_changed_name] = make_on_control_message(
listener_message=setting_type.msg,
- listener_id=f"#{control_setting_id}",
+ listener_id=f"#{setting_control_id}",
reactive=new_dict[setting_name],
post_message=new_dict[SettingChanged_name],
)
diff --git a/src/obs_ctl/interface/molecule/settings_block/settings_block.py b/src/obs_ctl/interface/molecule/settings_block/settings_block.py
index 2149bfedc53927a432a15b8b8428d4060991f5a2..93c20110e64c8dd2f9f2d4252cb9730e72c8051a 100644 (file)
OBSWebSocketPane = pane.make_pane(
"OBSWebSocketPane",
- pane.make_pane_widget("host", pane.SettingsWidget.STRING, "localhost", compact=True),
- pane.make_pane_widget("port", pane.SettingsWidget.INT, 4455, compact=True),
+ pane.make_pane_widget("host", pane.SettingsWidget.STRING, "", compact=True),
+ pane.make_pane_widget("port", pane.SettingsWidget.INT, 0, compact=True),
pane.make_pane_widget("password", pane.SettingsWidget.STRING, "", compact=True),
- pane.make_pane_widget("timeout", pane.SettingsWidget.FLOAT, 1.0, compact=True),
+ pane.make_pane_widget("timeout", pane.SettingsWidget.FLOAT, 0.0, compact=True),
)
OBSCTLPane = pane.make_pane(
index 8217b0368cb5d4ff2e99da989eb559fd7ca59abd..b213f152e8e17f8e777f190442e54fa535645dd8 100644 (file)
available_disk_space: float = 0.0
"""Available space in MB"""
record_directory: str = ""
+ websocket_host: str = ""
+ websocket_port: int = 0
+ websocket_password: str = ""
+ websocket_timeout: float = 0.0
@property
def estimated_hours_remaining(self):
class APIOutputMSGBase(MSGBase):
state = None
def __init__(self, control: "API", data: Any) -> None:
- super().__init__(control)
+ super().__init__(control, _log_me=True)
_state = API.OutputStates(data.output_state)
- logger.debug(f"{self} is an obs.API Message; state: {_state.name}")
+ logger.debug(f"{self} is an obs.API Output Message; state: {_state.name}")
self.state = _state
class StreamStateUpdated(APIOutputMSGBase):
class Report(MSGBase):
"""Posted ever self.interval second. data is an OBSStats object."""
def __init__(self: Self, control: "API", data: OBSStats):
- # this message is sent too often to log it.
- super().__init__(control, _log_me = False)
+ super().__init__(control)
self.data = data
ws_host: str = "localhost",
ws_port: int = 4455,
ws_password: str = "",
- ws_timeout: float = 3.0,
+ ws_timeout: float = 1.0,
*args: Any,
**kwargs: Any,
):
def on_mount(self):
self._reset_timer(self.interval)
-
-
- def on_stream_state_changed(self: Self, data: Any) -> None:
- self.post_message(API.StreamStateUpdated(self, data))
+
+
+ def watch_interval(self, value: float) -> None:
+ self._reset_timer(value)
- def on_record_state_changed(self: Self, data: Any) -> None:
- self.post_message(API.RecordStateUpdated(self, data))
+ def watch_ws_host(self, value: str) -> None:
+ if self.is_connected:
+ self.disconnect()
+
+
+ def watch_ws_port(self, value: int) -> None:
+ if self.is_connected:
+ self.disconnect()
-
- def watch_interval(self, value: float) -> None:
- self._reset_timer(value)
+
+ def watch_ws_password(self, value: str) -> None:
+ if self.is_connected:
+ self.disconnect()
+
+
+ def watch_ws_timeout(self, value: float) -> None:
+ if self.is_connected:
+ self.disconnect()
+
+
+ # LOGGER
+ def connect(self, host: str | None=None, port: int | None=None, password: str | None=None, timeout: float | None=None) -> bool:
+ if host is None:
+ host = self.ws_host
+ else:
+ self.ws_host = host
+ if port is None:
+ port = self.ws_port
+ else:
+ self.ws_port = port
+ if password is None:
+ password = self.ws_password
+ else:
+ self.ws_password = password
+ if timeout is None:
+ timeout = self.ws_timeout
+ else:
+ self.ws_timeout = timeout
+
+ if self._make_clients(host, port, password, timeout):
+ self.is_connected = True
+ self.event_client.callback.register(self.on_stream_state_changed)
+ self.event_client.callback.register(self.on_record_state_changed)
+ return True
+ else:
+ self.is_connected = False
+ return False
+
+
+ def disconnect(self) -> bool:
+ logger.info("Disconnecting from OBS WebSocket")
+ self.request_client.disconnect()
+ self.event_client.disconnect()
+ self.event_client.callback.deregister(self.on_stream_state_changed)
+ self.event_client.callback.deregister(self.on_record_state_changed)
+ self.is_connected = False
+ return True
def stop_stream(self) -> None:
available_disk_space=obs_stats.available_disk_space,
record_directory=record_directory.record_directory,
+
+ websocket_host=self.ws_host,
+ websocket_port=self.ws_port,
+ websocket_password=self.ws_password,
+ websocket_timeout=self.ws_timeout,
)
return output
+
+ # Callbacks for obsws API rather than Textual
+ def on_stream_state_changed(self: Self, data: Any) -> None:
+ self.post_message(API.StreamStateUpdated(self, data))
- # LOGGER
- def connect(self, host: str | None=None, port: int | None=None, password: str | None=None, timeout: float | None=None) -> None:
- host = self.ws_host if host is None else host
- port = self.ws_port if port is None else port
- password = self.ws_password if password is None else password
- timeout = self.ws_timeout if timeout is None else timeout
- self.ws_host = host
- self.ws_port = port
- self.ws_password = password
- self.ws_timeout = timeout
+
+ def on_record_state_changed(self: Self, data: Any) -> None:
+ self.post_message(API.RecordStateUpdated(self, data))
+
+
+ def _make_clients(self, host: str, port: int, password: str, timeout: float) -> bool:
try:
logger.info(f"Connecting to OBS WebSocket at {host}:{port}")
self.request_client = obs.ReqClient(
timeout=timeout
)
except (ConnectionRefusedError, TimeoutError) as err:
- logger.debug(err)
+ logger.error(err)
+ return False
else:
- self.is_connected = True
- self.event_client.callback.register(self.on_stream_state_changed) # type: ignore
- self.event_client.callback.register(self.on_record_state_changed) # type: ignore
-
-
- def disconnect(self) -> None:
- logger.info("Disconnecting from OBS WebSocket")
- self.request_client.disconnect()
- self.event_client.disconnect()
- self.is_connected = False
-
+ return True
+
# Regarding the pattern in _try_request --
# number = 5
# while number > 0
time.sleep(TRY_AGAIN)
continue
- elif err.code in API.KnownReqErrs: # type: ignore
- logger.debug(f"Error handled: {API.KnownReqErrs(err.code)}") # type: ignore
+ elif err.code in API.KnownReqErrs:
+ logger.debug(f"Error handled: {API.KnownReqErrs(err.code)}")
handled = True
else: