From: Skye Kelley <10640425-sb24kelley@users.noreply.gitlab.com> Date: Tue, 9 Dec 2025 18:02:37 +0000 (-0500) Subject: Make pane factory into a Metaclass X-Git-Url: https://skyeroc.xyz/gitweb/?a=commitdiff_plain;h=refs%2Fheads%2Fdevelopment;p=obs-ctl Make pane factory into a Metaclass --- diff --git a/src/obs_ctl/interface/molecule/settings_block/pane.py b/src/obs_ctl/interface/molecule/settings_block/pane.py index 210fe60..2750a7a 100644 --- a/src/obs_ctl/interface/molecule/settings_block/pane.py +++ b/src/obs_ctl/interface/molecule/settings_block/pane.py @@ -1,7 +1,7 @@ from enum import Enum from itertools import zip_longest from logging import getLogger -from typing import Any, Mapping, Sequence +from typing import Any, Mapping, Sequence, Type from textual import on from textual.app import ComposeResult @@ -9,6 +9,7 @@ from textual.containers import ScrollableContainer, Grid from textual.reactive import reactive from textual.widget import Widget from textual.widgets import TabPane, Input, Label, Select +from textual.message_pump import _MessagePumpMeta, _MessagePumpMetaSub from ...message import MSGBase @@ -78,14 +79,8 @@ type KWArgsType = Mapping[str, Any] type ParamsType = tuple[ArgsType, KWArgsType] | None type ParamsListType = Sequence[ParamsType] type LoopListType = Sequence[MakeWidgetOutputType] -def make_pane( - cls_name: str, - *args, - names_list: list[str] | LoopListType | None=None, - settings_type_list: list[SettingsWidget] | None=None, - defaults_list: list[Any] | None=None, - params_list: ParamsListType | None=None, - ) -> type[PaneBase]: + +class PaneMeta(_MessagePumpMeta): """Creates a new TabPane subclass with cls_name as its name. All the arguments must be keyword arguments, or all of the arguments @@ -121,79 +116,93 @@ def make_pane( An on_record_path_changed function, which (upon Input submission) emits ... A RecordPathChanged message containing the submitted value. """ - # Detect whether this is a "looplist" style input or a parameters style input - ROW_HEIGHT = 1 - if names_list is not None: - params_list = params_list if params_list is not None else ( ((), dict()), ) - loop_list: LoopListType = list(zip_longest(names_list, settings_type_list, defaults_list, params_list)) # type: ignore - else: - loop_list: LoopListType = args # type: ignore - - 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: 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: SettingsWidgetMsgType, - listener_id: str, - reactive: reactive[Any], - post_message: type[SettingsMSGBase]): - def on_func(self: PaneBase, event: SettingsWidgetMsg): - 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 compose(self: PaneBase) -> ComposeResult: - with ScrollableContainer(classes="--container"): - with Grid(): - for setting, setting_type, setting_default, param_spec in loop_list: - # param_spec might be None if it was a short list or had to be padded - param_spec = param_spec if param_spec is not None else ((), dict()) - ctrl_id = f"{setting_type.widget.__name__.lower()}_{setting}" - yield Label(setting) - yield setting_type.widget( - *param_spec[0], - **param_spec[1], - value=setting_type.expected_type(setting_default), - id=ctrl_id) - return compose - - new_dict: dict[str, Any] = dict() - for setting_name, setting_type, setting_default, _ in loop_list: - setting_control_id: str = f"{setting_type.widget.__name__.lower()}_{setting_name}" - SettingChanged_name: str = f"{_snake_to_camel(setting_name)}Submitted" - watch_setting_name: str = f"watch_{setting_name}" - on_setting_changed_name: str = f"on_{setting_name}_submitted" - # For each desired setting (e.g. 'record_path'): - # Set up a record_path reactive - new_dict[setting_name] = reactive(setting_type.type(setting_default)) - # Set up a watch_record_path function - new_dict[watch_setting_name] = make_watch_reactive(f"#{setting_control_id}") - # Set up a RecordPathChanged message for when user interacts with control - # (see SettingTypes enum) - new_dict[SettingChanged_name] = type(SettingChanged_name, (SettingsMSGBase, ), {}) - # Set up on_record_path_changed function to update record_path - # and post RecordPathChanged - new_dict[on_setting_changed_name] = make_on_control_message( - listener_message=setting_type.msg, - listener_id=f"#{setting_control_id}", - reactive=new_dict[setting_name], - post_message=new_dict[SettingChanged_name], - ) - # Set up the default CSS - new_dict["DEFAULT_CSS"] = DEFAULT_CSS - # Set up compose() - new_dict["compose"] = make_compose() - - # Construct the actual class and return it - new_pane: type[PaneBase] = type(cls_name, (PaneBase,), new_dict) - return new_pane + + def __new__( + cls: type[_MessagePumpMetaSub], + name: str, + bases: tuple[type, ...], + classvars: dict[str, Any], + names_list: list[str] | LoopListType | None=None, + settings_type_list: list[SettingsWidget] | None=None, + defaults_list: list[Any] | None=None, + params_list: ParamsListType | None=None, + panes_list: tuple[LoopListType] | None=None, + **kwargs: Any) -> _MessagePumpMetaSub: + ROW_HEIGHT = 1 + if panes_list is None: + params_list = params_list if params_list is not None else ( ((), dict()), ) + loop_list: LoopListType = list(zip_longest(names_list, settings_type_list, defaults_list, params_list)) # type: ignore + else: + loop_list: LoopListType = panes_list # type: ignore + + DEFAULT_CSS = f"Grid {{grid-size:2;grid-rows:{ROW_HEIGHT};min-height:{len(loop_list)*ROW_HEIGHT}}}" + + new_dict: dict[str, Any] = classvars + for setting_name, setting_type, setting_default, _ in loop_list: + setting_control_id: str = f"{setting_type.widget.__name__.lower()}_{setting_name}" + SettingChanged_name: str = f"{_snake_to_camel(setting_name)}Submitted" + watch_setting_name: str = f"watch_{setting_name}" + on_setting_changed_name: str = f"on_{setting_name}_submitted" + # For each desired setting (e.g. 'record_path'): + # Set up a record_path reactive + new_dict[setting_name] = reactive(setting_type.type(setting_default)) + # Set up a watch_record_path function + new_dict[watch_setting_name] = make_watch_reactive(f"#{setting_control_id}") + # Set up a RecordPathChanged message for when user interacts with control + # (see SettingTypes enum) + new_dict[SettingChanged_name] = type(SettingChanged_name, (SettingsMSGBase, ), {}) + # Set up on_record_path_changed function to update record_path + # and post RecordPathChanged + new_dict[on_setting_changed_name] = make_on_control_message( + listener_message=setting_type.msg, + listener_id=f"#{setting_control_id}", + reactive=new_dict[setting_name], + post_message=new_dict[SettingChanged_name], + ) + # Set up the default CSS + new_dict["DEFAULT_CSS"] = DEFAULT_CSS + # Set up compose() + new_dict["compose"] = make_compose(loop_list) + + return super().__new__(cls, name, bases, dict(new_dict), **kwargs) #type: ignore + # return super().__new__(cls, cls_name, (PaneBase, ), dict(new_dict)) + + +def make_watch_reactive(watch_id: str): + 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: SettingsWidgetMsgType, + listener_id: str, + reactive: reactive[Any], + post_message: type[SettingsMSGBase]): + def on_func(self: PaneBase, event: SettingsWidgetMsg): + 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(loop_list: LoopListType): + def compose(self: PaneBase) -> ComposeResult: + with ScrollableContainer(classes="--container"): + with Grid(): + for setting, setting_type, setting_default, param_spec in loop_list: + # param_spec might be None if it was a short list or had to be padded + param_spec = param_spec if param_spec is not None else ((), dict()) + ctrl_id = f"{setting_type.widget.__name__.lower()}_{setting}" + yield Label(setting) + yield setting_type.widget( + *param_spec[0], + **param_spec[1], + value=setting_type.expected_type(setting_default), + id=ctrl_id) + return compose + def _snake_to_camel(string: str): return str.join("", [str.capitalize(x) for x in string.split("_")]) \ No newline at end of file diff --git a/src/obs_ctl/interface/molecule/settings_block/pane.pyi b/src/obs_ctl/interface/molecule/settings_block/pane.pyi index 510acda..c55a32c 100644 --- a/src/obs_ctl/interface/molecule/settings_block/pane.pyi +++ b/src/obs_ctl/interface/molecule/settings_block/pane.pyi @@ -4,6 +4,8 @@ from typing import Any, ClassVar, Literal, Mapping, Self, Sequence, overload from textual.widget import Widget from textual.widgets import Input, Select, TabPane from textual.reactive import reactive +from textual.message_pump import _MessagePumpMeta + from ...message import MSGBase class SettingsMSGBase(MSGBase): value: Any @@ -37,6 +39,26 @@ type ArgsType = Sequence[Any] type KWArgsType = Mapping[str, Any] type ParamsType = tuple[ArgsType, KWArgsType] type ParamsListType = Sequence[ParamsType | None] + + +class PaneMeta(_MessagePumpMeta): ... +# @overload +# def __new__( +# cls, +# cls_name: str, +# *, +# names_list: Sequence[str], +# settings_type_list: Sequence[SettingsWidget], +# defaults_list: Sequence[Any], +# params_list: ParamsListType | None=None,) -> type[PaneBase]: ... +# @overload +# def __new__( +# cls, +# cls_name: str, +# *args: MakeWidgetOutputType, +# ) -> type[PaneBase]: ... + + @overload def make_pane( cls_name: str, 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 9dad39a..ac2e8a0 100644 --- a/src/obs_ctl/interface/molecule/settings_block/settings_block.py +++ b/src/obs_ctl/interface/molecule/settings_block/settings_block.py @@ -4,25 +4,28 @@ from textual.containers import Container from . import pane from .... import const -OBSSettingsPane = pane.make_pane( - "OBSSettingsPane", +class OBSSettingsPane( + pane.PaneBase, + metaclass=pane.PaneMeta, names_list=["record_path"], settings_type_list=[pane.SettingsWidget.STRING], defaults_list=[""], params_list=[((), {"compact":True}),] -) + ): ... -OBSWebSocketPane = pane.make_pane( - "OBSWebSocketPane", - pane.make_pane_widget("host", pane.SettingsWidget.STRING, const.CONFIG.host, compact=True), +class OBSWebSocketPane( + pane.PaneBase, + metaclass=pane.PaneMeta, + panes_list = (pane.make_pane_widget("host", pane.SettingsWidget.STRING, const.CONFIG.host, compact=True), pane.make_pane_widget("port", pane.SettingsWidget.INT, const.CONFIG.port, compact=True), pane.make_pane_widget("password", pane.SettingsWidget.STRING, const.CONFIG.password, compact=True), - pane.make_pane_widget("timeout", pane.SettingsWidget.FLOAT, const.CONFIG.timeout, compact=True), -) + pane.make_pane_widget("timeout", pane.SettingsWidget.FLOAT, const.CONFIG.timeout, compact=True),) + ): ... -OBSCTLPane = pane.make_pane( - "OBSCTLPane", - pane.make_pane_widget( +class OBSCTLPane( + pane.PaneBase, + metaclass=pane.PaneMeta, + panes_list = (pane.make_pane_widget( "theme", pane.SettingsWidget.LIST, const.CONFIG.theme, @@ -34,8 +37,9 @@ OBSCTLPane = pane.make_pane( pane.SettingsWidget.FLOAT, const.CONFIG.update_interval, compact=True, - ) -) + ))): ... + + class SettingsBlock(Container): def compose(self): with TabbedContent(classes="settings_block--tabbed-content"):