StreamController/src/CLI.py at main · StreamController/StreamController · GitHub

History

515 lines (413 loc) · 25.5 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

334

335

336

337

338

339

340

341

342

343

344

345

346

347

348

349

350

351

352

353

354

355

356

357

358

359

360

361

362

363

364

365

366

367

368

369

370

371

372

373

374

375

376

377

378

379

380

381

382

383

384

385

386

387

388

389

390

391

392

393

394

395

396

397

398

399

400

401

402

403

404

405

406

407

408

409

410

411

412

413

414

415

416

417

418

419

420

421

422

423

424

425

426

427

428

429

430

431

432

433

434

435

436

437

438

439

440

441

442

443

444

445

446

447

448

449

450

451

452

453

454

455

456

457

458

459

460

461

462

463

464

465

466

467

468

469

470

471

472

473

474

475

476

477

478

479

480

481

482

483

484

485

486

487

488

489

490

491

492

493

494

495

496

497

498

499

500

501

502

503

504

505

506

507

508

509

510

511

512

513

514

515

"""

Author: Core447

Year: 2026

This program is free software: you can redistribute it and/or modify

it under the terms of the GNU General Public License as published by

the Free Software Foundation, either version 3 of the License, or

any later version.

This programm comes with ABSOLUTELY NO WARRANTY!

You should have received a copy of the GNU General Public License

along with this program. If not, see <https://www.gnu.org/licenses/>.

"""

The command line front end, and the fast path that answers it from an already

running instance.

Importing main.py costs ~2.4 s before a single line of CLI code runs: it pulls

in GTK/libadwaita, the deck, plugin and store backends, and - through globals ->

HelperMethods - matplotlib, cairosvg and requests. That is the entire cost of

something like --set-icon; the DBus round trip to the running instance is a few

milliseconds. So run_against_running_instance() is called at the very top of

main.py, before any of those imports, and dispatches straight to the DBus API in

src/api.py using nothing but argparse and dbus-python (~60 ms all in).

Anything this module does not handle - no instance running, --close-running,

macOS, or a flag with no API equivalent such as --list-actions - returns False,

and main.py boots as usual and handles the command itself (editing the page json

directly, buffering page/state changes for the instance it is about to start, or

just launching the app).

The argparser lives here rather than in globals.py for the same reason: the fast

path has to parse the command line without importing globals. globals.py still

owns the `argparser` object itself, so `gl.argparser` keeps working everywhere.

dbus-python is used for blocking method calls only and is never attached to a

main loop - see tests/test_dbus_mainloop.py.

"""

import argparse

import json

import os

import sys

DBUS_NAME = "com.core447.StreamController"

DBUS_PATH = "/com/core447/StreamController"

MAX_REASONABLE_X = 10

MAX_REASONABLE_Y = 10

VALID_EMULATE_INPUT_EVENTS = ("press", "long-press")

def build_argparser() -> argparse.ArgumentParser:

argparser = argparse.ArgumentParser()

argparser.add_argument("-b", help="Open in background", action="store_true")

argparser.add_argument("--daemon-only", help="Run without creating the main window until reopened", action="store_true")

argparser.add_argument("--devel", help="Developer mode (disables auto update)", action="store_true")

argparser.add_argument("--skip-load-hardware-decks", help="Skips initilization/use of hardware decks", action="store_true")

argparser.add_argument("--close-running", help="Close running", action="store_true")

argparser.add_argument("--data", help="Data path", type=str)

argparser.add_argument("--json", help="Print machine-readable JSON instead of human-readable text for read/list commands", action="store_true")

argparser.add_argument("--change-page", action="append", nargs=2, help="Change the page for a device", metavar=("SERIAL_NUMBER", "PAGE_NAME"))

argparser.add_argument("--list-devices", help="List all connected StreamDeck devices and their properties", action="store_true")

argparser.add_argument("--list-pages", help="List all available pages", action="store_true")

argparser.add_argument("--change-state", action="append", nargs=4,

help="Change the state of a StreamDeck item. Format: SERIAL PAGE COORDS STATE\n"

" SERIAL: Device serial number (e.g., CL123456789)\n"

" PAGE: Page name (e.g., Main, Soundboard) \n"

" COORDS: Position as x,y (e.g., 0,0 for top-left)\n"

" STATE: State number to change to (0, 1, 2, etc.)\n"

"Example: --change-state CL123456789 Main 0,0 1",

metavar=("SERIAL", "PAGE", "COORDS", "STATE"))

argparser.add_argument("--emulate-input", action="append", nargs=4,

help="Trigger a button action the same way a physical press would. Format: EVENT SERIAL PAGE COORDS\n"

" EVENT: press or long-press\n"

" SERIAL: Device serial number (e.g., CL123456789)\n"

" PAGE: Page name (e.g., Main, Soundboard)\n"

" COORDS: Position as x,y (e.g., 0,0 for top-left)\n"

"Example: --emulate-input press CL123456789 Main 0,0",

metavar=("EVENT", "SERIAL", "PAGE", "COORDS"))

argparser.add_argument("--list-actions", action="append", nargs=3,

help="List the actions configured on a StreamDeck item. Format: PAGE COORDS STATE\n"

" PAGE: Page name (e.g., Main, Soundboard)\n"

" COORDS: Position as x,y (e.g., 0,0 for top-left)\n"

" STATE: State number to inspect (0, 1, 2, etc.)\n"

"Example: --list-actions Main 0,0 0",

metavar=("PAGE", "COORDS", "STATE"))

# Page management

argparser.add_argument("--create-page", action="append", nargs=1, help="Create a new, empty page", metavar="NAME")

argparser.add_argument("--delete-page", action="append", nargs=1, help="Delete a page", metavar="NAME")

argparser.add_argument("--rename-page", action="append", nargs=2, help="Rename a page", metavar=("NAME", "NEW_NAME"))

argparser.add_argument("--duplicate-page", action="append", nargs=2, help="Duplicate a page under a new name", metavar=("NAME", "NEW_NAME"))

argparser.add_argument("--export-page", action="append", nargs=2, help="Export a single page (with its assets) to a .scpage file", metavar=("NAME", "DEST_PATH"))

argparser.add_argument("--export-all", action="append", nargs=1, help="Export all pages (with their assets) to a .zip file, same as Page Manager's 'Export All'", metavar="DEST_PATH")

# State management

argparser.add_argument("--add-state", action="append", nargs=2, help="Add a new state to a StreamDeck item", metavar=("PAGE", "COORDS"))

argparser.add_argument("--remove-state", action="append", nargs=3, help="Remove a state from a StreamDeck item", metavar=("PAGE", "COORDS", "STATE"))

# Labels

argparser.add_argument("--get-label", action="append", nargs=5,

help="Get a label property. PROPERTY: text, font-family, font-size, font-style, font-weight, color, outline-width, outline-color, align",

metavar=("PAGE", "COORDS", "STATE", "POSITION", "PROPERTY"))

argparser.add_argument("--set-label", action="append", nargs=6,

help="Set a label property. PROPERTY: text, font-family, font-size, font-style, font-weight, color, outline-width, outline-color, align. "

"COLOR values use 'R,G,B' or 'R,G,B,A' (0-255)",

metavar=("PAGE", "COORDS", "STATE", "POSITION", "PROPERTY", "VALUE"))

# Background

argparser.add_argument("--get-background-color", action="append", nargs=3, help="Get the background color of a state", metavar=("PAGE", "COORDS", "STATE"))

argparser.add_argument("--set-background-color", action="append", nargs=4,

help="Set the background color of a state. COLOR: 'R,G,B' or 'R,G,B,A' (0-255)",

metavar=("PAGE", "COORDS", "STATE", "COLOR"))

# Icon / media

argparser.add_argument("--get-icon", action="append", nargs=3, help="Get the icon (media path) of a state", metavar=("PAGE", "COORDS", "STATE"))

argparser.add_argument("--set-icon", action="append", nargs=4,

help="Set the icon of a state from an image/gif/video file. The file is imported through the asset manager",

metavar=("PAGE", "COORDS", "STATE", "PATH"))

argparser.add_argument("--get-icon-layout", action="append", nargs=4,

help="Get an icon layout property. PROPERTY: size, valign, halign, fill-mode",

metavar=("PAGE", "COORDS", "STATE", "PROPERTY"))

argparser.add_argument("--set-icon-layout", action="append", nargs=5,

help="Set an icon layout property. PROPERTY: size, valign, halign, fill-mode",

metavar=("PAGE", "COORDS", "STATE", "PROPERTY", "VALUE"))

# Brightness

argparser.add_argument("--get-brightness", action="append", nargs=1, help="Get the brightness of a device (0-100)", metavar="SERIAL")

argparser.add_argument("--set-brightness", action="append", nargs=2, help="Set the brightness of a device (0-100)", metavar=("SERIAL", "VALUE"))

# Screensaver / sleep

argparser.add_argument("--sleep", action="append", nargs=1, help="Put a device to sleep (show the screensaver now). Requires a running instance", metavar="SERIAL")

argparser.add_argument("--wake", action="append", nargs=1, help="Wake a device from sleep. Requires a running instance", metavar="SERIAL")

argparser.add_argument("app_args", nargs="*")

return argparser

# ── Validation ───────────────────────────────────────────────────────

def validate_state_change_args(args):

"""

Validate CLI arguments for --change-state

Returns (is_valid, error_message)

"""

if not args.change_state:

return True, None

for i, (serial_number, page_name, coords, state_number) in enumerate(args.change_state):

# Validate serial number format (basic check)

if not serial_number or not isinstance(serial_number, str):

return False, f"Invalid serial number in argument {i+1}: '{serial_number}'"

# Validate page name

if not page_name or not isinstance(page_name, str):

return False, f"Invalid page name in argument {i+1}: '{page_name}'"

# Validate coordinate format

if not coords or not isinstance(coords, str):

return False, f"Invalid coordinates in argument {i+1}: '{coords}'"

if ',' not in coords:

return False, f"Invalid coordinate format in argument {i+1}: '{coords}'. Expected format: 'x,y' (e.g., '0,0')"

try:

x, y = map(int, coords.split(','))

if x < 0 or y < 0:

return False, f"Coordinates must be non-negative in argument {i+1}: '{coords}'"

if x > MAX_REASONABLE_X or y > MAX_REASONABLE_Y: # Reasonable bounds check

return False, f"Coordinates seem too large in argument {i+1}: '{coords}'. Most StreamDecks have coordinates 0-4"

except ValueError:

return False, f"Invalid coordinate format in argument {i+1}: '{coords}'. Expected integers like '0,0'"

# Validate state number

try:

state_num = int(state_number)

if state_num < 0:

return False, f"State number must be non-negative in argument {i+1}: '{state_number}'"

if state_num > 20: # Reasonable bounds check

return False, f"State number seems too large in argument {i+1}: '{state_number}'. Most items have 1-5 states"

except ValueError:

return False, f"Invalid state number in argument {i+1}: '{state_number}'. Must be an integer"

return True, None

def validate_emulate_input_args(args):

"""

Validate CLI arguments for --emulate-input

Returns (is_valid, error_message)

"""

if not args.emulate_input:

return True, None

for i, (event, serial_number, page_name, coords) in enumerate(args.emulate_input):

if event not in VALID_EMULATE_INPUT_EVENTS:

return False, f"Invalid event in argument {i+1}: '{event}'. Must be one of: {', '.join(VALID_EMULATE_INPUT_EVENTS)}"

if not serial_number or not isinstance(serial_number, str):

return False, f"Invalid serial number in argument {i+1}: '{serial_number}'"

if not page_name or not isinstance(page_name, str):

return False, f"Invalid page name in argument {i+1}: '{page_name}'"

if not coords or not isinstance(coords, str) or ',' not in coords:

return False, f"Invalid coordinate format in argument {i+1}: '{coords}'. Expected format: 'x,y' (e.g., '0,0')"

try:

x, y = map(int, coords.split(','))

if x < 0 or y < 0:

return False, f"Coordinates must be non-negative in argument {i+1}: '{coords}'"

if x > MAX_REASONABLE_X or y > MAX_REASONABLE_Y:

return False, f"Coordinates seem too large in argument {i+1}: '{coords}'. Most StreamDecks have coordinates 0-4"

except ValueError:

return False, f"Invalid coordinate format in argument {i+1}: '{coords}'. Expected integers like '0,0'"

return True, None

def print_state_change_usage() -> None:

print("\nUsage examples:", file=sys.stderr)

print(" --change-state CL123456789 Main 0,0 1", file=sys.stderr)

print(" --change-state CL123456789 Soundboard 2,1 0", file=sys.stderr)

print("\nParameters:", file=sys.stderr)

print(" SERIAL_NUMBER: Device serial (e.g., CL123456789)", file=sys.stderr)

print(" PAGE_NAME: Page name (e.g., Main, Soundboard)", file=sys.stderr)

print(" COORDINATES: Position as x,y (e.g., 0,0 for top-left)", file=sys.stderr)

print(" STATE_NUMBER: State to change to (e.g., 0, 1, 2)", file=sys.stderr)

def print_emulate_input_usage() -> None:

print("\nUsage examples:", file=sys.stderr)

print(" --emulate-input press CL123456789 Main 0,0", file=sys.stderr)

print(" --emulate-input long-press CL123456789 Soundboard 2,1", file=sys.stderr)

print("\nParameters:", file=sys.stderr)

print(f" EVENT: {' or '.join(VALID_EMULATE_INPUT_EVENTS)}", file=sys.stderr)

print(" SERIAL_NUMBER: Device serial (e.g., CL123456789)", file=sys.stderr)

print(" PAGE_NAME: Page name (e.g., Main, Soundboard)", file=sys.stderr)

print(" COORDINATES: Position as x,y (e.g., 0,0 for top-left)", file=sys.stderr)

# ── DBus ─────────────────────────────────────────────────────────────

def get_dbus_api():

"""

One-shot blocking connection to a running instance's rich API

(com.core447.StreamController, see src/api.py), or None if no instance

answers. Never attached to a mainloop - see tests/test_dbus_mainloop.py.

"""

if sys.platform == "darwin":

return None

try:

import dbus

session_bus = dbus.SessionBus()

obj = session_bus.get_object(DBUS_NAME, DBUS_PATH)

return dbus.Interface(obj, DBUS_NAME)

except Exception:

# No session bus, no instance, or a malformed bus address - the caller

# falls back to doing the work itself

return None

# ── Fast path ────────────────────────────────────────────────────────

# Every flag that maps onto a method of the DBus API. Matched against argv

# before anything is parsed or imported, so a normal app start pays nothing

# for this.

FAST_PATH_FLAGS = frozenset((

"--change-page", "--change-state", "--emulate-input",

"--create-page", "--delete-page", "--rename-page", "--duplicate-page",

"--export-page", "--export-all",

"--add-state", "--remove-state",

"--get-label", "--set-label",

"--get-background-color", "--set-background-color",

"--get-icon", "--set-icon", "--get-icon-layout", "--set-icon-layout",

"--get-brightness", "--set-brightness",

"--sleep", "--wake",

))

def _has_fast_path_flag(argv) -> bool:

return any(arg.split("=", 1)[0] in FAST_PATH_FLAGS for arg in argv)

def _error_message(e) -> str:

"""The human readable half of a DBusException raised by src/api.py.

api.py raises dasbus' DBusError(error_name, message), and dasbus puts

str() of that whole tuple on the wire as the message, so what arrives here

is `('com.core447...PageNotFound', "Page 'x' not found")`. Unwrap it so the

CLI prints the message the API actually wrote."""

message = e.get_dbus_message()

if message.startswith("(") and message.endswith(")"):

try:

import ast

parts = ast.literal_eval(message)

if isinstance(parts, tuple) and parts:

return str(parts[-1])

except (ValueError, SyntaxError):

pass

return message

def print_get_results(results: list[dict], as_json: bool) -> None:

"""Shared by the fast path and main.py's on-disk reads so both print the

same thing. `value` is always a string, as the DBus API returns it."""

if as_json:

print(json.dumps(results, indent=2))

return

for entry in results:

label = f"{entry.get('page')} {entry.get('coords')} state={entry.get('state')}"

if "position" in entry:

label += f" {entry['position']}.{entry['property']}"

elif "property" in entry:

label += f" {entry['property']}"

if "error" in entry:

print(f"{label}: Error: {entry['error']}")

else:

print(f"{label}: {entry['value']}")

def run_against_running_instance() -> bool:

"""Handle the command by calling a running instance over DBus.

Returns True if the command was fully handled (the caller should exit),

False if main.py has to take over."""

argv = sys.argv[1:]

if sys.platform == "darwin" or not _has_fast_path_flag(argv):

return False

if "--close-running" in argv:

# The request is meant for the instance that is about to replace the

# running one, so main.py buffers it and applies it after startup

return False

api = get_dbus_api()

if api is None:

return False

args = build_argparser().parse_args()

is_valid, error_msg = validate_state_change_args(args)

if not is_valid:

print(f"Error: {error_msg}", file=sys.stderr)

print_state_change_usage()

sys.exit(1)

is_valid, error_msg = validate_emulate_input_args(args)

if not is_valid:

print(f"Error: {error_msg}", file=sys.stderr)

print_emulate_input_usage()

sys.exit(1)

if _dispatch(args, api):

sys.exit(1)

return True

def _dispatch(args, api) -> bool:

"""Runs every requested command against `api`. Returns True if any failed."""

import dbus

had_error = False

def call(fn):

nonlocal had_error

try:

return fn()

except dbus.exceptions.DBusException as e:

print(f"Error: {_error_message(e)}", file=sys.stderr)

had_error = True

return None

def read(entry: dict, fn, results: list[dict]) -> None:

"""A read reports its failure inside the result row rather than on

stderr, so --json output still has one row per requested read - same

shape main.py produces when it reads the page json itself."""

nonlocal had_error

try:

entry["value"] = str(fn())

except dbus.exceptions.DBusException as e:

entry["error"] = _error_message(e)

had_error = True

results.append(entry)

# ── Runtime page / state / input control ────────────────────────

for serial, page_name in args.change_page or []:

call(lambda serial=serial, page_name=page_name: api.ChangePage(serial, page_name))

for serial, page_name, coords, state in args.change_state or []:

call(lambda serial=serial, page_name=page_name, coords=coords, state=state:

api.ChangeState(serial, page_name, coords, int(state)))

for event, serial, page_name, coords in args.emulate_input or []:

call(lambda event=event, serial=serial, page_name=page_name, coords=coords:

api.EmulateInput(event, serial, page_name, coords))

# ── Page management ─────────────────────────────────────────────

for (name,) in args.create_page or []:

call(lambda name=name: api.AddPage(name, "{}"))

for (name,) in args.delete_page or []:

call(lambda name=name: api.RemovePage(name))

for name, new_name in args.rename_page or []:

call(lambda name=name, new_name=new_name: api.RenamePage(name, new_name))

for name, new_name in args.duplicate_page or []:

call(lambda name=name, new_name=new_name: api.DuplicatePage(name, new_name))

for name, dest_path in args.export_page or []:

dest_path = os.path.abspath(dest_path)

if call(lambda name=name, dest_path=dest_path: api.ExportPage(name, dest_path)) is not None:

print(f"Exported '{name}' to {dest_path}")

for (dest_path,) in args.export_all or []:

dest_path = os.path.abspath(dest_path)

if call(lambda dest_path=dest_path: api.ExportAll(dest_path)) is not None:

print(f"Exported all pages to {dest_path}")

# ── States ───────────────────────────────────────────────────────

for name, coords in args.add_state or []:

new_index = call(lambda name=name, coords=coords: api.AddState(name, coords))

if new_index is not None:

print(f"Added state {new_index} to {name} {coords}")

for name, coords, state in args.remove_state or []:

call(lambda name=name, coords=coords, state=state: api.RemoveState(name, coords, int(state)))

# ── Labels ───────────────────────────────────────────────────────

results = []

for name, coords, state, position, prop in args.get_label or []:

read({"page": name, "coords": coords, "state": state, "position": position, "property": prop},

lambda name=name, coords=coords, state=state, position=position, prop=prop:

api.GetLabel(name, coords, int(state), position, prop), results)

if results:

print_get_results(results, args.json)

for name, coords, state, position, prop, value in args.set_label or []:

call(lambda name=name, coords=coords, state=state, position=position, prop=prop, value=value:

api.SetLabel(name, coords, int(state), position, prop, value))

# ── Background color ────────────────────────────────────────────

results = []

for name, coords, state in args.get_background_color or []:

read({"page": name, "coords": coords, "state": state},

lambda name=name, coords=coords, state=state: api.GetBackgroundColor(name, coords, int(state)), results)

if results:

print_get_results(results, args.json)

for name, coords, state, color in args.set_background_color or []:

call(lambda name=name, coords=coords, state=state, color=color:

api.SetBackgroundColor(name, coords, int(state), color))

# ── Icon / media ─────────────────────────────────────────────────

results = []

for name, coords, state in args.get_icon or []:

read({"page": name, "coords": coords, "state": state},

lambda name=name, coords=coords, state=state: api.GetIcon(name, coords, int(state)), results)

if results:

print_get_results(results, args.json)

for name, coords, state, path in args.set_icon or []:

path = os.path.abspath(path)

internal_path = call(lambda name=name, coords=coords, state=state, path=path:

api.SetIcon(name, coords, int(state), path))

if internal_path is not None:

print(f"Set icon for {name} {coords} state {state}: {internal_path}")

results = []

for name, coords, state, prop in args.get_icon_layout or []:

read({"page": name, "coords": coords, "state": state, "property": prop},

lambda name=name, coords=coords, state=state, prop=prop:

api.GetIconLayout(name, coords, int(state), prop), results)

if results:

print_get_results(results, args.json)

for name, coords, state, prop, value in args.set_icon_layout or []:

call(lambda name=name, coords=coords, state=state, prop=prop, value=value:

api.SetIconLayout(name, coords, int(state), prop, value))

# ── Brightness ───────────────────────────────────────────────────

for (serial,) in args.get_brightness or []:

value = call(lambda serial=serial: api.GetBrightness(serial))

if value is not None:

print(int(value))

for serial, value in args.set_brightness or []:

try:

value_int = int(value)

except ValueError:

print(f"Error: Invalid brightness '{value}'. Must be an integer 0-100", file=sys.stderr)

had_error = True

continue

call(lambda serial=serial, value_int=value_int: api.SetBrightness(serial, value_int))

# ── Sleep / wake ─────────────────────────────────────────────────

for (serial,) in args.sleep or []:

call(lambda serial=serial: api.Sleep(serial))

for (serial,) in args.wake or []:

call(lambda serial=serial: api.Wake(serial))

return had_error

StreamController/src/CLI.py at main · StreamController/StreamController · GitHub

Uh oh!

Expand file tree

Breadcrumbs

CLI.py

Latest commit

History

Breadcrumbs

CLI.py

File metadata and controls