separate the cli commands into different modules

this makes it easier to use them as components of both the cli and the gui.
This commit is contained in:
onyx-and-iris 2025-06-26 06:15:48 +01:00
parent 76815926e9
commit ae86785ba6
4 changed files with 129 additions and 164 deletions

View File

@ -1,164 +0,0 @@
import logging
from datetime import datetime
import clypi
import FreeSimpleGUI as fsg
import obsws_python as obsws
from clypi import ClypiConfig, ClypiException, Command, Positional, arg, configure
from typing_extensions import override
logger = logging.getLogger(__name__)
config = ClypiConfig(
nice_errors=(ClypiException,),
)
configure(config)
highlight = clypi.Styler(fg="green")
error = clypi.Styler(fg="red", bold=True)
class Start(Command):
"""Start recording."""
filename: Positional[str] = arg(
default="default_name",
help="Name of the recording",
prompt="Enter the name for the recording",
)
host: str = arg(inherited=True)
port: int = arg(inherited=True)
password: str = arg(inherited=True)
@staticmethod
def get_timestamp():
return datetime.now().strftime("%Y-%m-%d %H-%M-%S")
@override
async def run(self):
if not self.filename:
raise ClypiException("Recording name cannot be empty.")
with obsws.ReqClient(
host=self.host, port=self.port, password=self.password
) as client:
resp = client.get_record_status()
if resp.output_active:
raise ClypiException("Recording is already active.")
filename = f"{self.filename} {self.get_timestamp()}"
client.set_profile_parameter(
"Output",
"FilenameFormatting",
filename,
)
client.start_record()
print(f"Recording started with filename: {highlight(filename)}")
class Stop(Command):
"""Stop recording."""
host: str = arg(inherited=True)
port: int = arg(inherited=True)
password: str = arg(inherited=True)
@override
async def run(self):
with obsws.ReqClient(
host=self.host, port=self.port, password=self.password
) as client:
resp = client.get_record_status()
if not resp.output_active:
raise ClypiException("Recording is not active.")
client.stop_record()
print("Recording stopped successfully.")
def theme_parser(value: str) -> str:
"""Parse the theme argument."""
themes = [
"Light Purple",
"Neutral Blue",
"Reds",
"Sandy Beach",
"Kayak",
"Light Blue 2",
"Dark Teal1",
]
if value not in themes:
raise ClypiException(
f"Invalid theme: {value}. Available themes: {', '.join(themes)}"
)
return value
class SimpleRecorder(Command):
subcommand: Start | Stop | None = None
host: str = arg(default="localhost", env="OBS_HOST")
port: int = arg(default=4455, env="OBS_PORT")
password: str | None = arg(default=None, env="OBS_PASSWORD")
theme: str = arg(default="Reds", parser=theme_parser, env="OBS_THEME")
@override
async def run(self):
fsg.theme(self.theme)
input_text = fsg.InputText("", key="-FILENAME-")
start_record_button = fsg.Button("Start Recording", key="Start Recording")
stop_record_button = fsg.Button("Stop Recording", key="Stop Recording")
layout = [
[fsg.Text("Enter recording filename:")],
[input_text],
[start_record_button, stop_record_button],
[fsg.Text("Status: Not started", key="-OUTPUT-")],
]
window = fsg.Window("Simple Recorder", layout, finalize=True)
status_text = window["-OUTPUT-"]
input_text.bind("<Return>", "-ENTER-")
start_record_button.bind("<Return>", "-ENTER-")
stop_record_button.bind("<Return>", "-ENTER-")
while True:
event, values = window.read()
logger.debug(f"Event: {event}, Values: {values}")
if event == fsg.WIN_CLOSED:
break
elif event in (
"Start Recording",
"Start Recording-ENTER-",
"-FILENAME--ENTER-",
):
try:
await Start(
filename=input_text.get(),
host=self.host,
port=self.port,
password=self.password,
).run()
status_text.update("Status: Recording started", text_color="green")
except ClypiException as e:
status_text.update(str(e), text_color="red")
logger.error(f"Error starting recording: {e}")
elif event in ("Stop Recording", "Stop Recording-ENTER-"):
try:
await Stop(
host=self.host,
port=self.port,
password=self.password,
).run()
status_text.update("Status: Recording stopped", text_color="green")
except ClypiException as e:
status_text.update(str(e), text_color="red")
logger.error(f"Error stopping recording: {e}")
def run():
"""Run the application."""
SimpleRecorder.parse().start()
if __name__ == "__main__":
run()

View File

@ -0,0 +1,57 @@
import logging
from clypi import ClypiConfig, ClypiException, Command, arg, configure
from typing_extensions import override
from .errors import SimpleRecorderError
from .gui import SimpleRecorderWindow
from .start import Start
from .stop import Stop
logger = logging.getLogger(__name__)
config = ClypiConfig(
nice_errors=(SimpleRecorderError,),
)
configure(config)
def theme_parser(value: str) -> str:
"""Parse the theme argument."""
themes = [
"Light Purple",
"Neutral Blue",
"Reds",
"Sandy Beach",
"Kayak",
"Light Blue 2",
"Dark Teal1",
]
if value not in themes:
raise ClypiException(
f"Invalid theme: {value}. Available themes: {', '.join(themes)}"
)
return value
class SimpleRecorder(Command):
subcommand: Start | Stop | None = None
host: str = arg(default="localhost", env="OBS_HOST", help="OBS WebSocket host")
port: int = arg(default=4455, env="OBS_PORT", help="OBS WebSocket port")
password: str | None = arg(
default=None, env="OBS_PASSWORD", help="OBS WebSocket password"
)
theme: str = arg(
default="Reds", parser=theme_parser, env="OBS_THEME", help="OBS WebSocket theme"
)
@override
async def run(self):
"""Run the Simple Recorder CLI."""
window = SimpleRecorderWindow(self.host, self.port, self.password, self.theme)
await window.run()
def run():
"""Run the application."""
SimpleRecorder.parse().start()

View File

@ -0,0 +1,46 @@
from datetime import datetime
import obsws_python as obsws
from clypi import Command, Positional, arg
from typing_extensions import override
from .errors import SimpleRecorderError
from .styler import highlight
class Start(Command):
"""Start recording."""
filename: Positional[str] = arg(
default="default_name",
help="Name of the recording",
prompt="Enter the name for the recording",
)
host: str = arg(inherited=True)
port: int = arg(inherited=True)
password: str = arg(inherited=True)
@staticmethod
def get_timestamp():
return datetime.now().strftime("%Y-%m-%d %H-%M-%S")
@override
async def run(self):
if not self.filename:
raise SimpleRecorderError("Recording name cannot be empty.")
with obsws.ReqClient(
host=self.host, port=self.port, password=self.password
) as client:
resp = client.get_record_status()
if resp.output_active:
raise SimpleRecorderError("Recording is already active.")
filename = f"{self.filename} {self.get_timestamp()}"
client.set_profile_parameter(
"Output",
"FilenameFormatting",
filename,
)
client.start_record()
print(f"Recording started with filename: {highlight(filename)}")

View File

@ -0,0 +1,26 @@
import obsws_python as obsws
from clypi import Command, arg
from typing_extensions import override
from .errors import SimpleRecorderError
from .styler import highlight
class Stop(Command):
"""Stop recording."""
host: str = arg(inherited=True)
port: int = arg(inherited=True)
password: str = arg(inherited=True)
@override
async def run(self):
with obsws.ReqClient(
host=self.host, port=self.port, password=self.password
) as client:
resp = client.get_record_status()
if not resp.output_active:
raise SimpleRecorderError("Recording is not active.")
client.stop_record()
print(highlight("Recording stopped successfully."))