Extensive type-hinting

diff --git a/src/obs_ctl/__init__.py b/src/obs_ctl/__init__.py
index e01e7a684323cab5587a83fbc33f774e08d316a6..5e4f061dcabcc1116037e0ebc046a897cf9cdffe 100644 (file)
--- 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 af00defcbb661b1b6fde05f53fb445179d14c699..d4f0b1575b0cbdc69a970eba91510eddb9d45323 100644 (file)
--- 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 161ded2b2f5750dbb34663e2f55e737ba5b4081a..a7adfe05b0e542ad311b4f1aee5c4868ee965b2c 100644 (file)
--- 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 9aa087f98e1d184e791f34459adde8870c8329aa..33589d203e670a4794336f0f5faa67cdcbf54040 100644 (file)
--- 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 175958e1f0d3cd885cc04da70b0050385ad6f856..eedfe6d3907f653facc1fd7053daa9a77693793e 100644 (file)
--- 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 1b60115ccd78e2b9af14972040e13275de5d4e39..84c7dca5540f4220770a24110577499e965417df 100644 (file)
--- 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 e99cb643e6d902e9546c6af9b1f9213f5a7bc530..5fb75a3e9e55b1c64c082609b83a3fc315c5e2bc 100644 (file)
--- 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 ba9a9d9ca8835fbd8f2dd376e860623678f453d0..19babc5070a874737782503fc18e0ad4128f3810 100644 (file)
--- 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 4a1e96c6c885222e3fbfdfefc2649282616ef475..028821727b85b8c6916304c9b37f3545f09a5596 100644 (file)
--- 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 b8bfc6a5b2e4cacf3e113efbfa33dc07f406fa36..c595d6f38f354f6e0470a2a6ca467e34a2f63533 100644 (file)
--- 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 ff4a4d1296910afb5f9bbc75c2176395feac4e12..abf522ba7ff47f7f4842b0725285fb8292146109 100644 (file)
--- 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 (file)
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 b0fb94012f84adeecb1dae3121f82756f3ddcd60..0cb9520518bd8df5b5723ee8fd0b239e2c44fa33 100644 (file)
--- 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 80cbfe9a2a82b061fd4699e33562bf81f4f1f48c..abd473b9561d2a4ab3030576c053ec952597d058 100644 (file)
--- 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 2037bad65865c4d233e562bd0bf6f2cf704d74cc..56b9db536d031be0d5d569aec6d8414e67b44185 100644 (file)
--- 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 7af709bade39763d06e34ce81928bc68570af6d4..c00cb174f134d992146a002162089d0be788fcdc 100644 (file)
--- 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 656a7467aebd106b13dfe09f7c5d7e6ee041344d..8217b0368cb5d4ff2e99da989eb559fd7ca59abd 100644 (file)
--- 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 c93c788c90b2bc650de35c3b4b1395e437ff10f1..5d209a9994b48038459a355ead8a3070bc7c1fe3 100644 (file)
--- 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 436cb57dc524ae9401bcd8ffa6ebc6c14cb52456..4b29666572f219b127d2c178a36228a2b0638bbf 100644 (file)
--- 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