summary | shortlog | log | commit | commitdiff | tree
raw | patch | inline | side by side (parent: 2bceadd)
raw | patch | inline | side by side (parent: 2bceadd)
| author | sbkelley <sb24kelley@gmail.com> | |
| Fri, 28 Nov 2025 08:23:34 +0000 (03:23 -0500) | ||
| committer | sbkelley <sb24kelley@gmail.com> | |
| Fri, 28 Nov 2025 08:23:34 +0000 (03:23 -0500) |
| src/obs_ctl/__init__.py | patch | blob | history | |
| src/obs_ctl/data/layout.tcss | patch | blob | history | |
| src/obs_ctl/widget/__init__.py | patch | blob | history |
index c370fa37a6d0a2ee415904e3f38aef46cbb0d7fa..8ea9005b692830d457a034e33187facb0ad91c03 100644 (file)
--- a/src/obs_ctl/__init__.py
+++ b/src/obs_ctl/__init__.py
import obsws_python as obs
import obsws_python.error as obs_error
+from textual import on
from textual.app import App
-from textual.containers import HorizontalGroup, VerticalGroup, Vertical, Horizontal, Container
+from textual.containers import HorizontalGroup, VerticalGroup, Vertical, Horizontal, Container, Grid
from textual.widgets import Footer, Header, Label, Placeholder
+from textual.css.query import NoMatches
from . import log, data
-from .widget import StreamControls, RecordControls, HorizontalLabels, VerticalLabels
+from .widget import SingleStatusControl, LabeledLabel, LabeledLabel
logger = log.getLogger(__name__)
obs_req_client = obs.ReqClient()
yield Header()
yield Footer()
logger.debug("Composing OBSCtlApp")
- yield Container(
- Label("Controls", id="controls_label", classes="labels"),
- StreamControls(id="stream_ctl", classes="controls"),
- RecordControls(id="record_ctl", classes="controls"),
- id="controls",
- classes="blocks"
- )
- yield Container(
- Label("Statistics", id="stats1_label", classes="labels"),
- Horizontal(
- VerticalLabels("CPU Usage", "-.-%", id="cpu_usage", classes="stats"),
- VerticalLabels("Memory Usage", "-b", id="memory_usage", classes="stats"),
- VerticalLabels("FPS", "--.--", id="fps", classes="stats")
- ),
- classes="blocks",
- )
- yield Container(
- Label("More Statistics", id="stats2_label", classes="labels"),
- HorizontalLabels("Disk full in (approx):", "--:--:--", id="disk_full", classes="stats"),
- HorizontalLabels("Dropped frames (network):", "-/- (-.-%)", id="stream_dropped_frames", classes="stats"),
- classes="blocks",
- )
- yield Container(
- Label("Frames Missed", id="missed_label", classes="labels"),
- Horizontal(
- VerticalLabels("Rendering Lag", "-/- (-.-%)", id="rendering_lag", classes="stats"),
- VerticalLabels("Encoding Lag", "-/- (-.-%)", id="encoding_lag", classes="stats"),
- ),
- classes="blocks"
- )
-
-
- def on_stream_controls_stopped(self, message):
- logger.debug("Stream Stopped message rcv'd")
+ # yield Label("Controls", id="controls_label", classes="labels"),
+ with Grid(id="controls_container", classes="blocks"):
+ yield SingleStatusControl(label="Recording", id="record_control", classes="controls")
+ yield SingleStatusControl(label="Paused", on_label="Pause", off_label="Unpause", swap_red=True, id="pause_control", classes="controls")
+ yield SingleStatusControl(label="Streaming", id="stream_control", classes="controls")
+ with Grid(id="machine_container", classes="blocks"):
+ yield LabeledLabel("CPU Usage", "-.-%", id="cpu_usage", classes="stats")
+ yield LabeledLabel("RAM Usage", "-b", id="memory_usage", classes="stats")
+ yield LabeledLabel("Time 'til Full:", "--:--:--", id="disk_full", classes="stats")
+ with Grid(id="obs_container", classes="blocks"):
+ yield LabeledLabel("FPS", "--.--", id="fps", classes="stats")
+ yield LabeledLabel("Dropped frames (network):", "-/- (-.-%)", id="stream_dropped_frames", classes="stats")
+ yield LabeledLabel("Dropped frames (rendering)", "-/- (-.-%)", id="rendering_lag", classes="stats")
+ yield LabeledLabel("Dropped frames (encoding)", "-/- (-.-%)", id="encoding_lag", classes="stats")
+
+
+ @on(SingleStatusControl.TurnedOff, "#stream_control")
+ def stream_off(self, message):
+ logger.debug(f"Stop message rcv'd from {message.control}")
try:
obs_req_client.stop_stream()
except obs_error.OBSSDKRequestError as err:
else:
raise
-
- def on_stream_controls_started(self, message):
+ @on(SingleStatusControl.TurnedOn, "#stream_control")
+ def stream_on(self, message):
logger.debug("Stream Started message rcv'd")
try:
obs_req_client.start_stream()
else:
raise
-
- def on_record_controls_stopped(self, message):
+ @on(SingleStatusControl.TurnedOff, "#record_control")
+ def record_off(self, message):
logger.debug("Record Stopped message rcv'd")
try:
obs_req_client.stop_record()
else:
raise
-
- def on_record_controls_started(self, message):
+ @on(SingleStatusControl.TurnedOn, "#record_control")
+ def record_on(self, message):
logger.debug("Record Started message rcv'd")
try:
obs_req_client.start_record()
raise
- def on_record_controls_paused(self, message):
+ @on(SingleStatusControl.TurnedOn, "#pause_control")
+ def pause_on(self, message):
logger.debug("Record Paused message rcv'd")
try:
obs_req_client.pause_record()
raise
- def on_record_controls_unpaused(self, message):
+ @on(SingleStatusControl.TurnedOff, "#pause_control")
+ def pause_off(self, message):
logger.debug("Record Unpaused message rcv'd")
try:
obs_req_client.resume_record()
def update_monitors(self, obs_stats: OBSStats):
- stream_ctl: StreamControls = self.query_exactly_one("#stream_ctl")
- record_ctl: RecordControls = self.query_exactly_one("#record_ctl")
- cpu_usage: VerticalLabels = self.query_exactly_one("#cpu_usage")
- memory_usage: VerticalLabels = self.query_exactly_one("#memory_usage")
- active_fp: VerticalLabels = self.query_exactly_one("#fps")
- disk_full: HorizontalLabels = self.query_exactly_one("#disk_full")
- stream_dropped_frames: HorizontalLabels = self.query_exactly_one("#stream_dropped_frames")
- rendering_lag: VerticalLabels = self.query_exactly_one("#rendering_lag")
- encoding_lag: VerticalLabels = self.query_exactly_one("#encoding_lag")
-
- stream_ctl.set_light(obs_stats.stream_output_active)
- record_ctl.set_light(obs_stats.record_output_active)
- record_ctl.set_pause_light(obs_stats.record_output_paused)
+ stream_ctl: SingleStatusControl = self.query_exactly_one("#stream_control")
+ record_ctl: SingleStatusControl = self.query_exactly_one("#record_control")
+ pause_ctl: SingleStatusControl = self.query_exactly_one("#pause_control")
+ cpu_usage: LabeledLabel = self.query_exactly_one("#cpu_usage")
+ memory_usage: LabeledLabel = self.query_exactly_one("#memory_usage")
+ active_fp: LabeledLabel = self.query_exactly_one("#fps")
+ disk_full: LabeledLabel = self.query_exactly_one("#disk_full")
+ stream_dropped_frames: LabeledLabel = self.query_exactly_one("#stream_dropped_frames")
+ rendering_lag: LabeledLabel = self.query_exactly_one("#rendering_lag")
+ encoding_lag: LabeledLabel = self.query_exactly_one("#encoding_lag")
+
+ stream_ctl.status = obs_stats.stream_output_active
+ record_ctl.status = obs_stats.record_output_active
+ pause_ctl.status = obs_stats.record_output_paused
cpu_usage.text = f"{obs_stats.cpu_usage_percent:0.5}%"
memory_usage.text = f"{obs_stats.memory_usage_mb:0.5}MB"
active_fp.text = f"{obs_stats.active_fps}"
index 0d872bb3d92e73bb7dfa65632b8b71f3a3d9d4fb..f724044601f1c6bdb7cc9b481b0bf13850bcc002 100644 (file)
-OBSCtlApp {
- layout: grid;
- grid-size: 2 2;
-}
+Screen {
+ layout:grid;
+ grid-size:2 2;
+ background:$background;
-StreamControls {
- layout:horizontal;
- max-height:1;
- & Button {
+ .blocks {
+ background:$background;
+ grid-size:2 2;
width:1fr;
- }
-}
-RecordControls {
- layout:vertical;
- max-height:2;
- & Button {
- width:1fr;
- }
- # & RadioButton {
- # max-width:20;
- # }
-}
+ border:solid $primary;
+
+ SingleStatusControl {
+ background:$surface;
+ border:blank;
+ .singlestatuscontrol--light {
+ width:1fr;
+ height:1fr;
+ background:$panel;
+ text-align:center;
+ align:center bottom;
+ }
+ .singlestatuscontrol--on_button {
+ width:1fr;
+ height:1fr;
+ }
+ .singlestatuscontrol--off_button {
+ width:1fr;
+ height:1fr;
+ }
+ }
+
+ LabeledLabel {
+ width:1fr;
+ border:solid $primary;
+ background:$surface;
+ .labeledlabel--constant {
+ width:1fr;
+ height:1;
+ align:center middle;
+ }
+ .labeledlabel--variable {
+ align:center middle;
+ }
+ }
-VerticalLabels {
- background:#ffffff;
- max-width:5;
+ &#controls_container {
+ width: 1fr;
+ column-span:2;
+ }
+
+ }
}
\ No newline at end of file
index d5750e45f9e05dbd23483f8cdd89573a2a0178f3..4403fc98700abe4c6f4522be6e25ec4436d88448 100644 (file)
from textual.reactive import reactive
from textual.widgets import Label, RadioButton, Button
from textual.widget import Widget
+from textual.css import query
+
from .. import log
logger = log.getLogger(__name__)
+class SingleStatusControl(Container):
+ COMPONENT_CLASSES = {
+ "singlestatuscontrol--light",
+ "singlesttauscontrol--on_button",
+ "singlestatuscontrol--off_button",
+ }
+ status = reactive(False)
+ label = reactive("Status")
+ on_label = reactive("Start")
+ off_label = reactive("Stop")
+ swap_red = reactive(False)
-class SRControls(Widget):
- light_active = reactive(False)
class MsgBase(Message):
- def __init__(self, srcontrols: "SRControls"):
- logger.debug(f"{self} sent from {srcontrols}")
- self.srcontrol = srcontrols
+ control: "SingleStatusControl" = None
+ def __init__(self, control):
super().__init__()
+ self.control = control
+ class TurnedOn(MsgBase): ...
+ class TurnedOff(MsgBase): ...
+
+
+ def __init__(
+ self,
+ *children,
+ label: str = "Status",
+ on_label = "Start",
+ off_label = "Stop",
+ swap_red = False,
+ layout = "vertical",
+ name = None,
+ id = None,
+ classes = None,
+ disabled = False,
+ markup = True):
+ super().__init__(*children, name=name, id=id, classes=classes, disabled=disabled, markup=markup)
+ self.status = False
+ self.label = label
+ self.on_label = on_label
+ self.off_label = off_label
+ self.swap_red = swap_red
+ self.set_styles(layout=layout)
- def on_button_pressed(self, event: Button.Pressed):
- # light: RadioButton = self.query_exactly_one("#light")
- if event.button.id == "start":
- logger.debug(f"Start pressed on {self}")
- self.post_message(self.Started(self))
- # light.value = True
- elif event.button.id == "stop":
- logger.debug(f"Stop pressed on {self}")
- self.post_message(self.Stopped(self))
- # light.value = False
-
-
- def set_light(self, value: bool) -> None:
- light: RadioButton = self.query_exactly_one("#light")
- light.value = value
-
-
-class StreamControls(SRControls):
- # DEFAULT_CSS = """
- # StreamControls {
- # layout: grid;
- # grid-size:3 1;
- # height:1fr;
- # width:1fr;
- # # grid-columns:2fr 1fr 1fr;
- # # grid-rows:1fr;
- # }
- # """
- class Started(SRControls.MsgBase): ...
- class Stopped(SRControls.MsgBase): ...
def compose(self) -> ComposeResult:
- logger.debug("Composing StreamControls")
- yield RadioButton(
- label="Streaming",
- value=False,
- button_first=False,
- id="light",
- compact=True,
- disabled=True
- )
+ logger.debug("Composing SingleStatusControl")
+ with Container(classes="singlestatuscontrol--light"):
+ yield RadioButton(
+ label="",
+ value=False,
+ button_first=False,
+ id="light",
+ compact=True,
+ disabled=True,
+ )
yield Button(
label="Start",
- id="start",
+ id="on",
variant="success",
compact=True,
+ classes="singlestatuscontrol--on_button"
)
yield Button(
label="Stop",
- id="stop",
+ id="off",
variant="error",
compact=True,
+ classes = "singlestatuscontrol--off_button"
)
-class RecordControls(SRControls):
- # DEFAULT_CSS = """
- # RecordControls {
- # layout: grid;
- # grid-size:3 2;
- # height:1fr;
- # width:1fr;
- # grid-columns: 2fr 1fr 1fr;
- # grid-rows: 1fr 1fr;
- # }
- # """
-
- class Started(SRControls.MsgBase): ...
- class Stopped(SRControls.MsgBase): ...
- class Paused(SRControls.MsgBase): ...
- class Unpaused(SRControls.MsgBase): ...
-
-
- def compose(self):
- logger.debug("Composing RecordControls")
- with Horizontal():
- yield RadioButton(
- label="Recording",
- value=False,
- button_first=False,
- compact=True,
- disabled=True,
- id="light",
- )
- yield Button(
- "Start",
- id="start",
- variant="success",
- compact = True,
- )
- yield Button(
- "Stop",
- id="stop",
- variant="error",
- compact=True,
- )
- with Container():
- yield RadioButton(
- label="Paused",
- value=False,
- button_first=False,
- id="pause_light",
- compact=True,
- disabled=True
- )
- yield Button(
- "Pause",
- id="pause_record",
- variant="error",
- compact=True,
- )
- yield Button(
- "Unpause",
- id="unpause_record",
- variant="success",
- compact=True,
- )
+ def _on_mount(self, event):
+ # super()._on_mount(event)
+ self.watch_status(self.status)
+ self.watch_label(self.label)
+ self.watch_on_label(self.on_label)
+ self.watch_off_label(self.off_label)
+ self.watch_swap_red(self.swap_red)
+ self.refresh()
def on_button_pressed(self, event: Button.Pressed):
- if event.button.id == "pause_record":
- logger.debug(f"Pause pressed on {self}")
- self.post_message(self.Paused(self))
- elif event.button.id == "unpause_record":
- logger.debug(f"Unpause pressed on {self}")
- self.post_message(self.Unpaused(self))
-
-
- def set_pause_light(self, value):
- light: RadioButton = self.query_exactly_one("#pause_light")
- light.value = value
-
-
-class HorizontalLabels(Widget):
- # DEFAULT_CSS = """
- # HorizontalLabels {
- # layout: horizontal;
- # height:1fr;
- # width:1fr;
- # }
- # """
- text = reactive("")
- def __init__(self, label_one, label_two, *children, name = None, id = None, classes = None, disabled = False, markup = True):
- self._label_one_param = label_one
- self._label_two_param = label_two
- super().__init__(*children, name=name, id=id, classes=classes, disabled=disabled, markup=markup)
-
-
- def compose(self):
- yield Label(self._label_one_param, id="constant")
- yield Label(self._label_two_param, id="variable")
+ if event.button.id == "on":
+ logger.debug(f"'On' pressed on {event.control}")
+ self.post_message(self.TurnedOn(self))
+ elif event.button.id == "off":
+ logger.debug(f"'Off' pressed on {event.control}")
+ self.post_message(self.TurnedOff(self))
+
+
+ def watch_status(self, value: bool) -> None:
+ try:
+ light: RadioButton = self.query_exactly_one("#light")
+ light.value = value
+ except query.NoMatches:
+ if self.is_mounted:
+ raise
+
+
+ def watch_label(self, value):
+ try:
+ light: RadioButton = self.query_exactly_one("#light")
+ light.label = value
+ except query.NoMatches:
+ if self.is_mounted:
+ raise
+
+ def watch_on_label(self, value):
+ try:
+ on: Button = self.query_exactly_one("#on")
+ on.label = value
+ except query.NoMatches:
+ if self.is_mounted:
+ raise
+
+ def watch_off_label(self, value):
+ try:
+ off: Button = self.query_exactly_one("#off")
+ off.label = value
+ except query.NoMatches:
+ if self.is_mounted:
+ raise
-
- def watch_text(self, text: str):
- logger.debug(f"Setting label {self} text to {text}")
- label: Label = self.query_exactly_one("#variable")
- label.update(text)
-
-class VerticalLabels(Widget):
- # DEFAULT_CSS = """
- # VerticalLabels {
- # layout: vertical;
- # height: 1fr;
- # width:1fr;
- # }
- # """
+ def watch_swap_red(self, value):
+ try:
+ on: Button = self.query_exactly_one("#on")
+ off: Button = self.query_exactly_one("#off")
+ off.variant = "success" if value else "error"
+ on.variant = "error" if value else "success"
+ except query.NoMatches:
+ if self.is_mounted:
+ raise
+
+
+class LabeledLabel(Container):
+ COMPONENT_CLASSES = {
+ "labeledlabel--constant",
+ "labeledlabel--variable",
+ }
text = reactive("")
- def __init__(self, label_one, label_two, *children, name = None, id = None, classes = None, disabled = False, markup = True):
+ def __init__(self, label_one, label_two, *children, layout = "vertical", name = None, id = None, classes = None, disabled = False, markup = True):
+ super().__init__(*children, name=name, id=id, classes=classes, disabled=disabled, markup=markup)
self._label_one_param = label_one
self._label_two_param = label_two
- super().__init__(*children, name=name, id=id, classes=classes, disabled=disabled, markup=markup)
+ self.set_styles(layout=layout)
+
+ # def _on_mount(self):
+ # self.query_exactly_one("#constant").styles = self.get_component_styles("labeledlabel--constant")
+ # self.query_exactly_one("#variable").styles = self.get_component_styles("labeledlabel--variable")
def compose(self):
- yield Label(self._label_one_param, id="constant")
- yield Label(self._label_two_param, id="variable")
+ yield Container(Label(self._label_one_param, id="constant"), classes="labeledlabel--constant")
+ yield Container(Label(self._label_two_param, id="variable"), classes="labeledlabel--variable")
def watch_text(self, text: str):
- logger.debug(f"Setting label {self} text to {text}")
label: Label = self.query_exactly_one("#variable")
label.update(text)
\ No newline at end of file