From 3a7f0235192f416dd9ff1aeda604114dc950a498 Mon Sep 17 00:00:00 2001 From: sbkelley Date: Sun, 30 Nov 2025 17:04:49 -0500 Subject: [PATCH] Finish restructure -- easier to maintain now --- src/obs_ctl/__init__.py | 212 ++++------------- src/obs_ctl/interface/__init__.py | 57 +---- src/obs_ctl/interface/message.py | 8 +- src/obs_ctl/interface/molecule/__init__.py | 3 +- .../interface/molecule/record_controls.py | 8 +- .../interface/molecule/settings_pane.py | 55 +++++ src/obs_ctl/interface/obs.py | 215 ++++++++++++++++++ 7 files changed, 322 insertions(+), 236 deletions(-) create mode 100644 src/obs_ctl/interface/molecule/settings_pane.py create mode 100644 src/obs_ctl/interface/obs.py diff --git a/src/obs_ctl/__init__.py b/src/obs_ctl/__init__.py index a4326e9..82b20e5 100644 --- a/src/obs_ctl/__init__.py +++ b/src/obs_ctl/__init__.py @@ -2,8 +2,6 @@ from dataclasses import dataclass 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 @@ -11,71 +9,25 @@ from textual.widgets import Footer, Header, Label, Placeholder, TabbedContent, T 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") @@ -95,124 +47,47 @@ class OBSCtlApp(App): 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") @@ -222,9 +97,10 @@ class OBSCtlApp(App): 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}" @@ -236,18 +112,6 @@ class OBSCtlApp(App): - 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 diff --git a/src/obs_ctl/interface/__init__.py b/src/obs_ctl/interface/__init__.py index 2015550..9d6c3e8 100644 --- a/src/obs_ctl/interface/__init__.py +++ b/src/obs_ctl/interface/__init__.py @@ -1,56 +1 @@ -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 diff --git a/src/obs_ctl/interface/message.py b/src/obs_ctl/interface/message.py index 650e06b..583094d 100644 --- a/src/obs_ctl/interface/message.py +++ b/src/obs_ctl/interface/message.py @@ -1,8 +1,14 @@ 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 ace480d..8271c2f 100644 --- a/src/obs_ctl/interface/molecule/__init__.py +++ b/src/obs_ctl/interface/molecule/__init__.py @@ -1 +1,2 @@ -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 18e1450..24ed87b 100644 --- a/src/obs_ctl/interface/molecule/record_controls.py +++ b/src/obs_ctl/interface/molecule/record_controls.py @@ -2,7 +2,7 @@ from textual.containers import ScrollableContainer, Grid 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 @@ -16,9 +16,9 @@ class RecordControls(ScrollableContainer): } """ - 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 new file mode 100644 index 0000000..47202d0 --- /dev/null +++ b/src/obs_ctl/interface/molecule/settings_pane.py @@ -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 new file mode 100644 index 0000000..80793af --- /dev/null +++ b/src/obs_ctl/interface/obs.py @@ -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 -- 2.47.3