From: sbkelley Date: Sun, 7 Dec 2025 20:45:28 +0000 (-0500) Subject: record controls go yellow on certain conditions X-Git-Url: https://skyeroc.xyz/gitweb/?a=commitdiff_plain;h=3462df7fc46966f6b52b166332f5be8ef3189da2;p=obs-ctl record controls go yellow on certain conditions --- diff --git a/src/obs_ctl/app.py b/src/obs_ctl/app.py index 79819d9..9fd3f5b 100644 --- a/src/obs_ctl/app.py +++ b/src/obs_ctl/app.py @@ -1,3 +1,4 @@ +from collections import namedtuple import importlib.resources from logging import getLogger import typing @@ -45,6 +46,7 @@ class OBSCtlApp(App): # type: ignore obsws_pane.port = api.ws_port obsws_pane.password = api.ws_password obsws_pane.timeout = api.ws_timeout + # self.get_api().on_record_state_changed(namedtuple("test", ("output_state"))(output_state="OBS_WEBSOCKET_OUTPUT_STARTED")) @on(obs.API.Report, "#api") @@ -52,38 +54,102 @@ class OBSCtlApp(App): # type: ignore self.update_monitors(message.data) - @on(RecordControls.RecordStarted, "#record_controls") - def on_record_started(self, message: RecordControls.RecordStarted): + @on(obs.API.Connected, "#api") + def on_api_connected(self, message: obs.API.Connected) -> None: + stats = self.get_api().get_obs_stats() + record_controls = self.get_record_controls() + record_controls.obsws_connected = True + record_controls.recording = stats.record_output_active + record_controls.streaming = stats.stream_output_active + record_controls.paused = stats.record_output_paused + self.get_websocket_pane().contents_disabled = True + + + @on(obs.API.Disconnected, "#api") + def on_api_disconnected(self, message: obs.API.Disconnected) -> None: + record_controls = self.get_record_controls() + record_controls.obsws_connected = False + record_controls.recording = False + record_controls.streaming = False + record_controls.paused = False + record_controls.obsws_warn = False + record_controls.recording_warn = False + record_controls.streaming_warn = False + record_controls.paused_warn = False + self.get_websocket_pane().contents_disabled = False + + + @on(obs.API.RecordStateUpdated, "#api") + def on_record_state_updated(self, message: obs.API.RecordStateUpdated): + record_controls = self.get_record_controls() + settings_pane = self.get_settings_pane() + match message.state: + case obs.OutputStates.STARTING: + record_controls.recording_warn = settings_pane.contents_disabled = True + case obs.OutputStates.STARTED: + record_controls.recording_warn = False + record_controls.recording = True + case obs.OutputStates.STOPPING: + record_controls.recording_warn = True + if record_controls.paused: + record_controls.paused_warn = True + case obs.OutputStates.STOPPED: + record_controls.recording_warn = record_controls.paused_warn = False + record_controls.recording = record_controls.paused = settings_pane.contents_disabled = False + case obs.OutputStates.PAUSED: + record_controls.paused = True + case obs.OutputStates.RESUMED: + record_controls.paused = False + + + @on(obs.API.StreamStateUpdated, "#api") + def on_stream_state_updated(self, message: obs.API.StreamStateUpdated): + record_controls = self.get_record_controls() + match message.state: + case obs.OutputStates.STARTING: + record_controls.streaming_warn = True + case obs.OutputStates.STARTED: + record_controls.streaming_warn = False + record_controls.streaming = True + case obs.OutputStates.STOPPING: + record_controls.streaming_warn = True + case obs.OutputStates.STOPPED: + record_controls.streaming_warn = False + record_controls.streaming = False + + + @on(RecordControls.RecordStartPressed, "#record_controls") + def on_record_started(self, message: RecordControls.RecordStartPressed): self.get_api().start_record() - @on(RecordControls.RecordStopped, "#record_controls") - def on_record_stopped(self, message: RecordControls.RecordStopped): + @on(RecordControls.RecordStopPressed, "#record_controls") + def on_record_stopped(self, message: RecordControls.RecordStopPressed): self.get_api().stop_record() - @on(RecordControls.StreamStarted, "#record_controls") - def on_stream_started(self, message: RecordControls.StreamStarted): + @on(RecordControls.StreamStartPressed, "#record_controls") + def on_stream_started(self, message: RecordControls.StreamStartPressed): self.get_api().start_stream() - @on(RecordControls.StreamStopped, "#record_controls") - def on_stream_stopped(self, message: RecordControls.StreamStopped): + @on(RecordControls.StreamStopPressed, "#record_controls") + def on_stream_stopped(self, message: RecordControls.StreamStopPressed): self.get_api().stop_stream() - @on(RecordControls.RecordPaused, "#record_controls") - def on_record_paused(self, message: RecordControls.RecordPaused): - self.get_api().start_pause + @on(RecordControls.RecordPausePressed, "#record_controls") + def on_record_paused(self, message: RecordControls.RecordPausePressed): + self.get_api().start_pause() - @on(RecordControls.RecordUnpaused, "#record_controls") - def on_record_unpaused(self, message: RecordControls.RecordUnpaused): + @on(RecordControls.RecordUnpausePressed, "#record_controls") + def on_record_unpaused(self, message: RecordControls.RecordUnpausePressed): self.get_api().stop_pause() - @on(RecordControls.OBSWSConnected, "#record_controls") - def on_obs_ws_connected(self, message: RecordControls.OBSWSConnected | None): + @on(RecordControls.OBSWSConnectedPressed, "#record_controls") + def on_obs_ws_connected(self, message: RecordControls.OBSWSConnectedPressed | None): obs_websocket_pane: OBSWebSocketPane = self.get_websocket_pane() host_setting = obs_websocket_pane.host port_setting = obs_websocket_pane.port @@ -107,20 +173,8 @@ class OBSCtlApp(App): # type: ignore def update_monitors(self, obs_stats: obs.OBSStats): - record_controls: RecordControls = self.get_record_controls() - obs_settings: OBSSettingsPane = self.get_settings_pane() - stats_display: StatsDisplay = self.get_stats_display() - obs_websocket_pane: OBSWebSocketPane = self.get_websocket_pane() - - record_controls.obsws_connected = obs_stats.obsws_connection_active - record_controls.streaming = obs_stats.stream_output_active - record_controls.recording = \ - obs_websocket_pane.contents_disabled = \ - obs_stats.record_output_active - record_controls.paused = obs_stats.record_output_paused - obs_settings.record_path = obs_stats.record_directory - stats_display.update_text_readouts(obs_stats) - + self.get_settings_pane().record_path = obs_stats.record_directory + self.get_stats_display().update_text_readouts(obs_stats) def get_api(self) -> obs.API: return typing.cast(obs.API, self.query_exactly_one("#api")) diff --git a/src/obs_ctl/interface/element/bool_control.py b/src/obs_ctl/interface/element/bool_control.py index b3e5b56..32c928f 100644 --- a/src/obs_ctl/interface/element/bool_control.py +++ b/src/obs_ctl/interface/element/bool_control.py @@ -40,15 +40,17 @@ class BoolControl(containers.Container): bool_control--buttonbox (Center) """ DEFAULT_CSS = """ - .bool_control--light{ - min-width:16; - text-align: center; - } - .-warning { - # background:$warning; + BoolControl{ .bool_control--light{ - .toggle--button { - color:$text-warning; + min-width:16; + text-align: center; + } + &.-warning { + # background:$warning; + .bool_control--light{ + .toggle--button { + color:$text-warning !important; + } } } } diff --git a/src/obs_ctl/interface/molecule/record_controls.py b/src/obs_ctl/interface/molecule/record_controls.py index 0985f47..18d5ca1 100644 --- a/src/obs_ctl/interface/molecule/record_controls.py +++ b/src/obs_ctl/interface/molecule/record_controls.py @@ -63,17 +63,21 @@ class RecordControls(ScrollableContainer): "record_controls--pause_control", } recording: reactive[bool] = reactive(False) + recording_warn: reactive[bool] = reactive(False) paused: reactive[bool] = reactive(False) + paused_warn: reactive[bool] = reactive(False) streaming: reactive[bool] = reactive(False) + streaming_warn: reactive[bool] = reactive(False) obsws_connected: reactive[bool] = reactive(False) - - class RecordStarted(MSGBase): ... - class RecordStopped(MSGBase): ... - class RecordPaused(MSGBase): ... - class RecordUnpaused(MSGBase): ... - class StreamStarted(MSGBase): ... - class StreamStopped(MSGBase): ... - class OBSWSConnected(MSGBase): ... + obsws_warn: reactive[bool] = reactive(False) + + class RecordStartPressed(MSGBase): ... + class RecordStopPressed(MSGBase): ... + class RecordPausePressed(MSGBase): ... + class RecordUnpausePressed(MSGBase): ... + class StreamStartPressed(MSGBase): ... + class StreamStopPressed(MSGBase): ... + class OBSWSConnectedPressed(MSGBase): ... class OBSWSDisconnected(MSGBase): ... @@ -103,23 +107,23 @@ class RecordControls(ScrollableContainer): def on_bool_control_set_true(self, message: BoolControl.SetTrue): match message.control.id: case "record_control": - self.post_message(RecordControls.RecordStarted(self)) + self.post_message(RecordControls.RecordStartPressed(self)) case "pause_control": - self.post_message(RecordControls.RecordPaused(self)) + self.post_message(RecordControls.RecordPausePressed(self)) case "stream_control": - self.post_message(RecordControls.StreamStarted(self)) + self.post_message(RecordControls.StreamStartPressed(self)) case "obsws_control": - self.post_message(RecordControls.OBSWSConnected(self)) + self.post_message(RecordControls.OBSWSConnectedPressed(self)) def on_bool_control_set_false(self, message: BoolControl.SetFalse): match message.control.id: case "record_control": - self.post_message(RecordControls.RecordStopped(self)) + self.post_message(RecordControls.RecordStopPressed(self)) case "pause_control": - self.post_message(RecordControls.RecordUnpaused(self)) + self.post_message(RecordControls.RecordUnpausePressed(self)) case "stream_control": - self.post_message(RecordControls.StreamStopped(self)) + self.post_message(RecordControls.StreamStopPressed(self)) case "obsws_control": self.post_message(RecordControls.OBSWSDisconnected(self)) @@ -128,18 +132,36 @@ class RecordControls(ScrollableContainer): self._set_control_value("#record_control", value) + def watch_recording_warn(self, value: bool) -> None: + self._set_control_warning("#record_control", value) + + def watch_paused(self, value: bool) -> None: self._set_control_value("#pause_control", value) + def watch_paused_warn(self, value: bool) -> None: + self._set_control_warning("#pause_control", value) + + def watch_streaming(self, value: bool) -> None: self._set_control_value("#stream_control", value) + def watch_streaming_warn(self, value: bool) -> None: + self._set_control_warning("#stream_control", value) + + def watch_obsws_connected(self, value: bool) -> None: self._set_control_value("#obsws_control", value) + def watch_obsws_warn(self, value: bool) -> None: + self._set_control_warning("#obsws_control", value) + + def _set_control_value(self, control_id: str, value: bool) -> None: - control: BoolControl = typing.cast(BoolControl, self.query_exactly_one(control_id)) - control.value = value \ No newline at end of file + self.query_exactly_one(control_id).value = value # type: ignore + + def _set_control_warning(self, control_id: str, value: bool) -> None: + self.query_exactly_one(control_id).warning = value # type: ignore \ No newline at end of file diff --git a/src/obs_ctl/interface/obs/__init__.py b/src/obs_ctl/interface/obs/__init__.py index c00cb17..f184840 100644 --- a/src/obs_ctl/interface/obs/__init__.py +++ b/src/obs_ctl/interface/obs/__init__.py @@ -1,2 +1,2 @@ """Provides the OBS WebSocket API widget, 'API', and OBSStats dataclass""" -from .obs import API as API, OBSStats as OBSStats \ No newline at end of file +from .obs import API as API, OBSStats as OBSStats, OutputStates as OutputStates \ No newline at end of file diff --git a/src/obs_ctl/interface/obs/obs.py b/src/obs_ctl/interface/obs/obs.py index b213f15..6c38660 100644 --- a/src/obs_ctl/interface/obs/obs.py +++ b/src/obs_ctl/interface/obs/obs.py @@ -68,6 +68,26 @@ class OBSStats: return output + +class OutputStates(StrEnum): + """Named output states from OBS WebSocket + + UKNOWN should never occur. Otherwise, they are relatively-self-explanatory. + + UNKNOWN, STARTED, STARTING, PAUSED, RESUMED, STOPPED, STOPPING, + RECONNECTING, RECONNECTED. + """ + UNKNOWN = "OBS_WEBSOCKET_OUTPUT_UNKNOWN" + STARTED = "OBS_WEBSOCKET_OUTPUT_STARTED" + STARTING = "OBS_WEBSOCKET_OUTPUT_STARTING" + PAUSED = "OBS_WEBSOCKET_OUTPUT_PAUSED" + RESUMED = "OBS_WEBSOCKET_OUTPUT_RESUMED" + STOPPED = "OBS_WEBSOCKET_OUTPUT_STOPPED" + STOPPING = "OBS_WEBSOCKET_OUTPUT_STOPPING" + RECONNECTING = "OBS_WEBSOCKET_OUTPUT_RECONNECTING" + RECONNECTED = "OBS_WEBSOCKET_OUTPUT_RECONNECTED" + + class API(Static): """The API is a widget for access to textual's message pump. @@ -129,6 +149,8 @@ class API(Static): event_client: obs.EventClient request_client: obs.ReqClient is_connected: reactive[bool] = reactive(False) + _last_stream_state: reactive["OutputStates"] = reactive(OutputStates.STOPPED) + _last_record_state: reactive["OutputStates"] = reactive(OutputStates.STOPPED) _trying_request: bool = False _timer: Timer | None = None @@ -139,24 +161,6 @@ class API(Static): ws_timeout: reactive[float] = reactive(3.0) # ENUMS - class OutputStates(StrEnum): - """Named output states from OBS WebSocket - - UKNOWN should never occur. Otherwise, they are relatively-self-explanatory. - - UNKNOWN, STARTED, STARTING, PAUSED, RESUMED, STOPPED, STOPPING, - RECONNECTING, RECONNECTED. - """ - UNKNOWN = "OBS_WEBSOCKET_OUTPUT_UNKNOWN" - STARTED = "OBS_WEBSOCKET_OUTPUT_STARTED" - STARTING = "OBS_WEBSOCKET_OUTPUT_STARTING" - PAUSED = "OBS_WEBSOCKET_OUTPUT_PAUSED" - RESUMED = "OBS_WEBSOCKET_OUTPUT_RESUMED" - STOPPED = "OBS_WEBSOCKET_OUTPUT_STOPPED" - STOPPING = "OBS_WEBSOCKET_OUTPUT_STOPPING" - RECONNECTING = "OBS_WEBSOCKET_OUTPUT_RECONNECTING" - RECONNECTED = "OBS_WEBSOCKET_OUTPUT_RECONNECTED" - class KnownReqErrs(IntEnum): """Known obs websocket request statuses/error codes""" NOT_READY = 207 @@ -171,29 +175,32 @@ class API(Static): # MESSAGES class APIOutputMSGBase(MSGBase): - state = None def __init__(self, control: "API", data: Any) -> None: super().__init__(control, _log_me=True) - _state = API.OutputStates(data.output_state) - logger.debug(f"{self} is an obs.API Output Message; state: {_state.name}") - self.state = _state + self.state = data + logger.debug(f"{self} is an obs.API Output Message; state: {self.state.name}") class StreamStateUpdated(APIOutputMSGBase): """Posted when OBS WebSocket reports a change to the Stream state. - state should be something from API.OutputStates.""" + state should be something from OutputStates.""" class RecordStateUpdated(APIOutputMSGBase): """Posted when OBS WebSocket reports a change to the Record state. - state should be something from API.OutputStates.""" + state should be something from OutputStates.""" class Report(MSGBase): """Posted ever self.interval second. data is an OBSStats object.""" def __init__(self: Self, control: "API", data: OBSStats): super().__init__(control) self.data = data - + + class Connected(MSGBase): + """Posted when a connection is made.""" + + class Disconnected(MSGBase): + """Posted when the connection is dropped""" def __init__( self: Self, @@ -248,6 +255,14 @@ class API(Static): self.disconnect() + def watch__last_stream_state(self, value): + # logger.debug(value) + self.post_message(API.StreamStateUpdated(self, value)) + + def watch__last_record_state(self, value): + # logger.debug(value) + self.post_message(API.RecordStateUpdated(self, value)) + # 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: @@ -271,6 +286,7 @@ class API(Static): self.is_connected = True self.event_client.callback.register(self.on_stream_state_changed) self.event_client.callback.register(self.on_record_state_changed) + self.post_message(API.Connected(self)) return True else: self.is_connected = False @@ -284,6 +300,7 @@ class API(Static): self.event_client.callback.deregister(self.on_stream_state_changed) self.event_client.callback.deregister(self.on_record_state_changed) self.is_connected = False + self.post_message(API.Disconnected(self)) return True @@ -371,11 +388,11 @@ class API(Static): # Callbacks for obsws API rather than Textual def on_stream_state_changed(self: Self, data: Any) -> None: - self.post_message(API.StreamStateUpdated(self, data)) + self._last_stream_state = OutputStates(data.output_state) def on_record_state_changed(self: Self, data: Any) -> None: - self.post_message(API.RecordStateUpdated(self, data)) + self._last_record_state = OutputStates(data.output_state) def _make_clients(self, host: str, port: int, password: str, timeout: float) -> bool: