from dataclasses import dataclass
import importlib.resources
+from pathlib import Path
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, Grid
-from textual.widgets import Footer, Header, Label, Placeholder
+from textual.containers import Vertical, Horizontal, Grid, ScrollableContainer, Container
+from textual.widgets import Footer, Header, Label, Placeholder, TabbedContent, TabPane, Markdown, Input
from textual.css.query import NoMatches
from . import log, data
-from .widget import SingleStatusControl, LabeledLabel, LabeledLabel
+from .widget import SingleStatusControl, LabeledLabel, OBSSettingsPane
logger = log.getLogger(__name__)
obs_req_client = obs.ReqClient()
memory_usage_mb: float
active_fps: float
available_disk_space: float
+ record_directory: str
@property
def estimated_hours_remaining(self):
yield Header()
yield Footer()
logger.debug("Composing OBSCtlApp")
- # 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")
+ with ScrollableContainer(id="content_container"):
+ with Vertical(id="blocks_v_container", classes="blocks_containers"):
+ with Horizontal(id="blocks_h_container1", classes="blocks_h_containers"):
+ with Grid(id="controls_container", classes="blocks gridblocks"):
+ 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 TabbedContent(id="settings_container", classes="blocks"):
+ with TabPane("OBS Settings", id="settings-tab"):
+ yield OBSSettingsPane(id="settings-content")
+ with TabPane("OBS Connection", id="connection-tab"):
+ yield Placeholder("CONNECTION BLOCK", id="conection-plc")
+ with TabPane("obs-ctl Settings", id="clients-tab"):
+ yield Placeholder("CLIENT SETTINGS BLOCK", id="client-plc")
+ with Horizontal(id="blocks_h_container2", classes="blocks_h_containers"):
+ with Grid(id="machine_container", classes="blocks gridblocks"):
+ yield LabeledLabel("CPU Usage", "-.-%", id="cpu_usage", classes="stats")
+ yield LabeledLabel("RAM Usage", "-b", id="memory_usage", classes="stats")
+ yield LabeledLabel("Time 'til Full", "N/A", id="disk_full", classes="stats")
+ with Grid(id="obs_container", classes="blocks gridblocks"):
+ yield LabeledLabel("FPS", "--.--", id="fps", classes="stats")
+ yield LabeledLabel("Netwrk frames dropped", "-/- (-.-%)", id="stream_dropped_frames", classes="stats")
+ yield LabeledLabel("Render frames dropped", "-/- (-.-%)", id="rendering_lag", classes="stats")
+ yield LabeledLabel("Encdng frames dropped", "-/- (-.-%)", id="encoding_lag", classes="stats")
@on(SingleStatusControl.TurnedOff, "#stream_control")
raise
+ @on(OBSSettingsPane.RecordPathChanged, "#settings-content")
+ def set_record_path(self, message: OBSSettingsPane.RecordPathChanged):
+ logger.debug(f"Set record path rcv'd: {message.record_path}")
+ if Path(message.record_path).exists():
+ try:
+ obs_req_client.set_record_directory(message.record_path)
+ except obs_error.OBSSDKRequestError as err:
+ if err.code == 500:
+ logger.debug("Recording is running.")
+ else:
+ raise
+ else:
+ logger.debug("The path doesn't exist...")
+
@staticmethod
def get_obs_stats():
record_status = obs_req_client.get_record_status()
stream_status = obs_req_client.get_stream_status()
obs_stats = obs_req_client.get_stats()
+ record_directory = obs_req_client.get_record_directory()
# logger.debug(record_status)
# logger.debug(stream_status)
# logger.debug(obs_stats.attrs())
memory_usage_mb=obs_stats.memory_usage,
active_fps=obs_stats.active_fps,
available_disk_space=obs_stats.available_disk_space,
+ record_directory=record_directory.record_directory,
)
return output
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")
+ obs_settings: OBSSettingsPane = self.query_exactly_one("#settings-content")
stream_ctl.status = obs_stats.stream_output_active
record_ctl.status = obs_stats.record_output_active
stream_dropped_frames.text = f"{obs_stats.stream_output_skipped_frames}/{obs_stats.stream_output_total_frames}"
rendering_lag.text = f"{obs_stats.stats_render_skipped_frames}/{obs_stats.stats_render_total_frames}"
encoding_lag.text = f"{obs_stats.stats_output_skipped_frames}/{obs_stats.stats_output_total_frames}"
+ obs_settings.record_path = obs_stats.record_directory
+
def process_obs_stats(self):
from textual.app import App, ComposeResult
-from textual.containers import Horizontal, Vertical, Grid, Container
+from textual.containers import Horizontal, VerticalScroll, Grid, Container
from textual.message import Message
from textual.reactive import reactive
-from textual.widgets import Label, RadioButton, Button
+from textual.widgets import Label, RadioButton, Button, Input
from textual.widget import Widget
from textual.css import query
logger = log.getLogger(__name__)
+class OBSSettingsPane(VerticalScroll):
+ DEFAULT_CSS = """
+ .obssettingspane--input {
+ width:1fr;
+ }
+ Grid {
+ grid-size:2 1;
+ grid-columns:21 1fr;
+ Label {
+ width:1fr;
+ text-align:right;
+ padding-right:1;
+ }
+ }
+ """
+ COMPONENT_CLASSES = {
+ "obssettingspane--input",
+ }
+ record_path: reactive[str] = reactive("")
+ class RecordPathChanged(Message):
+ control: "OBSSettingsPane" = None
+ record_path: str = None
+ def __init__(self, ctrl, path):
+ super().__init__()
+ self.control = ctrl
+ self.record_path = path
+
+ def compose(self):
+ with Grid():
+ yield Label("Record path:")
+ yield Input("path", id="input_record_path", classes="obssettingspane--input", compact=True)
+
+ def on_input_submitted(self, event:Input.Submitted):
+ self.set_reactive(OBSSettingsPane.record_path, event.value)
+ if event.control.id == "input_record_path":
+ self.post_message(OBSSettingsPane.RecordPathChanged(self, event.value))
+
+ def watch_record_path(self, value):
+ input: Input = self.query_exactly_one("#input_record_path")
+ input.value = value
+
+
class SingleStatusControl(Container):
+ DEFAULT_CSS = """
+ .singlestatuscontrol--light {
+ background:$panel;
+ text-align:center;
+ align:center bottom;
+ width:1fr;
+ height:1fr;
+ }
+ .singlestatuscontrol--on_button {
+ width:1fr;
+ height:1fr;
+ }
+ .singlestatuscontrol--off_button {
+ width:1fr;
+ height:1fr;
+ }
+ """
COMPONENT_CLASSES = {
"singlestatuscontrol--light",
"singlesttauscontrol--on_button",