From: sbkelley Date: Tue, 2 Dec 2025 17:58:09 +0000 (-0500) Subject: reorganize obs module X-Git-Url: https://skyeroc.xyz/gitweb/?a=commitdiff_plain;h=55271fac20adbd53ca9ba847acc2ba6f57718c1e;p=obs-ctl reorganize obs module --- diff --git a/src/obs_ctl/interface/obs.py b/src/obs_ctl/interface/obs.py deleted file mode 100644 index 870a504..0000000 --- a/src/obs_ctl/interface/obs.py +++ /dev/null @@ -1,355 +0,0 @@ -from dataclasses import dataclass -from enum import IntEnum, StrEnum -from functools import partial -import functools -from pathlib import Path -from time import ctime -import time - -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""" -TRY_AGAIN = 3 -"""Time to wait before trying again if OBS WebSocket isn't ready for requests.""" -logger = getLogger(__name__) - -@dataclass(frozen=True) -class OBSStats: - """A dataclass for transmitting OBS stats.""" - obsws_connection_active: bool = False - record_output_active: bool = False - record_output_paused: bool = False - record_output_bytes: float = 0.0 - """Total bytes to disk.""" - record_output_duration: float = 0.0 - stream_output_active: bool = False - stream_output_skipped_frames: float = 0.0 - stream_output_total_frames: float = 0.0 - stats_render_skipped_frames: float = 0.0 - stats_render_total_frames: float = 0.0 - stats_output_skipped_frames: float = 0.0 - stats_output_total_frames: float = 0.0 - cpu_usage_percent: float = 0.0 - memory_usage_mb: float = 0.0 - active_fps: float = 0.0 - available_disk_space: float = 0.0 - 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, Report""" - event_client: obs.EventClient = None - request_client: obs.ReqClient = None - is_connected: reactive[bool] = False - _trying_request: bool = False - _timer: Timer = None - - interval: reactive[float] = reactive(1.0) - ws_host: reactive[str] = reactive("localhost") - ws_port: reactive[int] = reactive(4455) - ws_password: reactive[str] = reactive("") - ws_timeout: (reactive[float] | reactive[int] | reactive[None]) = reactive(None) - - # 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 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 - - # MESSAGES - 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): - """Posted when OBS WebSocket reports a change to the Stream state. - - state should be something from API.OutputStates.""" - - class RecordStateUpdated(APIOutputMSGBase): - """Posted when OBS WebSocket reports a change to the Record state. - - state should be something from API.OutputStates.""" - pass - - class Report(MSGBase): - """Posted ever self.interval second. data is an OBSStats object.""" - 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, - ws_host: str = "localhost", - ws_port: int = 4455, - ws_password: str = "", - ws_timeout: int | float | None = None, - *, - content = "", - expand = False, - shrink = False, - markup = True, - name = None, - id = None, - classes = None, - disabled = False - ): - logger.debug(f"Initializing obs.API widget: {self}") - super().__init__(content, expand=expand, shrink=shrink, markup=markup, name=name, id=id, classes=classes, disabled=disabled) - self.set_reactive(API.interval, interval) - self.set_reactive(API.ws_host, ws_host) - self.set_reactive(API.ws_port, ws_port) - self.set_reactive(API.ws_password, ws_password) - self.set_reactive(API.ws_timeout, ws_timeout) - self.connect(host=ws_host, port=ws_port, password=ws_password, timeout=ws_timeout) - - - def on_mount(self): - self._reset_timer(self.interval) - - - 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 watch_interval(self, value): - self._reset_timer(value) - - - def stop_stream(self): - """Requests to stop streaming.""" - self._try_request(self.request_client.stop_stream) - - - def start_stream(self): - """Requests to start streaming.""" - self._try_request(self.request_client.start_stream) - - - def start_record(self): - """Requests to start recording.""" - self._try_request(self.request_client.start_record) - - - def stop_record(self): - """Requests to stop recording.""" - self._try_request(self.request_client.stop_record) - - - def start_pause(self): - """Requests to pause recording.""" - self._try_request(self.request_client.pause_record) - - - def stop_pause(self): - """Requests to resume recording.""" - self._try_request(self.request_client.resume_record) - - - # LOGGER - def set_record_path(self, path: str): - """Sets OBS's record path to path, if the path exists.""" - 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: - if self.is_connected: - try: - record_status = self._try_request(self.request_client.get_record_status) - stream_status = self._try_request(self.request_client.get_stream_status) - obs_stats = self._try_request(self.request_client.get_stats) - record_directory = self._try_request(self.request_client.get_record_directory) - except AttributeError: - return OBSStats() - else: - return OBSStats() - - output = OBSStats( - obsws_connection_active = True, # being here implies that it's true. - 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 - - - # LOGGER - def connect(self, host=None, port=None, password=None, timeout=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 - try: - logger.info(f"Connecting to OBS WebSocket at {host}:{port}") - self.request_client = obs.ReqClient( - host=host, - port=port, - password=password, - timeout=timeout - ) - self.event_client = obs.EventClient( - host=host, - port=port, - password=password, - timeout=timeout - ) - except ConnectionRefusedError as err: - logger.debug(err) - else: - self.is_connected = True - self.event_client.callback.register(self.on_stream_state_changed) - self.event_client.callback.register(self.on_record_state_changed) - - - def disconnect(self): - logger.info("Disconnecting from OBS WebSocket") - self.request_client = None - self.event_client = None - self.is_connected = False - - - # Regarding the pattern in _try_request -- - # number = 5 - # while number > 0 - # try: - # raise Error - # except Error: - # continue - # finally: - # number -= 1 - # - # Tested; finally is reached. No infinite while loop. - # LOGGER - 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}") - - output = None - _sentry = 3 - handled: bool = False - while not handled and _sentry >= 0: - try: - output = request() - handled = True - except obs_error.OBSSDKRequestError as err: - if err.code == API.KnownReqErr.NOT_READY: - logger.info(f"OBS WebSocket not ready; trying again in 3 seconds.") - logger.debug(f"At most {_sentry} more tries.") - - time.sleep(TRY_AGAIN) - continue - elif err.code in API.KnownReqErr: - logger.debug(f"Error handled: {API.KnownReqErr(err.code)}") - - handled = True - else: - raise - except Exception as err: - logger.error(err) - - self.disconnect() - handled = True - finally: - _sentry -= 1 - if _sentry < 0: - logger.info(f"Maximum tries attempted ({TRY_AGAIN}). Aborting request {name}.") - return output - - - # untested - def _reset_timer(self, interval): - self._timer: Timer = self.set_interval(interval, self._send_report) - - - def _send_report(self): - data = self.get_obs_stats() - self.post_message(API.Report(self, data)) \ No newline at end of file diff --git a/src/obs_ctl/interface/obs/__init__.py b/src/obs_ctl/interface/obs/__init__.py new file mode 100644 index 0000000..54af907 --- /dev/null +++ b/src/obs_ctl/interface/obs/__init__.py @@ -0,0 +1 @@ +from obs import API, OBSStats \ No newline at end of file diff --git a/src/obs_ctl/interface/obs/obs.py b/src/obs_ctl/interface/obs/obs.py new file mode 100644 index 0000000..01f7f68 --- /dev/null +++ b/src/obs_ctl/interface/obs/obs.py @@ -0,0 +1,355 @@ +from dataclasses import dataclass +from enum import IntEnum, StrEnum +from functools import partial +import functools +from pathlib import Path +from time import ctime +import time + +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""" +TRY_AGAIN = 3 +"""Time to wait before trying again if OBS WebSocket isn't ready for requests.""" +logger = getLogger(__name__) + +@dataclass(frozen=True) +class OBSStats: + """A dataclass for transmitting OBS stats.""" + obsws_connection_active: bool = False + record_output_active: bool = False + record_output_paused: bool = False + record_output_bytes: float = 0.0 + """Total bytes to disk.""" + record_output_duration: float = 0.0 + stream_output_active: bool = False + stream_output_skipped_frames: float = 0.0 + stream_output_total_frames: float = 0.0 + stats_render_skipped_frames: float = 0.0 + stats_render_total_frames: float = 0.0 + stats_output_skipped_frames: float = 0.0 + stats_output_total_frames: float = 0.0 + cpu_usage_percent: float = 0.0 + memory_usage_mb: float = 0.0 + active_fps: float = 0.0 + available_disk_space: float = 0.0 + 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, Report""" + event_client: obs.EventClient = None + request_client: obs.ReqClient = None + is_connected: reactive[bool] = False + _trying_request: bool = False + _timer: Timer = None + + interval: reactive[float] = reactive(1.0) + ws_host: reactive[str] = reactive("localhost") + ws_port: reactive[int] = reactive(4455) + ws_password: reactive[str] = reactive("") + ws_timeout: (reactive[float] | reactive[int] | reactive[None]) = reactive(None) + + # 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 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 + + # MESSAGES + 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): + """Posted when OBS WebSocket reports a change to the Stream state. + + state should be something from API.OutputStates.""" + + class RecordStateUpdated(APIOutputMSGBase): + """Posted when OBS WebSocket reports a change to the Record state. + + state should be something from API.OutputStates.""" + pass + + class Report(MSGBase): + """Posted ever self.interval second. data is an OBSStats object.""" + 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, + ws_host: str = "localhost", + ws_port: int = 4455, + ws_password: str = "", + ws_timeout: int | float | None = None, + *, + content = "", + expand = False, + shrink = False, + markup = True, + name = None, + id = None, + classes = None, + disabled = False + ): + logger.debug(f"Initializing obs.API widget: {self}") + super().__init__(content, expand=expand, shrink=shrink, markup=markup, name=name, id=id, classes=classes, disabled=disabled) + self.set_reactive(API.interval, interval) + self.set_reactive(API.ws_host, ws_host) + self.set_reactive(API.ws_port, ws_port) + self.set_reactive(API.ws_password, ws_password) + self.set_reactive(API.ws_timeout, ws_timeout) + self.connect(host=ws_host, port=ws_port, password=ws_password, timeout=ws_timeout) + + + def on_mount(self): + self._reset_timer(self.interval) + + + 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 watch_interval(self, value): + self._reset_timer(value) + + + def stop_stream(self): + """Requests to stop streaming.""" + self._try_request(self.request_client.stop_stream) + + + def start_stream(self): + """Requests to start streaming.""" + self._try_request(self.request_client.start_stream) + + + def start_record(self): + """Requests to start recording.""" + self._try_request(self.request_client.start_record) + + + def stop_record(self): + """Requests to stop recording.""" + self._try_request(self.request_client.stop_record) + + + def start_pause(self): + """Requests to pause recording.""" + self._try_request(self.request_client.pause_record) + + + def stop_pause(self): + """Requests to resume recording.""" + self._try_request(self.request_client.resume_record) + + + # LOGGER + def set_record_path(self, path: str): + """Sets OBS's record path to path, if the path exists.""" + 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: + if self.is_connected: + try: + record_status = self._try_request(self.request_client.get_record_status) + stream_status = self._try_request(self.request_client.get_stream_status) + obs_stats = self._try_request(self.request_client.get_stats) + record_directory = self._try_request(self.request_client.get_record_directory) + except AttributeError: + return OBSStats() + else: + return OBSStats() + + output = OBSStats( + obsws_connection_active = True, # being here implies that it's true. + 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 + + + # LOGGER + def connect(self, host=None, port=None, password=None, timeout=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 + try: + logger.info(f"Connecting to OBS WebSocket at {host}:{port}") + self.request_client = obs.ReqClient( + host=host, + port=port, + password=password, + timeout=timeout + ) + self.event_client = obs.EventClient( + host=host, + port=port, + password=password, + timeout=timeout + ) + except ConnectionRefusedError as err: + logger.debug(err) + else: + self.is_connected = True + self.event_client.callback.register(self.on_stream_state_changed) + self.event_client.callback.register(self.on_record_state_changed) + + + def disconnect(self): + logger.info("Disconnecting from OBS WebSocket") + self.request_client = None + self.event_client = None + self.is_connected = False + + + # Regarding the pattern in _try_request -- + # number = 5 + # while number > 0 + # try: + # raise Error + # except Error: + # continue + # finally: + # number -= 1 + # + # Tested; finally is reached. No infinite while loop. + # LOGGER + 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}") + + output = None + _sentry = 3 + handled: bool = False + while not handled and _sentry >= 0: + try: + output = request() + handled = True + except obs_error.OBSSDKRequestError as err: + if err.code == API.KnownReqErr.NOT_READY: + logger.info(f"OBS WebSocket not ready; trying again in 3 seconds.") + logger.debug(f"At most {_sentry} more tries.") + + time.sleep(TRY_AGAIN) + continue + elif err.code in API.KnownReqErr: + logger.debug(f"Error handled: {API.KnownReqErr(err.code)}") + + handled = True + else: + raise + except Exception as err: + logger.error(err) + + self.disconnect() + handled = True + finally: + _sentry -= 1 + if _sentry < 0: + logger.info(f"Maximum tries attempted ({TRY_AGAIN}). Aborting request {name}.") + return output + + + # untested + def _reset_timer(self, interval): + self._timer: Timer = self.set_interval(interval, self._send_report) + + + def _send_report(self): + data = self.get_obs_stats() + self.post_message(API.Report(self, data)) \ No newline at end of file