r/conductofconduct
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Set configuration options for processes and save to toml

Browse source
This commit is contained in:
Ruben van de Ven 2025-03-25 21:51:05 +01:00
parent 9ed078505a
commit 78d3b40c03
9 changed files with 397 additions and 134 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

0
foucault/__init__.py Normal file
View file

BIN
foucault/__pycache__/__init__.cpython-313.pyc Normal file
View file

Binary file not shown.

BIN
foucault/__pycache__/gui.cpython-313.pyc Normal file
View file

Binary file not shown.

331
foucault/gui.py Normal file
View file

@ -0,0 +1,331 @@
"""
Foucault allows you to govern a set of commands, which can be useful for
i.e. an art installation setting, in which different scripts need to work
together. While one can be (re)started independently of the rest.
"""
import asyncio
from collections import defaultdict
import os
import pathlib
import random
from subprocess import Popen
import subprocess
from tempfile import mkstemp
import threading
import time
import tomllib
from typing import Optional
from nicegui import ui
import logging
import tomli_w
logger = logging.getLogger('procescontroller')
numbers = []
processes = []
class Argument():
def __init__(self, name: str, enabled: bool = True, inline: bool = False, boolean: bool = False, options: list[str] = [], selected: list[str] = []):
self.name = name
self.enabled = enabled
self.inline = inline
self.boolean = boolean
self.options = options
self.selected = selected
def as_list(self) -> list[str]:
if not self.enabled:
return []
l = []
if not self.inline:
l.append(self.name)
if not self.boolean:
l += self.selected
return l
def select(self, option: str):
self.selected = [option]
def toggle(self):
self.enabled = not self.enabled
@classmethod
def from_dict(cls, data: dict):
return cls(
data['name'],
data.get('enabled', False),
data.get('inline', False),
data.get('boolean', False),
data.get('options', []),
data.get('selected', []),
)
def to_dict(self) -> dict:
return self.__dict__
class SubprocessController():
def __init__(self, name: str, cmd: list[str], arguments: list[Argument], environment: dict[str,str] = {}, fn: Optional[pathlib.Path] = None):
self.cmd = cmd
self.arguments = arguments
self.name = name
self.filename = fn
# self.tmp_config_fp, self.tmp_config_filename = mkstemp(suffix=".yml", prefix=f"config_{self.name}_")
self.env_overrides = environment
# dict[str, str] = {
# "DISPLAY": ":1"
# }
self.proc=None
self.running_config_state: Optional[str] = None
self.saved_state = self.config_state()
@classmethod
def from_toml(cls, filename: os.PathLike):
path = pathlib.Path(filename)
name = path.stem
args = []
with path.open('rb') as fp:
data = tomllib.load(fp)
print(data)
for arg in data['arguments']:
args.append(Argument.from_dict(arg))
sc = cls(
name,
[data['cmd']] if data['cmd'] is str else data['cmd'],
args,
data.get('environment', {}),
path
)
return sc
def to_dict(self):
return {
"cmd": self.cmd,
"arguments": [a.to_dict() for a in self.arguments],
"environment": self.env_overrides,
}
def update_config(self):
pass
def get_environment(self):
default_env = os.environ.copy()
return default_env | self.env_overrides
def get_environment_strs(self):
return [f"{k}=\"{v}\"" for k, v in self.env_overrides.items()]
def rm_env(self, key):
if key in self.env_overrides:
del self.env_overrides[key]
def add_env(self, key, value):
self.env_overrides[key] = value
def get_arguments(self)-> list[str]:
r = []
for a in self.arguments:
r.extend(a.as_list())
return r
def as_bash_string(self):
return " " . join(self.get_environment_strs() + self.cmd + self.get_arguments())
def run(self):
logger.info(f"Run: {self.as_bash_string()}")
self.running_config_state = self.as_bash_string()
self.proc = Popen(
self.cmd + self.get_arguments(),
env=self.get_environment(),
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL)
def is_stale(self):
"""Config has changed since starting the process"""
return self.is_running() and self.running_config_state != self.as_bash_string()
def quit(self):
if self.is_running():
self.proc.terminate()
def restart(self):
self.quit()
self.run()
def return_code(self) -> Optional[int]:
if self.proc is None:
return None
return self.proc.returncode
def is_running(self):
return self.proc is not None and self.proc.poll() is None
def state(self):
return "::".join(
[self.name, str(self.is_running()), str(self.return_code())]
) + "::" + self.config_state() + (self.running_config_state if self.running_config_state else "")
def config_state(self):
return "::".join(
self.cmd + self.get_arguments() + ["+".join(a.options) for a in self.arguments] + [str(self.filename.stat().st_mtime) if self.filename else ""] +
list(self.env_overrides.values())
)
def save(self) -> str:
# print(self.to_dict())
with self.filename.open('wb') as fp:
tomli_w.dump(self.to_dict(), fp)
self.saved_state = self.config_state()
def unsaved_changes(self):
return self.config_state() != self.saved_state
class Foucault():
def __init__(self):
self.processes: list[SubprocessController] = []
self.uis: list[SubprocessUI] = []
def add_process(self, sc: SubprocessController):
self.processes.append(sc)
self.uis.append(SubprocessUI(sc))
def run_all(self):
for p in self.processes:
p.run()
def stop_all(self):
for p in self.processes:
p.quit()
def restart_all(self):
# when restarting all, make sure all are stopped
# before starting, so we start with a clean slate
self.stop_all()
self.run_all()
def ui(self):
with ui.row():
ui.button('▶', on_click=self.run_all)
ui.button('■', on_click=self.stop_all)
ui.button('↺', on_click=self.restart_all)
with ui.grid(columns="40em 40em 40em"):
for proc_ui in self.uis:
proc_ui.ui()
def run(self):
# t = threading.Thread(target=reloader, daemon=True)
t = threading.Thread(target=self.monitor, daemon=True)
t.start()
#build ui:
self.ui()
ui.page_title("Conduct of conduct")
ui.run(show=False, favicon="🫣")
def monitor(self):
states = defaultdict(lambda: "")
i=0
while True:
i+=1
# if we're only interested in running, we can use os.waitpid(), but we also check config changes
time.sleep(.3)
for proc_ui in self.uis:
state = proc_ui.sc.state()
if state != states[proc_ui.sc]:
proc_ui.ui.refresh()
states[proc_ui.sc] = state
class SubprocessUI:
def __init__(self, sc: SubprocessController):
self.sc = sc
@ui.refreshable_method
def ui(self):
card_class = "bg-teal" if self.sc.is_running() else ("bg-warning" if self.sc.return_code() else "bg-gray")
with ui.card().classes(card_class):
with ui.row(align_items="stretch").classes('w-full'):
if self.sc.return_code():
ui.label(f'⚠').classes('text-h5')
ui.label(f'{self.sc.name}').tooltip(str(self.sc.filename)).classes('text-h5')
ui.space()
with ui.button_group():
ui.button('▶', on_click=self.sc.run)
ui.button('■', on_click=self.sc.quit)
ui.button('↺', on_click=self.sc.restart, color='red' if self.sc.is_stale() else "primary")
ui.label(f'Running: {self.sc.is_running()}')
ui.code(f'{" ".join(self.sc.cmd)}')
ui.code(self.sc.as_bash_string())
ui.separator()
with ui.row().classes('w-full'):
edit = ui.switch(value=self.sc.unsaved_changes())
ui.space()
ui.button('🖫', on_click=self.sc.save, color="red" if self.sc.unsaved_changes() else 'lightgray').classes('text-lg')
with ui.grid().bind_visibility_from(edit, 'value'):
ui.label('Arguments').classes('text-h6')
for i, argument in enumerate(self.sc.arguments):
with ui.card().classes('w-full'):
with ui.row(align_items='center').classes('w-full'):
ui.button("⊗", on_click=lambda i=i: self.sc.arguments.pop(i))
ui.label(f'{argument.name}').classes('text-stone-400' if argument.inline else '')
ui.space()
ui.switch(value=argument.enabled, on_change=lambda i=argument: i.toggle())
if not argument.boolean:
with ui.dropdown_button(f"{len(argument.options)} | " + ' '.join(argument.selected), auto_close=False).classes('normal-case'):
# ui.badge('0', color='red').props('floating')
for o in argument.options:
# TODO)) Remove
ui.item(o +'i', on_click=(lambda a=argument, o=o: a.select(o) ))
with ui.item():
ui.input(placeholder="add new").on('keydown.enter', (lambda e, a=argument: a.options.append(e.sender.value) or print(a.options)))
# ui.label(f'{argument.name}')
# with ui.row():
with ui.card():
with ui.row().classes('w-full'):
name = ui.input(placeholder="argument").classes('w-full')#.on("keydown.enter", )
with ui.row():
enabled = ui.switch('enabled', value=True)
positional = ui.checkbox('positional') #inline
boolean = ui.checkbox('boolean')
ui.button("+", on_click=lambda e, name=name, enabled=enabled, positional=positional, boolean=boolean: self.sc.arguments.append(Argument(
name.value,
enabled.value,
positional.value,
boolean.value
)))
ui.label('Environment variables').classes('text-h6')
with ui.card():
for k, v in self.sc.env_overrides.items():
with ui.row(align_items="center"):
ui.button("⊗", on_click=lambda k=k: self.sc.rm_env(k))
ui.label(f"{k}=\"{v}\"")
with ui.row():
name = ui.input(placeholder="name")#.on("keydown.enter", )
value = ui.input(placeholder="value")#.on("keydown.enter", )
ui.button("+", on_click=lambda e, name=name, enabled=value: self.sc.add_env(name.value, value.value))

133
gui.py
View file

@ -1,133 +0,0 @@
import asyncio
import os
import pathlib
import random
from subprocess import Popen
from tempfile import mkstemp
import threading
import time
from nicegui import ui
numbers = []
processes = []
class SubprocessController():
def __init__(self, name: str, cmd):
self.cmd = cmd
self.name = name
# self.tmp_config_fp, self.tmp_config_filename = mkstemp(suffix=".yml", prefix=f"config_{self.name}_")
self.env_overrides: dict[str, str] = {
"DISPLAY": ":1"
}
self.proc=None
pass
def update_config(self):
pass
def get_environment(self):
default_env = os.environ.copy()
return default_env | self.env_overrides
def run(self):
self.proc = Popen(
self.cmd,
env=self.get_environment(),
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE)
def quit(self):
if self.is_running():
self.proc.terminate()
def is_running(self):
return self.proc is not None and self.proc.poll() is None
# class SubprocessUi():
# def __init__(self, controller: SubprocessController):
# self.controller = controller
# @ui.refreshable_method
# def card(self):
# with ui.card():
# ui.label(self.controller.name)
# ui.label(str(self.controller.is_running()))
class Foucault():
def __init__(self):
self.processes: list[SubprocessController] = []
self.uis: list[SubprocessUI] = []
def add_process(self, sc: SubprocessController):
self.processes.append(sc)
self.uis.append(SubprocessUI(sc))
def ui(self):
with ui.row():
for proc_ui in self.uis:
proc_ui.ui()
def run(self):
i=0
while True:
i+=1
time.sleep(1)
for proc_ui in self.uis:
proc_ui.ui.refresh()
# if i >10:
# print('cancelling')
# sc.quit()
conductofconduct = Foucault()
@ui.refreshable
def number_ui() -> None:
print(conductofconduct.processes)
ui.label("test: " + ','.join([str(process.is_running()) for process in conductofconduct.processes]))
# ui.label(', '.join(str(n) for n in sorted(conductofconduct.processes)))
def add_number() -> None:
numbers.append(random.randint(0, 100))
number_ui.refresh()
def reloader():
"""Test reloading"""
for i in range(200):
print("add")
numbers.append(i)
time.sleep(1)
number_ui.refresh()
class SubprocessUI:
def __init__(self, sc: SubprocessController):
self.sc = sc
@ui.refreshable_method
def ui(self):
with ui.card():
ui.label(f'{self.sc.name}')
ui.label(f'Running: {self.sc.is_running()}')
ui.button('Run', on_click=self.sc.run)
ui.button('Stop', on_click=self.sc.quit)
conductofconduct.add_process(SubprocessController("test1" , ["tail", '-f', "gui.py"]))
conductofconduct.add_process(SubprocessController("test2" , ["tail", '-f', "gui.py"]))
conductofconduct.ui()
# t = threading.Thread(target=reloader, daemon=True)
t = threading.Thread(target=conductofconduct.run, daemon=True)
t.start()
# number_ui()
# ui.button('Add random number', on_click=add_number)
ui.run(show=False)

22
main.py Normal file
View file

@ -0,0 +1,22 @@
from pathlib import Path
from foucault.gui import *
# print(sc)
logging.basicConfig(level=logging.INFO)
conductofconduct = Foucault()
sc = SubprocessController.from_toml(Path('processes/tail.toml'))
sc2 = SubprocessController.from_toml(Path('processes/tail2.toml'))
sc3 = SubprocessController.from_toml(Path('processes/tail.toml'))
conductofconduct.add_process(sc)
conductofconduct.add_process(sc2)
conductofconduct.add_process(sc3)
# conductofconduct.add_process(SubprocessController("test1" , ["tail"], ['-f', "gui.py"]))
# conductofconduct.add_process(SubprocessController("test2" , ["tail"], ['-f', "gui.py"]))
# conductofconduct.add_process(SubprocessController("test3 broken" , ["tail"], ['-f', "nonexistent.py"]))
# conductofconduct.add_process(SubprocessController("system status" , ["uv run status.py"], ['-f', "nonexistent.py"]))
conductofconduct.run()

29
processes/tail.toml Normal file
View file

@ -0,0 +1,29 @@
cmd = [
"tail",
]
[[arguments]]
name = "-f"
enabled = true
inline = false
boolean = true
options = []
selected = []
[[arguments]]
name = "name"
enabled = true
inline = true
boolean = false
options = [
"gui.py",
"nonexistent.py",
"test",
"meer",
]
selected = [
"gui.py",
]
[environment]
":DISPLAY" = "1"

1
pyproject.toml
View file

@ -6,4 +6,5 @@ readme = "README.md"
requires-python = ">=3.13"
dependencies = [
"nicegui>=2.12.1",
"tomli-w>=1.2.0",
]

15
uv.lock
View file

@ -210,10 +210,14 @@ version = "0.1.0"
source = { virtual = "." }
dependencies = [
{ name = "nicegui" },
{ name = "tomli-w" },
]
[package.metadata]
requires-dist = [{ name = "nicegui", specifier = ">=2.12.1" }]
requires-dist = [
{ name = "nicegui", specifier = ">=2.12.1" },
{ name = "tomli-w", specifier = ">=1.2.0" },
]
[[package]]
name = "h11"
@ -646,6 +650,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/a0/4b/528ccf7a982216885a1ff4908e886b8fb5f19862d1962f56a3fce2435a70/starlette-0.46.1-py3-none-any.whl", hash = "sha256:77c74ed9d2720138b25875133f3a2dae6d854af2ec37dceb56aef370c1d8a227", size = 71995 },
]
[[package]]
name = "tomli-w"
version = "1.2.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/19/75/241269d1da26b624c0d5e110e8149093c759b7a286138f4efd61a60e75fe/tomli_w-1.2.0.tar.gz", hash = "sha256:2dd14fac5a47c27be9cd4c976af5a12d87fb1f0b4512f81d69cce3b35ae25021", size = 7184 }
wheels = [
{ url = "https://files.pythonhosted.org/packages/c7/18/c86eb8e0202e32dd3df50d43d7ff9854f8e0603945ff398974c1d91ac1ef/tomli_w-1.2.0-py3-none-any.whl", hash = "sha256:188306098d013b691fcadc011abd66727d3c414c571bb01b1a174ba8c983cf90", size = 6675 },
]
[[package]]
name = "typing-extensions"
version = "4.12.2"