From: sbkelley Date: Sun, 7 Dec 2025 00:12:47 +0000 (-0500) Subject: Extensive type-hinting X-Git-Url: https://skyeroc.xyz/gitweb/?a=commitdiff_plain;h=92acae8c611939eb4d9399c8fabcf9daf393f7d2;p=obs-ctl Extensive type-hinting --- diff --git a/src/obs_ctl/__init__.py b/src/obs_ctl/__init__.py index e01e7a6..5e4f061 100644 --- a/src/obs_ctl/__init__.py +++ b/src/obs_ctl/__init__.py @@ -1,5 +1,5 @@ -from .app import OBSCtlApp -from .widget_testbed import TestApp +from .app import OBSCtlApp as OBSCtlApp +from .widget_testbed import TestApp as TestApp def run(): app = OBSCtlApp() diff --git a/src/obs_ctl/app.py b/src/obs_ctl/app.py index af00def..d4f0b15 100644 --- a/src/obs_ctl/app.py +++ b/src/obs_ctl/app.py @@ -1,26 +1,28 @@ import importlib.resources from logging import getLogger +import typing -from textual import on +from textual import on, events +from textual._path import CSSPathType from textual.app import App from textual.containers import HorizontalGroup, VerticalGroup, ScrollableContainer -from textual.widgets import Footer, Header, Input +from textual.widgets import Footer, Header # need to import log to initalize logging -from . import data, log -from .interface import RecordControls, StatsDisplay, SettingsBlock, obs, OBSWebSocketPane, OBSCTLPane, OBSSettingsPane +from . import data, log # type: ignore +from .interface import RecordControls, StatsDisplay, SettingsBlock, obs, OBSWebSocketPane, OBSSettingsPane logger = getLogger(__name__) -class OBSCtlApp(App): +class OBSCtlApp(App): # type: ignore DEFAULT_CSS = """ """ - CSS_PATH = [ + CSS_PATH = typing.cast(CSSPathType, [ importlib.resources.files(data).joinpath("critical.tcss"), importlib.resources.files(data).joinpath("default_layout.tcss"), importlib.resources.files(data).joinpath("default_style.tcss"), - ] + ]) def compose(self): yield Header() @@ -36,87 +38,105 @@ class OBSCtlApp(App): yield StatsDisplay(id="stats_display", classes="section") - def on_mount(self, message): + def on_mount(self, message: events.Mount): self.on_obs_ws_connected(None) @on(obs.API.Report, "#api") def on_api_report(self, message: obs.API.Report): self.update_monitors(message.data) - - # @on(obs.API.Report, "#api")(on_api_report) @on(RecordControls.RecordStarted, "#record_controls") def on_record_started(self, message: RecordControls.RecordStarted): - self._query_api("start_record") + self.get_api().start_record() @on(RecordControls.RecordStopped, "#record_controls") def on_record_stopped(self, message: RecordControls.RecordStopped): - self._query_api("stop_record") + self.get_api().stop_record() @on(RecordControls.StreamStarted, "#record_controls") def on_stream_started(self, message: RecordControls.StreamStarted): - self._query_api("start_stream") + self.get_api().start_stream() @on(RecordControls.StreamStopped, "#record_controls") def on_stream_stopped(self, message: RecordControls.StreamStopped): - self._query_api("stop_stream") + self.get_api().stop_stream() @on(RecordControls.RecordPaused, "#record_controls") def on_record_paused(self, message: RecordControls.RecordPaused): - self._query_api("start_pause") + self.get_api().start_pause @on(RecordControls.RecordUnpaused, "#record_controls") def on_record_unpaused(self, message: RecordControls.RecordUnpaused): - self._query_api("stop_pause") + self.get_api().stop_pause() @on(RecordControls.OBSWSConnected, "#record_controls") - def on_obs_ws_connected(self, message: RecordControls.OBSWSConnected): - api: obs.API = self.query_exactly_one("#api") - obs_websocket_pane: OBSWebSocketPane = self.query_exactly_one("#obs_websocket_pane") + def on_obs_ws_connected(self, message: RecordControls.OBSWSConnected | None): + api: obs.API = self.get_api() + obs_websocket_pane: OBSWebSocketPane = self.get_websocket_pane() host_setting = obs_websocket_pane.host port_setting = obs_websocket_pane.port password_setting = obs_websocket_pane.password timeout_setting = obs_websocket_pane.timeout - self._query_api("connect", host=host_setting, port=port_setting, password=password_setting, timeout=timeout_setting) + self.get_api().connect( + host=host_setting, + port=port_setting, + password=password_setting, + timeout=timeout_setting + ) if api.is_connected: obs_websocket_pane.disable_contents() @on(RecordControls.OBSWSDisconnected, "#record_controls") def on_obs_ws_disconnected(self, message: RecordControls.OBSWSDisconnected): - api: obs.API = self.query_exactly_one("#api") - obs_websocket_pane: OBSWebSocketPane = self.query_exactly_one("#obs_websocket_pane") - self._query_api("disconnect") + api: obs.API = self.get_api() + obs_websocket_pane: OBSWebSocketPane = self.get_websocket_pane() + self.get_api().disconnect() if not api.is_connected: obs_websocket_pane.enable_contents() - - def _query_api(self, query: str, *args, **kwargs): - api: obs.API = self.query_exactly_one("#api") - return getattr(api, query)(*args, **kwargs) @on(OBSSettingsPane.RecordPathChanged, "#obs_settings_pane") - def on_settings_block_record_path_changed(self, message): - api: obs.API = self.query_exactly_one("#api") - api.set_record_path(message.value) + def on_settings_block_record_path_changed(self, message: OBSSettingsPane.RecordPathChanged): + self.get_api().set_record_path(message.value) def update_monitors(self, obs_stats: obs.OBSStats): - record_controls: RecordControls = self.query_exactly_one("#record_controls") - obs_settings: OBSSettingsPane = self.query_exactly_one("#obs_settings_pane") - stats_display: StatsDisplay = self.query_exactly_one("#stats_display") + record_controls: RecordControls = self.get_record_controls() + obs_settings: OBSSettingsPane = self.get_settings_pane() + stats_display: StatsDisplay = self.get_stats_display() record_controls.obsws_connected = obs_stats.obsws_connection_active record_controls.streaming = obs_stats.stream_output_active record_controls.recording = 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) \ No newline at end of file + stats_display.update_text_readouts(obs_stats) + + + def get_api(self) -> obs.API: + return typing.cast(obs.API, self.query_exactly_one("#api")) + + + def get_websocket_pane(self) -> OBSWebSocketPane: + return typing.cast(OBSWebSocketPane, self.query_exactly_one("#obs_websocket_pane")) + + + def get_record_controls(self) -> RecordControls: + return typing.cast(RecordControls, self.query_exactly_one("#record_controls")) + + + def get_settings_pane(self) -> OBSSettingsPane: + return typing.cast(OBSSettingsPane, self.query_exactly_one("#obs_settings_pane")) + + + def get_stats_display(self) -> StatsDisplay: + return typing.cast(StatsDisplay, self.query_exactly_one("#stats_display")) \ No newline at end of file diff --git a/src/obs_ctl/interface/__init__.py b/src/obs_ctl/interface/__init__.py index 161ded2..a7adfe0 100644 --- a/src/obs_ctl/interface/__init__.py +++ b/src/obs_ctl/interface/__init__.py @@ -1,3 +1,3 @@ """Provides RecordControls, OBSSettingsPane, StatsDisplay, and the obs module""" -from .molecule import RecordControls, SettingsBlock, StatsDisplay, OBSSettingsPane, OBSCTLPane, OBSWebSocketPane -from . import obs \ No newline at end of file +from .molecule import RecordControls as RecordControls, SettingsBlock as SettingsBlock, StatsDisplay as StatsDisplay, OBSSettingsPane as OBSSettingsPane, OBSCTLPane as OBSCTLPane, OBSWebSocketPane as OBSWebSocketPane +from . import obs as obs \ No newline at end of file diff --git a/src/obs_ctl/interface/element/__init__.py b/src/obs_ctl/interface/element/__init__.py index 9aa087f..33589d2 100644 --- a/src/obs_ctl/interface/element/__init__.py +++ b/src/obs_ctl/interface/element/__init__.py @@ -1,5 +1,5 @@ """Provides BoolControl and TextReadout Only uses builtins, textual, and maybe log and const.""" -from .bool_control import BoolControl -from .text_readout import TextReadout \ No newline at end of file +from .bool_control import BoolControl as BoolControl +from .text_readout import TextReadout as TextReadout \ No newline at end of file diff --git a/src/obs_ctl/interface/element/bool_control.py b/src/obs_ctl/interface/element/bool_control.py index 175958e..eedfe6d 100644 --- a/src/obs_ctl/interface/element/bool_control.py +++ b/src/obs_ctl/interface/element/bool_control.py @@ -1,7 +1,9 @@ +from typing import Any +import typing from textual import on from textual.app import ComposeResult from textual import containers -from textual.message import Message +from textual.events import Mount from textual.reactive import reactive from textual.widgets import Button, RadioButton @@ -79,14 +81,14 @@ class BoolControl(containers.Container): def __init__( self, value: bool, - *args, + *args: Any, label: str = "Boolean", true_label: str = "Start", false_label: str = "Stop", swap_red: bool = False, compact: bool = True, - layout = "vertical", - **kwargs,): + layout: str = "vertical", + **kwargs: Any,): """Initialize BoolControl # Parameters @@ -137,7 +139,7 @@ class BoolControl(containers.Container): ) - def _on_mount(self, event): + def _on_mount(self, event: Mount): # super()._on_mount(event) self.watch_value(self.value) self.watch_label(self.label) @@ -154,45 +156,54 @@ class BoolControl(containers.Container): @on(Button.Pressed, "#true") - def true_pressed(self): + def true_pressed(self, event: Button.Pressed): self.post_message(self.SetTrue(self)) def watch_value(self, value: bool) -> None: - light: RadioButton = self.query_exactly_one("#light") + light: RadioButton = self.get_light() light.value = value - def watch_label(self, value): - light: RadioButton = self.query_exactly_one("#light") + def watch_label(self, value: str): + light: RadioButton = self.get_light() light.label = value - def watch_true_label(self, value): - true: Button = self.query_exactly_one("#true") + def watch_true_label(self, value: str): + true: Button = self.get_true_button() true.label = value - def watch_false_label(self, value): - false: Button = self.query_exactly_one("#false") + def watch_false_label(self, value: str): + false: Button = self.get_false_button() false.label = value - def watch_swap_red(self, value): - true: Button = self.query_exactly_one("#true") - false: Button = self.query_exactly_one("#false") + def watch_swap_red(self, value: bool): + true: Button = self.get_true_button() + false: Button = self.get_false_button() false.variant = "success" if value else "error" true.variant = "error" if value else "success" - def watch_compact(self, value): - light: RadioButton = self.query_exactly_one("#light") - true: Button = self.query_exactly_one("#true") - false: Button = self.query_exactly_one("#false") + def watch_compact(self, value: bool): + light: RadioButton = self.get_light() + true: Button = self.get_true_button() + false: Button = self.get_false_button() light.compact = value true.compact = value false.compact = value + + def get_true_button(self): + return typing.cast(Button, self.query_exactly_one("#true")) + + def get_false_button(self): + return typing.cast(Button, self.query_exactly_one("#false")) + + def get_light(self) -> RadioButton: + return typing.cast(RadioButton, self.query_exactly_one("#light")) - def watch_control_layout(self, value): + def watch_control_layout(self, value: str): self.set_styles(layout=value) \ No newline at end of file diff --git a/src/obs_ctl/interface/element/text_readout.py b/src/obs_ctl/interface/element/text_readout.py index 1b60115..84c7dca 100644 --- a/src/obs_ctl/interface/element/text_readout.py +++ b/src/obs_ctl/interface/element/text_readout.py @@ -1,3 +1,5 @@ +from typing import Any +import typing from textual.containers import Container from textual.reactive import reactive from textual.widgets import Label @@ -25,7 +27,7 @@ class TextReadout(Container): label = reactive("Label") readout_layout = reactive("vertical") - def __init__(self, label, text, *args, layout = "vertical", **kwargs): + def __init__(self, label: str, text: str, *args: Any, layout: str = "vertical", **kwargs: Any) -> None: """Initialize TextReadout # Parameters @@ -46,14 +48,14 @@ class TextReadout(Container): yield Container(Label(self.text, id="text"), classes="text_readout--text text_readout--readout") def watch_text(self, value: str): - text: Label = self.query_exactly_one("#text") + text: Label = typing.cast(Label, self.query_exactly_one("#text")) text.update(value) - def watch_label(self, value): - label: Label = self.query_exactly_one("#label") + def watch_label(self, value: str): + label: Label = typing.cast(Label, self.query_exactly_one("#label")) label.update(value) - def watch_readout_layout(self, value): + def watch_readout_layout(self, value: str): self.set_styles(layout=value) \ No newline at end of file diff --git a/src/obs_ctl/interface/message.py b/src/obs_ctl/interface/message.py index e99cb64..5fb75a3 100644 --- a/src/obs_ctl/interface/message.py +++ b/src/obs_ctl/interface/message.py @@ -1,3 +1,4 @@ +from typing import Self from textual.message import Message from textual.widget import Widget @@ -6,9 +7,13 @@ from logging import getLogger logger = getLogger(__name__) class MSGBase(Message): - control: Widget = None - def __init__(self, control, *, _log_me=True): + # ctrl: Widget + def __init__(self: Self, control: Widget, *, _log_me: bool=True) -> None: if _log_me: logger.debug(f"Message instantiated by {control}: {self}") - self.control = control - super().__init__() \ No newline at end of file + self.ctrl = control + super().__init__() + + @property + def control(self: Self) -> Widget: + return self.ctrl \ 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 ba9a9d9..19babc5 100644 --- a/src/obs_ctl/interface/molecule/__init__.py +++ b/src/obs_ctl/interface/molecule/__init__.py @@ -2,7 +2,6 @@ Uses obs_ctl.interface.element """ -from .record_controls import RecordControls -from .settings_block import SettingsBlock -from .stats_display import StatsDisplay -from .settings_block import SettingsBlock, OBSWebSocketPane, OBSCTLPane, OBSSettingsPane \ No newline at end of file +from .record_controls import RecordControls as RecordControls +from .settings_block import OBSCTLPane as OBSCTLPane, OBSSettingsPane as OBSSettingsPane, OBSWebSocketPane as OBSWebSocketPane, SettingsBlock as SettingsBlock +from .stats_display import StatsDisplay as StatsDisplay diff --git a/src/obs_ctl/interface/molecule/record_controls.py b/src/obs_ctl/interface/molecule/record_controls.py index 4a1e96c..0288217 100644 --- a/src/obs_ctl/interface/molecule/record_controls.py +++ b/src/obs_ctl/interface/molecule/record_controls.py @@ -1,3 +1,4 @@ +import typing from textual.containers import ScrollableContainer, Grid from textual.reactive import reactive @@ -111,6 +112,8 @@ class RecordControls(ScrollableContainer): self.post_message(RecordControls.StreamStarted(self)) case "obsws_control": self.post_message(RecordControls.OBSWSConnected(self)) + case _: + logger.debug("Unrecognized bool control...") def on_bool_control_set_false(self, message: BoolControl.SetFalse): @@ -123,24 +126,26 @@ class RecordControls(ScrollableContainer): self.post_message(RecordControls.StreamStopped(self)) case "obsws_control": self.post_message(RecordControls.OBSWSDisconnected(self)) + case _: + logger.debug("Unrecognized bool control...") - def watch_recording(self, value): + def watch_recording(self, value: bool) -> None: self._set_control_value("#record_control", value) - def watch_paused(self, value): + def watch_paused(self, value: bool) -> None: self._set_control_value("#pause_control", value) - def watch_streaming(self, value): + def watch_streaming(self, value: bool) -> None: self._set_control_value("#stream_control", value) - def watch_obsws_connected(self, value): + def watch_obsws_connected(self, value: bool) -> None: self._set_control_value("#obsws_control", value) - def _set_control_value(self, control_id: str, value): - control: BoolControl = self.query_exactly_one(control_id) + 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 diff --git a/src/obs_ctl/interface/molecule/settings_block/__init__.py b/src/obs_ctl/interface/molecule/settings_block/__init__.py index b8bfc6a..c595d6f 100644 --- a/src/obs_ctl/interface/molecule/settings_block/__init__.py +++ b/src/obs_ctl/interface/molecule/settings_block/__init__.py @@ -1 +1,6 @@ -from .settings_block import SettingsBlock, OBSSettingsPane, OBSCTLPane, OBSWebSocketPane \ No newline at end of file +from .settings_block import ( + SettingsBlock as SettingsBlock, + OBSSettingsPane as OBSSettingsPane, + OBSCTLPane as OBSCTLPane, + OBSWebSocketPane as OBSWebSocketPane +) \ No newline at end of file diff --git a/src/obs_ctl/interface/molecule/settings_block/pane.py b/src/obs_ctl/interface/molecule/settings_block/pane.py index ff4a4d1..abf522b 100644 --- a/src/obs_ctl/interface/molecule/settings_block/pane.py +++ b/src/obs_ctl/interface/molecule/settings_block/pane.py @@ -6,26 +6,29 @@ from textual import on from textual.containers import ScrollableContainer, Grid from textual.reactive import reactive from textual.widgets import TabPane, Input, Label -from textual.widget import Widget from ...message import MSGBase +from textual.widget import Widget logger = getLogger(__name__) class SettingsMSGBase(MSGBase): - value = None - - def __init__(self, control, value, *, _log_me=True): + # value: Any + def __init__(self, control: Widget, value: Any, *, _log_me: bool=True): super().__init__(control, _log_me=_log_me) self.value = value class PaneBase(TabPane): def disable_contents(self): - container: ScrollableContainer = self.query_exactly_one(".--container") + container = self.query_exactly_one(".--container") container.disabled = True def enable_contents(self): - container: ScrollableContainer = self.query_exactly_one(".--container") + container = self.query_exactly_one(".--container") container.disabled = False +SettingsTypeWidget = partial[Input] +SettingsTypeType = type[int | float | str] +SettingsTypeMsg = type[Input.Submitted] +SettingsTypeExpectedType = type[str] class SettingsType(Enum): """(widget_class, setting type, listener message, widget-expected type)""" STRING = (partial(Input, compact=True), str, Input.Submitted, str) @@ -33,27 +36,26 @@ class SettingsType(Enum): INT = (partial(Input, compact=True), int, Input.Submitted, str) @property - def widget(self): + def widget(self) -> SettingsTypeWidget: """The widget this SettingsType uses""" return self.value[0] @property - def type(self): + def type(self) -> SettingsTypeType: """The type of the reactive this SettingsType interacts with""" return self.value[1] @property - def msg(self): + def msg(self) -> SettingsTypeMsg: """The message this SettingsType's widget posts when user interacts""" return self.value[2] @property - def expected_type(self): + def expected_type(self) -> SettingsTypeExpectedType: """The type this SettingsType's widget expects for its value""" return self.value[3] - -def make_pane_subclass(cls_name: str, settings_list: list, settings_type_list: list, defaults_list) -> PaneBase: +def make_pane_subclass(cls_name: str, settings_list: list[str], settings_type_list: list[SettingsType], defaults_list: list[Any]) -> type[PaneBase]: """Creates a new TabPane subclass with cls_name as its name. settings_list should be a list of names for settings. @@ -68,45 +70,41 @@ def make_pane_subclass(cls_name: str, settings_list: list, settings_type_list: l function, which emits a RecordPathChanged message containing a value when the user submits the input.""" row_height = 1 - loop_list = list(zip(settings_list, settings_type_list, defaults_list)) + loop_list: list[tuple[str, SettingsType, Any]] = list(zip(settings_list, settings_type_list, defaults_list)) DEFAULT_CSS = f"Grid {{grid-size:2;grid-rows:{row_height};min-height:{len(loop_list)*row_height}}}" def make_watch_reactive(watch_id: str): - def watch_func(self, value): + def watch_func(self: PaneBase, value: Any): input = self.query_exactly_one(watch_id) if type(input) is Input: input.value = str(value) return watch_func - def make_on_control_message(listener_message, listener_id, reactive, post_message): - def on_func(self, event): + def make_on_control_message(listener_message: SettingsTypeMsg, listener_id: str, reactive: reactive[Any], post_message: type[SettingsMSGBase]): + def on_func(self: PaneBase, event: SettingsTypeMsg): self.set_reactive(reactive, event.value) self.post_message(post_message(self, event.value)) return on(listener_message, listener_id)(on_func) def make_compose(): - def output_func(self): + def compose(self: PaneBase): with ScrollableContainer(classes="--container"): with Grid(): - for setting, control_and_type, setting_default in loop_list: + for setting, setting_type, setting_default in loop_list: ctrl_id = f"{setting_type.widget.func.__name__.lower()}_{setting}" yield Label(setting) - yield control_and_type.widget(value=control_and_type.expected_type(setting_default), id=ctrl_id) - return output_func + yield setting_type.widget(value=setting_type.expected_type(setting_default), id=ctrl_id) + return compose - new_dict = dict() + new_dict: dict[str, Any] = dict() for setting_name, setting_type, setting_default in loop_list: - setting_name: str = setting_name - setting_type: SettingsType = setting_type - setting_default: Any = setting_default control_setting_id: str = f"{setting_type.widget.func.__name__.lower()}_{setting_name}" SettingChanged_name: str = f"{snake_to_camel(setting_name)}Changed" watch_setting_name: str = f"watch_{setting_name}" on_setting_changed_name: str = f"on_{setting_name}_changed" - # For each desired setting (e.g. 'record_path'): # Set up a record_path reactive new_dict[setting_name] = reactive(setting_type.type(setting_default)) @@ -129,7 +127,7 @@ def make_pane_subclass(cls_name: str, settings_list: list, settings_type_list: l new_dict["compose"] = make_compose() # Construct the actual class and return it - new_pane = type(cls_name, (PaneBase,), new_dict) + new_pane: type[PaneBase] = type(cls_name, (PaneBase,), new_dict) return new_pane def snake_to_camel(string: str): diff --git a/src/obs_ctl/interface/molecule/settings_block/pane.pyi b/src/obs_ctl/interface/molecule/settings_block/pane.pyi deleted file mode 100644 index aad9354..0000000 --- a/src/obs_ctl/interface/molecule/settings_block/pane.pyi +++ /dev/null @@ -1,29 +0,0 @@ -from typing import Any, Self, TypeVar -from enum import Enum -from functools import partial -from textual.message import Message -from textual.widget import Widget -from textual.widgets import TabPane, Input -from ...message import MSGBase - -class SettingsMSGBase(MSGBase): - value: Any - def __init__(self: Self, control: Widget, value: any, _log_me: bool): ... - -class PaneBase(TabPane): - def disable_contents(self: Self) -> None: ... - def enable_contents(self: Self) -> None: ... - -class SettingsType(Enum): - STRING = None - FLOAT = None - INT = None - widget: partial[Widget] - type: type - msg: Message - expected_type: type - -class MakePaneSubclassResult(PaneBase): - DEFAULT_CSS: str - -def make_pane_subclass(cls_name: str, settings: list[str], settings_type_list: list[SettingsType], defaults_list: list[Any]) -> MakePaneSubclassResult: ... \ No newline at end of file diff --git a/src/obs_ctl/interface/molecule/settings_block/settings_block.py b/src/obs_ctl/interface/molecule/settings_block/settings_block.py index b0fb940..0cb9520 100644 --- a/src/obs_ctl/interface/molecule/settings_block/settings_block.py +++ b/src/obs_ctl/interface/molecule/settings_block/settings_block.py @@ -1,6 +1,5 @@ from logging import getLogger -from textual import on -from textual.widgets import TabbedContent, TabPane, Placeholder +from textual.widgets import TabbedContent from textual.containers import Container from . import pane diff --git a/src/obs_ctl/interface/molecule/settings_block/settings_block.pyi b/src/obs_ctl/interface/molecule/settings_block/settings_block.pyi index 80cbfe9..abd473b 100644 --- a/src/obs_ctl/interface/molecule/settings_block/settings_block.pyi +++ b/src/obs_ctl/interface/molecule/settings_block/settings_block.pyi @@ -1,34 +1,47 @@ -from typing import Self, TypeVar +from typing import ClassVar, Self +from . import pane as pane +from textual import on as on +from textual.app import ComposeResult from textual.containers import Container -from textual.widgets import TabPane +from textual.widgets import TabPane as TabPane from textual.reactive import reactive -from textual.message import Message -from .pane import PaneBase, SettingsMSGBase, SettingsType +from logging import Logger -class OBSSettingsPane(PaneBase): - record_path: reactive[str] - class RecordPathChanged(SettingsMSGBase): ... - def watch_record_path(self: Self, value: str): ... - def on_record_path_changed(self: Self, event: Message): ... +from .pane import PaneBase -class OBSWebSocketPane(PaneBase): - host: reactive[str] - port: reactive[int] - password: reactive[str] - timeout: reactive[float] - class HostChanged(SettingsMSGBase): ... - class PortChanged(SettingsMSGBase): ... - class PasswordChanged(SettingsMSGBase): ... - class TimeoutChanged(SettingsMSGBase): ... - def watch_host(self: Self, value: str): ... - def watch_port(self: Self, value: int): ... - def watch_password(self: Self, value: str): ... - def watch_timeout(self: Self, value: float): ... - def on_host_changed(self: Self, event: Message): ... - def on_port_changed(self: Self, event: Message): ... - def on_password_changed(self: Self, event: Message): ... - def on_timeout_changed(self: Self, event: Message): ... +logger: Logger -class OBSCTLPane(PaneBase): ... +class PaneBaseSubclass(PaneBase): + DEFAULT_CSS: str + def compose(self) -> ComposeResult: ... -class SettingsBlock(Container): ... \ No newline at end of file +class OBSSettingsPane(PaneBaseSubclass): + record_path: ClassVar[reactive[str]] + class RecordPathChanged(pane.SettingsMSGBase): ... + def watch_record_path(self: Self, value: str) -> None: ... + def on_record_path_changed(self: Self, event: pane.SettingsMSGBase) -> None: ... + def compose(self: Self) -> ComposeResult: ... + +class OBSWebSocketPane(PaneBaseSubclass): + host: ClassVar[reactive[str]] + port: ClassVar[reactive[int]] + password: ClassVar[reactive[str]] + timeout: ClassVar[reactive[float]] + class HostChanged(pane.SettingsMSGBase): ... + class PortChanged(pane.SettingsMSGBase): ... + class PasswordChanged(pane.SettingsMSGBase): ... + class TimeoutChanged(pane.SettingsMSGBase): ... + def watch_host(self: Self, value: str) -> None: ... + def watch_port(self: Self, value: str) -> None: ... + def watch_password(self: Self, value: str) -> None: ... + def watch_timeout(self: Self, value: str) -> None: ... + def on_host_changed(self: Self, event: pane.SettingsMSGBase) -> None: ... + def on_port_changed(self: Self, event: pane.SettingsMSGBase) -> None: ... + def on_password_changed(self: Self, event: pane.SettingsMSGBase) -> None: ... + def on_timeout_changed(self: Self, event: pane.SettingsMSGBase) -> None: ... + def compose(self: Self) -> ComposeResult: ... + +class OBSCTLPane(PaneBaseSubclass): ... + +class SettingsBlock(Container): + def compose(self) -> ComposeResult: ... diff --git a/src/obs_ctl/interface/molecule/stats_display.py b/src/obs_ctl/interface/molecule/stats_display.py index 2037bad..56b9db5 100644 --- a/src/obs_ctl/interface/molecule/stats_display.py +++ b/src/obs_ctl/interface/molecule/stats_display.py @@ -1,5 +1,5 @@ +import typing from textual.containers import Grid, ScrollableContainer -from textual.reactive import reactive from ..element import TextReadout from .. import obs @@ -76,13 +76,13 @@ class StatsDisplay(ScrollableContainer): else: return 0.0 - cpu_usage: TextReadout = self.query_exactly_one("#cpu") - memory_usage: TextReadout = self.query_exactly_one("#ram") - active_fp: TextReadout = self.query_exactly_one("#fps") - disk_full: TextReadout = self.query_exactly_one("#hdd") - stream_dropped_frames: TextReadout = self.query_exactly_one("#dropped_net") - rendering_lag: TextReadout = self.query_exactly_one("#dropped_ren") - encoding_lag: TextReadout = self.query_exactly_one("#dropped_enc") + cpu_usage: TextReadout = typing.cast(TextReadout, self.query_exactly_one("#cpu")) + memory_usage: TextReadout = typing.cast(TextReadout, self.query_exactly_one("#ram")) + active_fp: TextReadout = typing.cast(TextReadout, self.query_exactly_one("#fps")) + disk_full: TextReadout = typing.cast(TextReadout, self.query_exactly_one("#hdd")) + stream_dropped_frames: TextReadout = typing.cast(TextReadout, self.query_exactly_one("#dropped_net")) + rendering_lag: TextReadout = typing.cast(TextReadout, self.query_exactly_one("#dropped_ren")) + encoding_lag: TextReadout = typing.cast(TextReadout, self.query_exactly_one("#dropped_enc")) netwrk_drp = obs_stats.stream_output_skipped_frames netwrk_tot = obs_stats.stream_output_total_frames render_drp = obs_stats.stats_render_skipped_frames diff --git a/src/obs_ctl/interface/obs/__init__.py b/src/obs_ctl/interface/obs/__init__.py index 7af709b..c00cb17 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, OBSStats \ No newline at end of file +from .obs import API as API, OBSStats as 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 index 656a746..8217b03 100644 --- a/src/obs_ctl/interface/obs/obs.py +++ b/src/obs_ctl/interface/obs/obs.py @@ -3,11 +3,12 @@ from enum import IntEnum, StrEnum from functools import partial import functools from pathlib import Path -from time import ctime import time +from typing import Any, Callable, Self +import typing -import obsws_python as obs -import obsws_python.error as obs_error +import obsws_python as obs # type: ignore +import obsws_python.error as obs_error # type: ignore from textual.timer import Timer from textual.reactive import reactive from textual.widgets import Static @@ -121,17 +122,17 @@ class API(Static): |Report |Posted every interval seconds, a report on the state of the WebSocket.|data |An OBSStats object| """ - event_client: obs.EventClient = None - request_client: obs.ReqClient = None - is_connected: reactive[bool] = False + event_client: obs.EventClient + request_client: obs.ReqClient + is_connected: reactive[bool] = reactive(False) _trying_request: bool = False - _timer: Timer = None + _timer: Timer | None = 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) + ws_timeout: reactive[float] = reactive(3.0) # ENUMS class OutputStates(StrEnum): @@ -167,7 +168,7 @@ class API(Static): # MESSAGES class APIOutputMSGBase(MSGBase): state = None - def __init__(self, control, data): + def __init__(self, control: "API", data: Any) -> None: super().__init__(control) _state = API.OutputStates(data.output_state) logger.debug(f"{self} is an obs.API Message; state: {_state.name}") @@ -185,22 +186,21 @@ class API(Static): class Report(MSGBase): """Posted ever self.interval second. data is an OBSStats object.""" - data: OBSStats = None - def __init__(self, control, data: OBSStats): + def __init__(self: Self, control: "API", data: OBSStats): # this message is sent too often to log it. super().__init__(control, _log_me = False) self.data = data def __init__( - self, - interval, + self: Self, + interval: float, ws_host: str = "localhost", ws_port: int = 4455, ws_password: str = "", - ws_timeout: int | float | None = None, - *args, - **kwargs, + ws_timeout: float = 3.0, + *args: Any, + **kwargs: Any, ): """Initialize the API. @@ -221,53 +221,53 @@ class API(Static): self._reset_timer(self.interval) - def on_stream_state_changed(self: "API", data): + def on_stream_state_changed(self: Self, data: Any) -> None: self.post_message(API.StreamStateUpdated(self, data)) - def on_record_state_changed(self: "API", data): + def on_record_state_changed(self: Self, data: Any) -> None: self.post_message(API.RecordStateUpdated(self, data)) - def watch_interval(self, value): + def watch_interval(self, value: float) -> None: self._reset_timer(value) - def stop_stream(self): + def stop_stream(self) -> None: """Requests to stop streaming.""" self._try_request(self.request_client.stop_stream) - def start_stream(self): + def start_stream(self) -> None: """Requests to start streaming.""" self._try_request(self.request_client.start_stream) - def start_record(self): + def start_record(self) -> None: """Requests to start recording.""" self._try_request(self.request_client.start_record) - def stop_record(self): + def stop_record(self) -> None: """Requests to stop recording.""" self._try_request(self.request_client.stop_record) - def start_pause(self): + def start_pause(self) -> None: """Requests to pause recording.""" self._try_request(self.request_client.pause_record) - def stop_pause(self): + def stop_pause(self) -> None: """Requests to resume recording.""" self._try_request(self.request_client.resume_record) # LOGGER - def set_record_path(self, path: str): + def set_record_path(self, path: str) -> None: """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)) + self._try_request(partial(self.request_client.set_record_directory, path)) # type: ignore else: logger.debug("The path doesn't exist...") @@ -311,7 +311,7 @@ class API(Static): # LOGGER - def connect(self, host=None, port=None, password=None, timeout=None): + def connect(self, host: str | None=None, port: int | None=None, password: str | None=None, timeout: float | None=None) -> 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 @@ -338,16 +338,14 @@ class API(Static): 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) + self.event_client.callback.register(self.on_stream_state_changed) # type: ignore + self.event_client.callback.register(self.on_record_state_changed) # type: ignore - def disconnect(self): + def disconnect(self) -> None: logger.info("Disconnecting from OBS WebSocket") self.request_client.disconnect() self.event_client.disconnect() - self.request_client = None - self.event_client = None self.is_connected = False @@ -363,11 +361,11 @@ class API(Static): # # Tested; finally is reached. No infinite while loop. # LOGGER - def _try_request(self, request:functools.partial): + def _try_request[T: Callable[..., Any]](self: Self, request: T | functools.partial[T]) -> Any: if type(request) is functools.partial: - name = request.func.__name__ + name = typing.cast(functools.partial[T], request).func.__name__ else: - name = request.__name__ + name: str = typing.cast(T, request).__name__ # logger.info(f"Sending request: {name}") output = None @@ -384,8 +382,8 @@ class API(Static): time.sleep(TRY_AGAIN) continue - elif err.code in API.KnownReqErrs: - logger.debug(f"Error handled: {API.KnownReqErrs(err.code)}") + elif err.code in API.KnownReqErrs: # type: ignore + logger.debug(f"Error handled: {API.KnownReqErrs(err.code)}") # type: ignore handled = True else: @@ -403,8 +401,8 @@ class API(Static): # untested - def _reset_timer(self, interval): - self._timer: Timer = self.set_interval(interval, self._send_report) + def _reset_timer(self: Self, interval: float) -> None: + self._timer = self.set_interval(interval, self._send_report) def _send_report(self): diff --git a/src/obs_ctl/log.py b/src/obs_ctl/log.py index c93c788..5d209a9 100644 --- a/src/obs_ctl/log.py +++ b/src/obs_ctl/log.py @@ -28,5 +28,5 @@ only_me_filter = logging.Filter("obs_ctl") console_hdlr.addFilter(only_me_filter) logfile_hdlr.addFilter(only_me_filter) -def getLogger(name): +def getLogger(name: str): return logging.getLogger(name) \ No newline at end of file diff --git a/src/obs_ctl/widget_testbed.py b/src/obs_ctl/widget_testbed.py index 436cb57..4b29666 100644 --- a/src/obs_ctl/widget_testbed.py +++ b/src/obs_ctl/widget_testbed.py @@ -1,11 +1,13 @@ import importlib.resources +import typing from textual.app import App -from textual.widgets import Header, Footer +from textual.widgets import Header +from textual._path import CSSPathType from . import data from .interface.molecule import StatsDisplay -class TestApp(App): - CSS_PATH = importlib.resources.files(data).joinpath("default_layout.tcss") +class TestApp(App): # type: ignore + CSS_PATH = typing.cast(CSSPathType, importlib.resources.files(data).joinpath("default_layout.tcss")) def compose(self): yield Header() yield StatsDisplay() \ No newline at end of file