summary | shortlog | log | commit | commitdiff | tree
raw | patch | inline | side by side (parent: cc01396)
raw | patch | inline | side by side (parent: cc01396)
| author | Skye Kelley <10640425-sb24kelley@users.noreply.gitlab.com> | |
| Tue, 9 Dec 2025 18:02:37 +0000 (13:02 -0500) | ||
| committer | Skye Kelley <10640425-sb24kelley@users.noreply.gitlab.com> | |
| Tue, 9 Dec 2025 18:02:37 +0000 (13:02 -0500) |
diff --git a/src/obs_ctl/interface/molecule/settings_block/pane.py b/src/obs_ctl/interface/molecule/settings_block/pane.py
index 210fe60a3b598ee1353d4fc4380a8d88ba4eabaa..2750a7a45e03cd7fddff82dfa085a7240d2473b0 100644 (file)
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
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
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
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 510acdaedb1f71bbe8665f82ad090b65bbd054a3..c55a32c6258544b5fac388997d0442d0432322d6 100644 (file)
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
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 9dad39a677f4d50a0791d0837b5223d6024cc652..ac2e8a0e598e4815ed7be9dd5913a861ae609fca 100644 (file)
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,
pane.SettingsWidget.FLOAT,
const.CONFIG.update_interval,
compact=True,
- )
-)
+ ))): ...
+
+
class SettingsBlock(Container):
def compose(self):
with TabbedContent(classes="settings_block--tabbed-content"):