summary | shortlog | log | commit | commitdiff | tree
raw | patch | inline | side by side (parent: 9253d4c)
raw | patch | inline | side by side (parent: 9253d4c)
| author | sbkelley <sb24kelley@gmail.com> | |
| Sun, 30 Nov 2025 22:04:49 +0000 (17:04 -0500) | ||
| committer | sbkelley <sb24kelley@gmail.com> | |
| Sun, 30 Nov 2025 22:04:49 +0000 (17:04 -0500) |
| src/obs_ctl/__init__.py | patch | blob | history | |
| src/obs_ctl/interface/__init__.py | patch | blob | history | |
| src/obs_ctl/interface/message.py | patch | blob | history | |
| src/obs_ctl/interface/molecule/__init__.py | patch | blob | history | |
| src/obs_ctl/interface/molecule/record_controls.py | patch | blob | history | |
| src/obs_ctl/interface/molecule/settings_pane.py | [new file with mode: 0644] | patch | blob |
| src/obs_ctl/interface/obs.py | [new file with mode: 0644] | patch | blob |
index a4326e96a6bb5b4f8eccfd66bc496454e1b82aea..82b20e57ec717ba67fcf6e02b57855cdc5ec82de 100644 (file)
--- a/src/obs_ctl/__init__.py
+++ b/src/obs_ctl/__init__.py
import importlib.resources
from pathlib import Path
-import obsws_python as obs
-import obsws_python.error as obs_error
from textual import on
from textual.app import App
from textual.containers import Vertical, Horizontal, Grid, ScrollableContainer, Container
from textual.css.query import NoMatches
from . import log, data
-from .interface import OBSSettingsPane
-
-from .interface.element.bool_control import BoolControl
from .interface.element.text_readout import TextReadout
-from .interface import RecordControls
+from .interface import RecordControls, OBSSettingsPane, obs
logger = log.getLogger(__name__)
-obs_req_client = obs.ReqClient()
-obs_evnt_client = obs.EventClient()
-
-@dataclass(frozen=True)
-class OBSStats:
- record_output_active: bool
- record_output_paused: bool
- record_output_bytes: float
- record_output_duration: float
- stream_output_active: bool
- stream_output_skipped_frames: float
- stream_output_total_frames: float
- stats_render_skipped_frames: float
- stats_render_total_frames: float
- stats_output_skipped_frames: float
- stats_output_total_frames: float
- cpu_usage_percent: float
- memory_usage_mb: float
- active_fps: float
- available_disk_space: float
- record_directory: str
-
- @property
- def estimated_hours_remaining(self):
- try:
- output = (
- (
- self.available_disk_space*1024*1024) / \
- (
- self.record_output_bytes / \
- self.record_output_duration
- )
- ) / 1000 / 60 / 60
- except:
- output = "N/A"
-
- return output
+
class OBSCtlApp(App):
logger.debug(importlib.resources.files(data))
CSS_PATH = importlib.resources.files(data).joinpath("layout.tcss")
- def on_mount(self):
- self.set_interval(1., self.process_obs_stats)
-
-
def compose(self):
yield Header()
yield Footer()
+ yield obs.API(1, id="api")
logger.debug("Composing OBSCtlApp")
with ScrollableContainer(id="content_container"):
with Vertical(id="blocks_v_container", classes="blocks_containers"):
with Horizontal(id="blocks_h_container1", classes="blocks_h_containers"):
yield RecordControls(id="record_controls")
- # with Grid(id="controls_container", classes="blocks gridblocks"):
- # yield BoolControl(False, label="Recording", id="record_control", classes="controls")
- # yield BoolControl(False, label="Paused", true_label="Pause", false_label="Unpause", swap_red=True, id="pause_control", classes="controls")
- # yield BoolControl(False, label="Streaming", id="stream_control", classes="controls")
with TabbedContent(id="settings_container", classes="blocks"):
with TabPane("OBS Settings", id="settings-tab"):
yield OBSSettingsPane(id="settings-content")
yield TextReadout("Encdng frames dropped", "-/- (-.-%)", id="encoding_lag", classes="stats")
- @on(RecordControls.StreamStopped, "#record_controls",)
- def stream_off(self, message):
- try:
- obs_req_client.stop_stream()
- except obs_error.OBSSDKRequestError as err:
- if err.code == 501:
- logger.debug("Stream already stopped.")
- else:
- raise
-
-
- @on(RecordControls.StreamStarted, "#record_controls")
- def stream_on(self, message):
- try:
- obs_req_client.start_stream()
- except obs_error.OBSSDKRequestError as err:
- if err.code == 500:
- logger.debug("Stream already started.")
- else:
- raise
-
-
+ @on(obs.API.Report, "#api")
+ def on_api_report(self, message: obs.API.Report):
+ self.update_monitors(message.data)
+
@on(RecordControls.RecordStarted, "#record_controls")
- def record_on(self, message):
- try:
- obs_req_client.start_record()
- except obs_error.OBSSDKRequestError as err:
- if err.code == 500:
- logger.debug("Record already started.")
- else:
- raise
-
+ def on_record_started(self, message: RecordControls.RecordStarted):
+ api: obs.API = self.query_exactly_one("#api")
+ api.start_record()
@on(RecordControls.RecordStopped, "#record_controls")
- def record_off(self, message):
- logger.debug("Record Stopped message rcv'd")
- try:
- obs_req_client.stop_record()
- except obs_error.OBSSDKRequestError as err:
- if err.code == 501:
- logger.debug("Record already stopped.")
- else:
- raise
+ def on_record_stopped(self, message: RecordControls.RecordStopped):
+ api: obs.API = self.query_exactly_one("#api")
+ api.stop_record()
+
+ @on(RecordControls.StreamStarted, "#record_controls")
+ def on_stream_started(self, message: RecordControls.StreamStarted):
+ api: obs.API = self.query_exactly_one("#api")
+ api.start_stream()
+ @on(RecordControls.StreamStopped, "#record_controls")
+ def on_stream_stopped(self, message: RecordControls.StreamStopped):
+ api: obs.API = self.query_exactly_one("#api")
+ api.stop_stream()
@on(RecordControls.RecordPaused, "#record_controls")
- def pause_on(self, message):
- logger.debug("Record Paused message rcv'd")
- try:
- obs_req_client.pause_record()
- except obs_error.OBSSDKRequestError as err:
- if err.code == 502:
- logger.debug("Record already paused.")
- else:
- raise
-
-
- @on(RecordControls.RecordUnpaused, "#pause_control")
- def pause_off(self, message):
- logger.debug("Record Unpaused message rcv'd")
- try:
- obs_req_client.resume_record()
- except obs_error.OBSSDKRequestError as err:
- if err.code == 503:
- logger.debug("Record already unpaused.")
- else:
- raise
+ def on_record_paused(self, message: RecordControls.RecordPaused):
+ api: obs.API = self.query_exactly_one("#api")
+ api.start_pause()
+ @on(RecordControls.RecordUnpaused, "#record_controls")
+ def on_record_unpaused(self, message: RecordControls.RecordUnpaused):
+ api: obs.API = self.query_exactly_one("#api")
+ api.stop_pause()
@on(OBSSettingsPane.RecordPathChanged, "#settings-content")
- def set_record_path(self, message: OBSSettingsPane.RecordPathChanged):
- logger.debug(f"Set record path rcv'd: {message.record_path}")
- if Path(message.record_path).exists():
- try:
- obs_req_client.set_record_directory(message.record_path)
- except obs_error.OBSSDKRequestError as err:
- if err.code == 500:
- logger.debug("Recording is running.")
- else:
- raise
- else:
- logger.debug("The path doesn't exist...")
-
- @staticmethod
- def get_obs_stats():
- record_status = obs_req_client.get_record_status()
- stream_status = obs_req_client.get_stream_status()
- obs_stats = obs_req_client.get_stats()
- record_directory = obs_req_client.get_record_directory()
- # logger.debug(record_status)
- # logger.debug(stream_status)
- # logger.debug(obs_stats.attrs())
- output = OBSStats(
- record_output_active=record_status.output_active,
- record_output_paused=record_status.output_paused,
- record_output_bytes=record_status.output_bytes,
- record_output_duration=record_status.output_duration,
- stream_output_active=stream_status.output_active,
- stream_output_skipped_frames=stream_status.output_skipped_frames,
- stream_output_total_frames=stream_status.output_total_frames,
- stats_render_skipped_frames=obs_stats.render_skipped_frames,
- stats_render_total_frames=obs_stats.render_total_frames,
- stats_output_skipped_frames=obs_stats.output_skipped_frames,
- stats_output_total_frames=obs_stats.output_total_frames,
- cpu_usage_percent=obs_stats.cpu_usage,
- memory_usage_mb=obs_stats.memory_usage,
- active_fps=obs_stats.active_fps,
- available_disk_space=obs_stats.available_disk_space,
- record_directory=record_directory.record_directory,
- )
-
- return output
-
+ def on_settings_record_path_changed(self, message: OBSSettingsPane.RecordPathChanged):
+ api: obs.API = self.query_exactly_one("#api")
+ api.set_record_path(message.record_path)
- def update_monitors(self, obs_stats: OBSStats):
- stream_ctl: BoolControl = self.query_exactly_one("#stream_control")
- record_ctl: BoolControl = self.query_exactly_one("#record_control")
- pause_ctl: BoolControl = self.query_exactly_one("#pause_control")
+ def update_monitors(self, obs_stats: obs.OBSStats):
+ record_controls: RecordControls = self.query_exactly_one("#record_controls")
cpu_usage: TextReadout = self.query_exactly_one("#cpu_usage")
memory_usage: TextReadout = self.query_exactly_one("#memory_usage")
active_fp: TextReadout = self.query_exactly_one("#fps")
encoding_lag: TextReadout = self.query_exactly_one("#encoding_lag")
obs_settings: OBSSettingsPane = self.query_exactly_one("#settings-content")
- stream_ctl.status = obs_stats.stream_output_active
- record_ctl.status = obs_stats.record_output_active
- pause_ctl.status = obs_stats.record_output_paused
+ record_controls.streaming = obs_stats.stream_output_active
+ record_controls.recording = obs_stats.record_output_active
+ record_controls.paused = obs_stats.record_output_paused
+
cpu_usage.text = f"{obs_stats.cpu_usage_percent:0.5}%"
memory_usage.text = f"{obs_stats.memory_usage_mb:0.5}MB"
active_fp.text = f"{obs_stats.active_fps}"
- def process_obs_stats(self):
- logger.debug("Processing OBS stats")
- stats = self.get_obs_stats()
- self.update_monitors(stats)
-
-
-
-
def main():
app = OBSCtlApp()
- # app = OBSLayoutTest()
- logger.debug("App entrypoint reached")
- logger.debug(f"Request Client: {obs_req_client}")
- logger.debug(f"Event Client: {obs_evnt_client}")
app.run()
\ No newline at end of file
index 2015550e60fc1fcd6504fe1d930523a3ffaacc8c..9d6c3e87f9a6bb0a85dd0a1419bc8d436fbb2a0e 100644 (file)
-from textual.app import App, ComposeResult
-from textual.containers import Horizontal, VerticalScroll, Grid, Container
-from textual.message import Message
-from textual.reactive import reactive
-from textual.widgets import Label, RadioButton, Button, Input
-from textual.widget import Widget
-from textual.css import query
-
-from .message import MSGBase
-from .molecule import RecordControls
-from .. import log
-
-
-
-logger = log.getLogger(__name__)
-
-class OBSSettingsPane(VerticalScroll):
- DEFAULT_CSS = """
- .obssettingspane--input {
- width:1fr;
- }
- Grid {
- grid-size:2 1;
- grid-columns:21 1fr;
- Label {
- width:1fr;
- text-align:right;
- padding-right:1;
- }
- }
- """
- COMPONENT_CLASSES = {
- "obssettingspane--input",
- }
- record_path: reactive[str] = reactive("")
- class RecordPathChanged(Message):
- control: "OBSSettingsPane" = None
- record_path: str = None
- def __init__(self, ctrl, path):
- super().__init__()
- self.control = ctrl
- self.record_path = path
-
- def compose(self):
- with Grid():
- yield Label("Record path:")
- yield Input("path", id="input_record_path", classes="obssettingspane--input", compact=True)
-
- def on_input_submitted(self, event:Input.Submitted):
- self.set_reactive(OBSSettingsPane.record_path, event.value)
- if event.control.id == "input_record_path":
- self.post_message(OBSSettingsPane.RecordPathChanged(self, event.value))
-
- def watch_record_path(self, value):
- input: Input = self.query_exactly_one("#input_record_path")
- input.value = value
+from .molecule import RecordControls, OBSSettingsPane
\ No newline at end of file
index 650e06baac4cfaaec2297e2c70309b61869a7974..583094d6fe7cfbdd681ee487d6f8d23ec9ac06f4 100644 (file)
from textual.message import Message
from textual.widget import Widget
+from ..log import getLogger
+
+logger = getLogger(__name__)
+
class MSGBase(Message):
control: Widget = None
- def __init__(self, control):
+ def __init__(self, control, *, _log_me=True):
+ if _log_me:
+ logger.debug(f"Message instantiated by {control}: {self}")
self.control = control
super().__init__()
\ No newline at end of file
diff --git a/src/obs_ctl/interface/molecule/__init__.py b/src/obs_ctl/interface/molecule/__init__.py
index ace480d67351851da6aa71286d9c1024047dfe80..8271c2fee028eccf9ca2a6a621b89649d85acb45 100644 (file)
-from .record_controls import RecordControls
\ No newline at end of file
+from .record_controls import RecordControls
+from .settings_pane import OBSSettingsPane
\ No newline at end of file
diff --git a/src/obs_ctl/interface/molecule/record_controls.py b/src/obs_ctl/interface/molecule/record_controls.py
index 18e1450c5ae398666afc25048213b5c014a46714..24ed87b825554a0984c73009c60598cbf1007eee 100644 (file)
from textual.message import Message
from textual.reactive import reactive
-from .. import MSGBase
+from ..message import MSGBase
from ..element import BoolControl
from ...log import getLogger
}
"""
- recording: reactive[bool] = False
- paused: reactive[bool] = False
- streaming: reactive[bool] = False
+ recording: reactive[bool] = reactive(False)
+ paused: reactive[bool] = reactive(False)
+ streaming: reactive[bool] = reactive(False)
class RecordStarted(MSGBase): ...
class RecordStopped(MSGBase): ...
diff --git a/src/obs_ctl/interface/molecule/settings_pane.py b/src/obs_ctl/interface/molecule/settings_pane.py
--- /dev/null
@@ -0,0 +1,55 @@
+from textual.app import App, ComposeResult
+from textual.containers import Horizontal, VerticalScroll, Grid, Container
+from textual.message import Message
+from textual.reactive import reactive
+from textual.widgets import Label, RadioButton, Button, Input
+from textual.widget import Widget
+from textual.css import query
+
+from . import RecordControls
+from ..message import MSGBase
+from ... import log
+
+logger = log.getLogger(__name__)
+
+class OBSSettingsPane(VerticalScroll):
+ DEFAULT_CSS = """
+ .obssettingspane--input {
+ width:1fr;
+ }
+ Grid {
+ grid-size:2 1;
+ grid-columns:21 1fr;
+ Label {
+ width:1fr;
+ text-align:right;
+ padding-right:1;
+ }
+ }
+ """
+ COMPONENT_CLASSES = {
+ "obssettingspane--input",
+ }
+ record_path: reactive[str] = reactive("")
+
+ class RecordPathChanged(Message):
+ control: "OBSSettingsPane" = None
+ record_path: str = None
+ def __init__(self, ctrl, path):
+ super().__init__()
+ self.control = ctrl
+ self.record_path = path
+
+ def compose(self):
+ with Grid():
+ yield Label("Record path:")
+ yield Input("path", id="input_record_path", classes="obssettingspane--input", compact=True)
+
+ def on_input_submitted(self, event:Input.Submitted):
+ self.set_reactive(OBSSettingsPane.record_path, event.value)
+ if event.control.id == "input_record_path":
+ self.post_message(OBSSettingsPane.RecordPathChanged(self, event.value))
+
+ def watch_record_path(self, value):
+ input: Input = self.query_exactly_one("#input_record_path")
+ input.value = value
diff --git a/src/obs_ctl/interface/obs.py b/src/obs_ctl/interface/obs.py
--- /dev/null
@@ -0,0 +1,215 @@
+from dataclasses import dataclass
+from enum import IntEnum, StrEnum
+from functools import partial
+import functools
+from pathlib import Path
+from time import ctime
+
+import obsws_python as obs
+import obsws_python.error as obs_error
+from textual.timer import Timer
+from textual.reactive import reactive
+from textual.widgets import Static
+
+from ..log import getLogger
+from .message import MSGBase
+
+MS_TO_HOURS = 1 / 1000 / 60 / 60
+"""Multiply milliseconds by this reciprocal to get hours out."""
+MB_TO_BYTES = 1_048_576
+"""Multiply megabytes by this number to get bytes. MB, not MiB"""
+
+logger = getLogger(__name__)
+
+@dataclass(frozen=True)
+class OBSStats:
+ """A dataclass for transmitting OBS stats."""
+ record_output_active: bool
+ record_output_paused: bool
+ record_output_bytes: float
+ """Total bytes to disk."""
+ record_output_duration: float
+ stream_output_active: bool
+ stream_output_skipped_frames: float
+ stream_output_total_frames: float
+ stats_render_skipped_frames: float
+ stats_render_total_frames: float
+ stats_output_skipped_frames: float
+ stats_output_total_frames: float
+ cpu_usage_percent: float
+ memory_usage_mb: float
+ active_fps: float
+ available_disk_space: float
+ record_directory: str
+
+ @property
+ def estimated_hours_remaining(self):
+ """The number of hours left on the disk, at current bitrate etc.
+
+ Gets more accurate over time.
+ """
+ try:
+ output = (
+ (self.available_disk_space * MB_TO_BYTES) /
+ (self.record_output_bytes / self.record_output_duration)
+ ) * MS_TO_HOURS
+ except:
+ output = "N/A"
+
+ return output
+
+class API(Static):
+ """The API is a widget for access to textual's message pump.
+
+ It stores no states. It reads, writes, and outputs states.
+
+ Messages: StreamStateUpdated, RecordStateUpdated"""
+ event_client: obs.EventClient = None
+ request_client: obs.ReqClient = None
+ _timer: Timer = None
+
+ interval: reactive[float] = reactive(1.0)
+
+ class OutputStates(StrEnum):
+ 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"
+
+ class KnownReqErr(IntEnum):
+ """Known obs websocket request statuses/error codes"""
+ NOT_READY = 207
+ """This usually occurs during OBS scene collection change or exit.
+
+ Requests may be tried again after a delay if this code is given.
+ """
+ OUTPUT_RUNNING = 500
+ OUTPUT_NOT_RUNNING = 501
+ OUTPUT_PAUSED = 502
+ OUTPUT_NOT_PAUSED = 503
+
+ class APIOutputMSGBase(MSGBase):
+ state = None
+ def __init__(self, control, data):
+ super().__init__(control)
+ _state = API.OutputStates(data.output_state)
+ logger.debug(f"{self} is an obs.API Message; state: {_state.name}")
+ self.state = _state
+
+ class StreamStateUpdated(APIOutputMSGBase): ...
+
+ class RecordStateUpdated(APIOutputMSGBase): ...
+
+ class Report(MSGBase):
+ data: OBSStats = None
+ def __init__(self, control, data: OBSStats):
+ # this message is sent too often to log it.
+ super().__init__(control, _log_me = False)
+ self.data = data
+
+
+ def __init__(self, interval, content = "", *, expand = False, shrink = False, markup = True, name = None, id = None, classes = None, disabled = False):
+ super().__init__(content, expand=expand, shrink=shrink, markup=markup, name=name, id=id, classes=classes, disabled=disabled)
+ logger.debug(f"Initializing obs.API widget: {self}")
+ self.request_client = obs.ReqClient()
+ self.event_client = obs.EventClient()
+ on_stream_state_changed = self.on_stream_state_changed
+ on_record_state_changed = self.on_record_state_changed
+ self.event_client.callback.register(on_stream_state_changed)
+ self.event_client.callback.register(on_record_state_changed)
+ self.set_reactive(API.interval, interval)
+
+
+ def on_mount(self):
+ self._timer: Timer = self.set_interval(self.interval, self.send_report)
+
+
+ def _try_request(self, request:functools.partial):
+ if type(request) is functools.partial:
+ name = request.func.__name__
+ else:
+ name = request.__name__
+ logger.info(f"Sending request: {name}")
+ try:
+ request()
+ except obs_error.OBSSDKRequestError as err:
+ if err.code in API.KnownReqErr:
+ logger.debug(f"Error handled: {API.KnownReqErr(err.code)}")
+ else:
+ raise
+
+
+ def on_stream_state_changed(self: "API", data):
+ self.post_message(API.StreamStateUpdated(self, data))
+
+
+ def on_record_state_changed(self: "API", data):
+ self.post_message(API.RecordStateUpdated(self, data))
+
+
+ def send_report(self):
+ self.post_message(API.Report(self, self.get_obs_stats()))
+
+
+ def stop_stream(self):
+ self._try_request(self.request_client.stop_stream)
+
+
+ def start_stream(self):
+ self._try_request(self.request_client.start_stream)
+
+
+ def start_record(self):
+ self._try_request(self.request_client.start_record)
+
+
+ def stop_record(self):
+ self._try_request(self.request_client.stop_record)
+
+
+ def start_pause(self):
+ self._try_request(self.request_client.pause_record)
+
+
+ def stop_pause(self):
+ self._try_request(self.request_client.resume_record)
+
+
+ def set_record_path(self, path: str):
+ if Path(path).exists():
+ self._try_request(partial(self.request_client.set_record_directory, path))
+ else:
+ logger.debug("The path doesn't exist...")
+
+
+ def get_obs_stats(self) -> OBSStats:
+ record_status = self.request_client.get_record_status()
+ stream_status = self.request_client.get_stream_status()
+ obs_stats = self.request_client.get_stats()
+ record_directory = self.request_client.get_record_directory()
+
+ output = OBSStats(
+ record_output_active=record_status.output_active,
+ record_output_paused=record_status.output_paused,
+ record_output_bytes=record_status.output_bytes,
+ record_output_duration=record_status.output_duration,
+
+ stream_output_active=stream_status.output_active,
+ stream_output_skipped_frames=stream_status.output_skipped_frames,
+ stream_output_total_frames=stream_status.output_total_frames,
+
+ stats_render_skipped_frames=obs_stats.render_skipped_frames,
+ stats_render_total_frames=obs_stats.render_total_frames,
+ stats_output_skipped_frames=obs_stats.output_skipped_frames,
+ stats_output_total_frames=obs_stats.output_total_frames,
+ cpu_usage_percent=obs_stats.cpu_usage,
+ memory_usage_mb=obs_stats.memory_usage,
+ active_fps=obs_stats.active_fps,
+ available_disk_space=obs_stats.available_disk_space,
+
+ record_directory=record_directory.record_directory,
+ )
+
+ return output
\ No newline at end of file