# # (c) 2025 Simon Funk ; https://sifter.org/~simon # from Procs import Process from TrioUtils import Queue, trio, sleep, pool, cancel, sub_pool, move_on_after, spawn from Dyn.Dyn import DynLocal, DynValue, Function from Dyn.DynBridge import DynBridge import termuxgui as tg from math import inf # Size options other than dp: SHRINK_WRAP = tg.View.WRAP_CONTENT class GobBase(object): def __init__(self, root, parent, ob=None): self.root = root # The top level Gob in this hierarchy, which will be a GobRoot (wraps termux Activity) self.parent = parent # Our immediate parent, which will be None if root = self. self.ob = ob # Termux object, unless we're a root. async def dialog(self): "Creates a new popup window in the context of this Gob and returns the corresponding GobRoot." return await Root(self.root.tmx, dialog=True) def dyn_var(self, path): "Returns the Dyn variable at the given path." return self.root.tmx.bridge.find((path, 'var')) def dyn_call(self, path): "Returns the Dyn Function at the given path." return self.root.tmx.bridge.find((path, 'call')) class GobHolder: # Mixin for Gobs that have children # # These are all methods to create children of this holder: # def text_view(self, text, **kargs): return Text(parent=self, text=text, **kargs) def button(self, text, **kargs): return Button(parent=self, text=text, **kargs) def scroller(self, **kargs): return Scroller(parent=self, **kargs) def column(self, **kargs): return ListGob(parent=self, vertical=True, **kargs) def row(self, **kargs): return ListGob(parent=self, vertical=False, **kargs) def space(self, **kargs): return Space(parent=self, **kargs) if False: # These are pretty clunky compared to SelectorButtons... def spinner(self, options=None, **kargs): return Spinner(parent=self, options=options, **kargs) def pager(self): return Pager(parent=self) def function(self, text, func, options, **kargs): # Button that brings up a selector of options... return SelectorButton(parent=self, text=text, func=func, options=options, **kargs) def dyn_value(self, path, label, fmt=None, **kargs): # Should work fine for ints too. """Path is the path to a dynamic value, which will be displayed on the button with the given label. If the value info has options or they are provided in kargs, the user will be able to select from those and change the value. """ var = self.dyn_var(path) return SelectorButton(parent=self, text=var, fmt=fmt or labler(f'{label}\n'), info=self.dyn_var(f"{path}/info"), disabler=var.offline, **kargs) def dyn_float(self, path, label, units='', bands=None, **kargs): """Just like a dyn_value, but formatted as a float. If bands is provided, it should be a sorted tuple of four values, for where the color should transition from red to yellow to green to yellow to red. (The green->red transition through yellow will be gradual through the band, so make it wide.) Use -inf, -inf, or inf, inf for the first or last two to have no min or max green value. [Technically the way this is implemented, if bands is a list it may be updated externally and future value changes will be reflected in the new scheme.] """ if bands: def colorer(val): amp = 200/255 # Matches current named color scheme v0, v1, v2, v3 = bands if val < v0 or val > v3: return (amp, 0, 0) if v1 <= val <= v2: return (0, amp, 0) if val >= v2: v0 = v3 v1 = v2 frac = (val-v1)/(v0-v1) return (amp*frac, amp*(1-frac), 0) else: colorer=None return self.dyn_value(path, label, fmt=labler(f'{label}\n', f'{{:.1f}}{units}'), colorer=colorer, **kargs) def dyn_function(self, path, label=None, label_path=None, **kargs): """Path is the path to a Dyn Function which will be called when the button is pressed, with args as selected by the user from options (either from info, or provided here). The button will be labeled either with static label or the Dyn label at label_path, or both. """ fmt = None if label_path is not None: text = self.dyn_var(label_path) if label is not None: fmt = labler(f'{label}\n') else: text = label func = self.dyn_call(path) info = self.dyn_var(f"{path}/info") return SelectorButton(parent=self, text=text, func=func.try_call, fmt=fmt, info=info, disabler=func.offline, **kargs) async def Root(tmx, dialog=False): if dialog: def error_catcher(e): tmx.log("DIALOG EXCEPTION: {e}") else: error_catcher = None return GobRoot(tmx, pool=await sub_pool(tmx.pool, error_catcher=error_catcher), dialog=dialog) class GobRoot(GobBase, GobHolder): def __init__(self, tmx, pool, dialog=False): GobBase.__init__(self, self, None) self.tmx = tmx self.activity = tg.Activity(tmx.txc, dialog=dialog) self.active = DynLocal(False) # This tracks whether the activity is visible, so we can do less when it's not. self.event_handlers = {} # Maps termux object id to event handling function. self.pool = pool # Component tasks will be created into this trio task pool, and this pool will be canceled when this activity exits! tmx.set_handler(self.activity.aid, self.handle_event) async def ready(self): "Returns when this Activity is visible." if self.active.value: return async for val in self.active: if val: return def set_handler(self, oid, func): """Call this to start dispatching events for object oid to func. """ if func is None: if oid in self.event_handlers: del self.event_handlers[oid] else: self.event_handlers[oid] = func def handle_event(self, ev): if ev.type == 'start': self.active.set_value(True) elif ev.type == 'stop': self.active.set_value(False) elif 'id' in ev.value: oid = ev.value['id'] if oid in self.event_handlers: self.event_handlers[oid](ev) else: self.log(f"{ev.type}: UNHANDLED: {ev.value}") elif ev.type in ('create', 'pause', 'resume', 'UserLeaveHint'): pass elif ev.type == 'destroy': #self.log(f"Canceling sub-pool for activity {self.activity.aid}") self.tmx.set_handler(self.activity.aid, None) # This is key to eventual garbage collection of the whole structure. cancel(self.pool) # And this too... else: self.log(f"{ev.type}: {ev.value}") def finish(self): self.activity.finish() def spawn(self, awaitable): spawn(self.pool, awaitable) def safe_spawn(self, awaitable, msg='Safe Spawn'): "Kick off a task in our root pool and safely log any exceptions." self.spawn(self.safety_wrap(awaitable, msg)) async def safety_wrap(self, awaitable, msg='Safety Wrap'): try: await awaitable except Exception as e: self.log(f"[{msg}] Caught {e}") def log(self, msg): self.tmx.log(msg) class Gob(GobBase): # GUI Object def __init__(self, parent, ob, color=None, height=None, weight=0, disabler=None): """If provided, disabler is a boolean-valued DynValue which is True/False depending on whether this Gob should be disabled. """ GobBase.__init__(self, parent.root, parent, ob) self.color = None # Cached, to avoid re-setting to same, and to facilitate enable/disable coloring # For this class, .disabled just reflects in a half-transparent color when disabled. # Subclasses may use it for other things (like ignoring button presses). if disabler: self.disabler = disabler self.disabled = disabler.value != False # None -> True by default. self.root.spawn(self.track_disabled()) else: self.disabler = None self.disabled = False if ob is not None: if color is not None or self.disabled: self.set_color(color) if height is not None: self.set_height(height) if weight is not None: self.set_weight(weight) def set_disabled(self, disabled): if disabled != self.disabled: self.disabled = disabled self.update_color() def set_color(self, color:int|float|tuple|str): if color != self.color: self.color = color self.update_color() def set_height(self, height:int|str): self.ob.setheight(height) def set_visible(self, visible:bool): self.ob.setvisibility(2 if visible else 0) def set_weight(self, weight=0): self.ob.setlinearlayoutparams(weight=weight) def parse_color(self, color, transparency=0): """Returns color as a [opacity, R, G, B] 4-byte integer. """ if isinstance(color, (str, type(None))): color = { None: ( 90, 90, 92), # The termux default color (on my phone, anyway). 'red': (200, 0, 0), 'green': ( 0, 200, 0), 'blue': ( 0, 0, 200), 'yellow': (150, 150, 0), 'cyan': ( 0, 200, 200), 'magenta': (200, 0, 200), }[color and color.lower()] if isinstance(color, (int, float)): color = (color, color, color) r, g, b = color if isinstance(r, float): r = int(r*255.99) if isinstance(g, float): g = int(g*255.99) if isinstance(b, float): b = int(b*255.99) if transparency: o = int((1-transparency)*255.99) # Opacity else: o = 255 o = min(255, max(0, o)) r = min(255, max(0, r)) g = min(255, max(0, g)) b = min(255, max(0, b)) return (o<<24) | (r<<16) | (g<<8) | b def update_color(self): self.ob.setbackgroundcolor(self.parse_color(self.color, 0.6 if self.disabled else 0)) async def track_disabled(self): async for val in self.disabler: self.set_disabled(val) await self.root.ready() # Returns immediately unless the app is backgrounded. def log(self, msg): self.root.log(msg) class Space(Gob): def __init__(self, parent, **kargs): Gob.__init__(self, parent, tg.TextView(activity=parent.root.activity, parent=parent.ob, text=''), **kargs) if False: # These are pretty clunky compared to SelectorButtons... class Spinner(Gob): def __init__(self, parent, options=None, on_select=None, **kargs): Gob.__init__(self, parent, tg.Spinner(activity=parent.root.activity, parent=parent.ob), **kargs) self.on_select = on_select self.root.set_handler(self.ob.id, self.handle_event) if options is not None: self.set_options(options) else: self.options = [] def handle_event(self, ev): if ev.type == 'itemselected': if self.on_select is None: self.log(f"SPINNER-SELECT: {ev.value}") else: self.on_select(ev.value['selected']) else: self.log(f"SPINNER {ev.type}: {ev.value}") def set_options(self, options:list[str]): self.options = tuple(options) # Read-only copy self.ob.setlist(options) class Text(Gob): def __init__(self, parent, text:str|DynValue, fmt=None, info:DynValue=None, option_names=None, colorer=None, tg_class=tg.TextView, **kargs): """If text is a DynValue, it will update dynamically (and be converted to str if it isn't already). If fmt is provided, fmt(val) will be used instead of str. If info is provided, it should be the /info for the text Dyn, for units, etc. option_names is an optional dict which a dyn text will be mapped through before display. This overrides any found in info. It is applied before fmt. If provided, colorer(val) returns the color to set the object to based on the current (raw) value. """ self.fmt = fmt self.option_names = option_names self.option_names_locked = bool(option_names) self.colorer = colorer if isinstance(text, DynValue): self.dyn_val = text parent.root.spawn(self.track_value()) text = self.format_value(text.value) else: self.dyn_val = None if info is not None: self.dyn_info = info parent.root.spawn(self.track_info()) else: self.dyn_info = None Gob.__init__(self, parent, tg_class(activity=parent.root.activity, text=text, parent=parent.ob), **kargs) def format_value(self, val): if self.option_names: val = self.option_names.get(val, val) if self.fmt is None: if val is None: return '' return str(val) return self.fmt(val) def set_text(self, val): "Accepts any python value, though None prints as empty with the default formatter." if self.colorer is not None: self.set_color(self.colorer(val)) text = self.format_value(val) #self.log(f"ST: {self.ob.id}->{text!r}") self.ob.settext(text) async def track_value(self): async for val in self.dyn_val: self.set_text(val) await self.root.ready() # Returns immediately unless the app is backgrounded. async def track_info(self): async for val in self.dyn_info: if val and isinstance(val, dict): self.new_info(val) await self.root.ready() def new_info(self, info): # Subclass can extend ; info is assured to be a dict here. if 'option_names' in info and not self.option_names_locked: option_names = info['option_names'] if isinstance(option_names, dict): self.option_names = option_names #self.log(f"INFO: {info}") # Use, e.g., Text(..., fmt=labler("Foo: ", '{.2f}')) def labler(label, fmt=None): return lambda v: label + ('N/A' if v is None else str(v) if fmt is None else fmt.format(v)) class Button(Text): def __init__(self, parent, text:str|DynValue, on_click=None, info:DynValue=None, **kargs): """on_click() is called when the button is pressed. """ Text.__init__(self, parent=parent, text=text, info=info, tg_class=tg.Button, **kargs) self.on_click = on_click self.root.set_handler(self.ob.id, self.handle_event) def handle_event(self, ev): if ev.type == 'click': if self.on_click is not None: self.on_click() else: self.log(f"BUTTON-CLICK: {ev.value}") else: self.log(f"BUTTON {ev.type}: {ev.value}") class SelectorButton(Button): def __init__(self, parent, text:str|DynValue, func=None, options=None, **kargs): """A Button with a built in handler. Brings up a list of options for the user to choose from, and sets the DynValue accordingly. options, by default a list of values, optionally overrides any options found in info. If (name, val) tuples, name will be displayed while val is used. option_names, if provided, maps options to more human friendly names. As with options, this overrides any found in info. If provided, func(val) will be called instead of text.try_set_value(val). In that case, it is valid to use a static str for text; otherwise text must be a DynValue. (Note this is distinct from on_click with a regular Button because a selected parameter value is passed.) NOTE if there is exactly one option, the effect is immediate (no options dialog since there's only one choice). """ Button.__init__(self, parent, text, on_click=lambda: self.root.safe_spawn(self.dyn_on_click()), **kargs) self.options = options # This will throw an Exception if func is None and text is a plain str: self.func = func or self.dyn_val.try_set_value async def dyn_on_click(self): """Click handler for dynamic values with info dicts. Here we pop up a menu of options for the user to select from. """ options = self.options if self.dyn_info is not None: info = self.dyn_info.value if isinstance(info, dict): if info.get('read_only', False): self.log("Ignoring tap on read-only button.") return options = options or info.get('options') # Note self.option_names is pre-cached and tracked by Text superclass. if options: if len(options) == 1: # If there's only one choice, just apply it now: opt = options[0] if isinstance(opt, tuple): opt_name, opt_val = opt self.call_func(opt_val) else: self.call_func(opt) else: # If there's more than one choice, create a dialog of options: root = (await self.dialog()).scroller().column() # Scroller is needed for long lists... for opt in options: if isinstance(opt, tuple): opt_name, opt_val = opt else: if self.option_names: opt_name = self.option_names.get(opt, opt) else: opt_name = opt opt_val = opt def click(val=opt_val): # This captures the current value of opt; otherwise they're all the last value of opt! self.call_func(val) root.root.finish() root.button(opt_name, on_click=click) return self.log("Could allow raw editing of value here?") def call_func(self, val): try: self.func(val) except Exception as e: self.log(f"SB.func({val}): {e}") class Scroller(Gob, GobHolder): def __init__(self, parent, **kargs): Gob.__init__(self, parent, tg.NestedScrollView(activity=parent.root.activity, parent=parent.ob), **kargs) class ListGob(Gob, GobHolder): def __init__(self, parent, vertical=True, **kargs): Gob.__init__(self, parent, tg.LinearLayout(activity=parent.root.activity, parent=parent.ob, vertical=vertical), **kargs) class Pager(GobBase): """A Pager can be created into a parent like a Gob, but to add children to it you need to create new pages and add children to those. """ def __init__(self, parent): GobBase.__init__(self, parent.root, parent) self.frame = parent.column() self.tg_tabs = tg.TabLayout(activity=parent.root.activity, parent=self.frame.ob) self.pages = [] # List of (name, root) tuples self.root.set_handler(self.tg_tabs.id, self.handle_event) self.tg_tabs.setlinearlayoutparams(0) self.tg_tabs.setheight(SHRINK_WRAP) def page(self, name): #page = self.frame.scroller().column(weight=1) # Scroller prevents the sections from spreading out. page = self.frame.column(weight=1) if self.pages: page.set_visible(False) self.pages.append((name, page)) self.tg_tabs.setlist([name for name, root in self.pages]) return page def handle_event(self, ev): if ev.type == 'itemselected': pageno = ev.value['selected'] if 0 <= pageno < len(self.pages): for i, (name, page) in enumerate(self.pages): page.set_visible(i==pageno) else: self.log(f"PAGER {ev.type}: {ev.value}") class Termux(Process): def __init__(self, process_manager, log_name): Process.__init__(self, process_manager, log_name) self.txc = None # Will become the termux Connection for this session. self.event_handlers = {} # Maps Activity id to event handling function. self.bridge = DynBridge(process_manager, f'{log_name}/bridge', local_port=None) # don't accept incoming, or publish anything. self.pool = None # will be trio Task scope for everything we want to cancel. async def build(self): pass def set_handler(self, aid, func): """Call this to start dispatching events for Activity aid to func. """ if func is None: if aid in self.event_handlers: del self.event_handlers[aid] else: self.event_handlers[aid] = func def handle_event(self, ev): aid = ev.value.get('aid') if aid in self.event_handlers: self.event_handlers[aid](ev) else: self.log(f"[{aid}] {ev.type}: {ev.value}") async def run_(self): self.txc = tg.Connection() try: async with pool() as self.pool: await self.build() # This may spin up some tasks on self.pool # # Wait for TermuxGui events in a background thread and push them to a foreground Queue: # evq = Queue(threaded=True) def handle_gui_events(): try: for event in self.txc.events(): evq.push_from_thread(event) if event.type == 'destroy' and event.value['aid'] == self.root.activity.aid: break except Exception as e: self.log(f"Background thread event loop died ({e})") self.log(f"Closing background thread.") self.spawn(trio.to_thread.run_sync(handle_gui_events)) # # Now poll events async style: # self.log("Waiting for events...") async for event in evq: #print(f"DEBUG: {event.type}: {event.value}") if event.type == 'destroy' and event.value['aid'] == self.root.activity.aid: break try: self.handle_event(event) except Exception as e: self.log(f"ERROR handling: {event.type} {event.value}") self.log_exception(e) self.log("Event Queue closed.") cancel(self.pool) # Stop all running tasks including the background thread finally: self.log("Closing Bridge") self.bridge.quit() self.bridge = None self.log("Closing termux-gui Connection") self.txc.close() self.txc = None def log_exception(self, e): self.log(str(e)) import traceback tb = traceback.extract_tb(e.__traceback__)[-1] filename = tb.filename.split('/')[-1] line_number = tb.lineno self.log(f"{line_number:4d} of {filename}") class MyTermux(Termux): async def build(self): self.root = await Root(self) pager = Pager(self.root) #self.buttonsize = SHRINK_WRAP self.buttonsize = 90 self.build_status_panel (pager.page("Status")) self.build_control_panel(pager.page("Control")) self.build_tv_panel (pager.page("TV")) self.build_therm_panel (pager.page("Therm")) def build_status_panel(self, frame): #frame.space(weight=1) row = frame.row() row.text_view('Misc') row = frame.row() row.dyn_value( 'doorlock/locked', 'Front\nDoor', fmt=(lambda v: 'Front\nDoor\nLocked' if v else 'Front\nDoor\nUnlocked'), colorer=(lambda v: 'green' if v else 'red')) row.dyn_float( 'freezer3/temp_sensor/temperature', 'Freezer', bands=(-10, -5, 5, 15)) row.dyn_float('water_heater/temp_sensor/temperature', 'Water\nHeater', bands=( 80, 90, 105, 120)) row.dyn_float( 'onyx/cpm30', 'Onyx', bands=( 0, 10, 50, 100)) row = frame.row() row.text_view('Motion') row = frame.row() def colorer(val): if val == 'Clear' or val == 0: return 'green' else: return 'red' for dev, label in [ ('aeotri_1', "M Bed"), ('motion_1', "G Bed"), ('zooz4in1_2', "DS Bath"), ('zooz4in1_1', "Garage"), ]: def fmt(val, label=label): if val == 'Clear' or val == 0: return f"{label}\n-" return f"{label}\nMOTION" row.dyn_value(f'{dev}/motion', label, fmt=fmt, colorer=colorer) #frame.space(weight=1) for prop, name, units, bands in [ ( 'radon', 'Radon', ' pCL', ( 0, 0, 2, 4)), ( 'co2', 'CO2', '/m', (200, 400, 600, 900)), ( 'vocs', 'VOCs', '/b', ( 0, 0, 250, 750)), ('humidity', 'Humidity', '%', {None: ( 30, 40, 50, 60), # Inside 'waveplus3': ( 20, 30, 70, 90), # Garage 'waveplus4': ( 20, 30, 70, 90)}), # Outside ( 'battery', 'Battery', 'V', (2.5, 2.6, 3.5, 3.5)), ]: row = frame.row() row.text_view(name) row = frame.row() for dev, label in [ ('waveplus1', 'Living\nRoom'), ('waveplus2', 'Office'), ('waveplus3', 'Garage'), ('waveplus4', 'Outside'), ]: if isinstance(bands, dict): b = bands.get(dev) if b is None: b = bands.get(None) else: b = bands row.dyn_float(f'{dev}/{prop}', label, units=units, bands=b) row = frame.row() bands = (10, 20, 100, 100) for dev, label in [ ( 'doorlock', "Door\nLock"), ( 'aeotri_1', "Master\nBedroom"), ('zooz4in1_1', "Garage\nZooz"), ('zooz4in1_2', "Kitchen\nZooz"), ]: row.dyn_float(f'{dev}/battery', label, units='%', bands=bands) #frame.space(weight=1) def build_control_panel(self, frame): frame.space(weight=1) beeper = self.bridge.find(('channels/zen' , 'inbox')) household = self.bridge.find(('channels/household', 'inbox')) row = frame.row() row.text_view('Lamps') row = frame.row(height=self.buttonsize) opts=[0, 20, 40, 60, 99] row.dyn_value('dimmer_1/brightness', "MB Win", options=opts) row.dyn_value('dimmer_2/brightness', "MB Door", options=opts) row.dyn_value('dimmer_5/brightness', "GB Door", options=opts) row.dyn_value('dimmer_4/brightness', "GB Win", options=opts) frame.space(weight=1) def food_in(minutes=0): if minutes: household.push(f"Food in {minutes}") else: household.push(f"Food") beeper.push('beep') row = frame.row() row.text_view('Beeps') row = frame.row(height=self.buttonsize) row.button("Food", on_click=food_in) row.button("in 5", on_click=lambda: food_in(5)) row.button("in 10", on_click=lambda: food_in(10)) row.button("Beep", on_click=lambda: beeper.push('beep')) frame.space(weight=1) row = frame.row() row.text_view('Misc') row = frame.row(height=self.buttonsize) row.dyn_float('watervalve/remain', 'Water', options=[0, 30, 60, 60*11.5], color='blue') if False: # Right now homebotty isn't set up to reset the timers based on a simple household message, # So we need to rethink this whole thing... frame.space(weight=1) row = frame.row() row.text_view('Kids') row = frame.row(height=self.buttonsize) for who in ['Time', 'Muse']: row.function(who, household.push, [('Pooped', f'{who} poop'), ('Asleep', f'{who} asleep'), ('Awake', f'{who} awake')]) frame.space(weight=1) def build_therm_panel(self, frame): frame.space(weight=1) row = frame.row(height=self.buttonsize) row.dyn_float('thermostat/temp_sensor/temperature' , 'Hall' , color='green') row.dyn_float('thermostat/remote_sensor/temperature', 'Bedroom', color='green') row.dyn_float( 'waveplus2/temperature', 'Office' ) row.dyn_float( 'waveplus4/temperature', 'Outside') row = frame.row(height=self.buttonsize) row.dyn_float('thermostat/heat' , 'Heat' , color='red') row.dyn_float('thermostat/heat_offset', '(offset)', color='red') row.dyn_float('thermostat/cool' , 'Cool' , color='blue') row.dyn_float('thermostat/cool_offset', '(offset)', color='blue') row = frame.row(height=self.buttonsize) row.dyn_value('thermostat/schedule', 'Schedule') row.dyn_value('thermostat/sensor' , 'Sensor') row.dyn_value('thermostat/fan' , 'Fan') row.dyn_value('thermostat/system' , 'System') row.dyn_value('thermostat/state' , 'State') frame.space(weight=1) def build_tv_panel(self, frame): frame.space(weight=1) row = frame.row() row.text_view('Status') row = frame.row(height=self.buttonsize) row.dyn_value('yamp/power' , label='Amp On') row.dyn_value('yamp/input' , label='Input') row.dyn_value('tvlg/power' , label='TV On') row.dyn_value('tvlg/application', label='App') frame.space(weight=1) row = frame.row() row.text_view('Scene') row = frame.row(height=self.buttonsize) row.dyn_function('scenes/run', label="Mode", label_path='scenes/scene', color='yellow') row.dyn_value('scenes/status', 'Status') row = frame.row(height=self.buttonsize) row.dyn_function('vlc/play_guest/play_movie', label='Movie', label_path='vlc/play_guest/playing', color='yellow') frame.space(weight=1) row = frame.row() row.text_view('Remote') row = frame.row(height=self.buttonsize) row.dyn_function('scenes/remote/stop' , label="STOP", color='red') row.dyn_function('scenes/remote/pause' , label="PAUSE", color='yellow') row.dyn_function('scenes/remote/play' , label="PLAY", color='green') row.dyn_function('scenes/remote/rewind', label="<<") row.dyn_function('scenes/remote/ffwd' , label=">>") row = frame.row(height=self.buttonsize) # Cache the control lines for later: self.yamp_mute = self.bridge.find(('yamp/mute' , 'var')) self.yamp_vol = self.bridge.find(('yamp/volume', 'var')) row.function( text=row.dyn_var('yamp/volume'), fmt=labler('Volume\n'), func=lambda vol: self.safe_spawn(self.adjust_volume(vol), "Set Vol"), options=[( "Quiet", '-40'), ("Normal", '-30')]) row.button( text=self.yamp_mute, fmt=labler("Mute\n"), on_click=lambda: self.safe_spawn(self.adjust_volume('mute'), "Mute Vol")) for name, vol in [ #( "MUTE", 'mute'), ( "Down", '+-2'), ( "Up", '+2'), ]: def func(vol=vol): self.safe_spawn(self.adjust_volume(vol), "Adj Vol") row.button(name, on_click=func) row.dyn_value('yamp/dialogue_level', label="Dialog\nBoost") row = frame.row(height=self.buttonsize) row.dyn_function('scenes/remote/subs' , label="Subtitles") frame.space(weight=1) async def adjust_volume(self, vol): """Here vol is a string, either 'mute', or '+i' or just 'i' where val is an int. To lower the volume, use, e.g., "+-2" """ with move_on_after(5): # Give up after 5 seconds so we don't hang the UI. if vol == 'mute': if await self.yamp_mute.current_value() != True: await self.yamp_mute.send_value(True) else: await self.yamp_mute.send_value(False) elif vol.startswith('+'): vol = int(vol[1:]) ovol = await self.yamp_vol.current_value() await self.yamp_mute.send_value(False) await self.yamp_vol.send_value(ovol + vol) else: vol = int(vol) await self.yamp_mute.send_value(False) await self.yamp_vol.send_value(vol)