summary | shortlog | log | commit | commitdiff | tree
raw | patch | inline | side by side (parent: 99e180f)
raw | patch | inline | side by side (parent: 99e180f)
| author | sbkelley <sb24kelley@gmail.com> | |
| Sun, 7 Dec 2025 20:45:28 +0000 (15:45 -0500) | ||
| committer | sbkelley <sb24kelley@gmail.com> | |
| Sun, 7 Dec 2025 20:45:28 +0000 (15:45 -0500) |
diff --git a/src/obs_ctl/app.py b/src/obs_ctl/app.py
index 79819d9193905f2af8b6e45caf1fdcb75d984216..9fd3f5b3a2be5bee2e1d196ec5fa6641593975b1 100644 (file)
--- a/src/obs_ctl/app.py
+++ b/src/obs_ctl/app.py
+from collections import namedtuple
import importlib.resources
from logging import getLogger
import typing
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")
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
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 b3e5b56479d2bcc307df93b0cdbabd193e817b19..32c928f01a6462e7543e94a39bf25660d4b7a576 100644 (file)
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 0985f476848fa3f348b0206fe1a8793e1b82f756..18d5ca14db5af7fabb34479a79243932fc604c68 100644 (file)
"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): ...
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))
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
index c00cb174f134d992146a002162089d0be788fcdc..f18484058ad1f4cdab8c7809f4fddca2a8c7863e 100644 (file)
"""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
index b213f152e8e17f8e777f190442e54fa535645dd8..6c386604e20502981ffbb2986060adbe88a2edb2 100644 (file)
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.
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
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
# 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,
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:
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
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
# 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: