Example Functions

import os
import subprocess
import threading
 
import matplotlib.pyplot as plt
import urwid
from bamboost.extensions.remote_manager import RemoteManager, RemoteSimulation
from bamboostcli import api
from bamboostcli.common import CONFIG_DIR, Caller
from bamboostcli.widgets.custom_widgets import cConfirmDialog, cEdit, cPopup

def plot_phi(size, key):
    sim = api.get_entry_in_focus()
 
    def plot_phi_callback(fields: str = None) -> None:
        subprocess.run(
            f"python3 {CONFIG_DIR}/plot_pyvista.py -s {sim.get_full_uid()} {fields}".split()
        )
 
    edit = cEdit(
        "Enter phasefields to plot (space separated):", callback=plot_phi_callback
    )
    ui = cPopup(urwid.Filler(edit, valign="top"), title="Plot phasefields")
    Caller.enter_widget(ui)

def sync_remote_sim(*args) -> None:
    db: RemoteManager = api.Database()._self.db
    sim = db.shallow_sim(api.Database().uid_in_focus())
    if not isinstance(db, RemoteManager):
        api.Database().set_footer(
            urwid.Text(("footer", "Selected entry is not a RemoteSimulation"))
        )
        return
 
    def sync_remote_sim_thread(sim: RemoteSimulation, text_widget: urwid.Text) -> None:
        def write_to_widget(s: str) -> None:
            text_widget.set_text(text_widget.text + s)
            Caller.main_loop.draw_screen()
 
        text_widget.write = write_to_widget
        sim.sync(stdout=text_widget)
        text_widget.set_text(text_widget.text + f"Synced <{sim.uid}>")
        Caller.main_loop.draw_screen()
 
    text_widget = urwid.Text("")
    popup = cPopup(
        urwid.Filler(text_widget, valign="top"),
        title="Syncing remote simulation",
        height=("relative", 60),
        footer=("footer", "q: Exit widget | Q: Exit app"),
    )
    Caller.enter_widget(popup)
    Caller.main_loop.draw_screen()
    threading.Thread(
        target=sync_remote_sim_thread,
        args=(
            sim,
            text_widget,
        ),
    ).start()

def show_job(size, key):
    sim = api.get_entry_in_focus()
    slurm_dict = sim.metadata.get("_slurm", None)
    if slurm_dict is None:
        api.Database().container.footer.set_text(f"No slurm job found for <{sim.uid}>")
        return
 
    job_id = slurm_dict.get("jobId", None)
 
    def read_process_output(widget, job_id):
        env = os.environ.copy()
        res = subprocess.run(["myjobs", "-j", job_id], env=env, capture_output=True)
        widget.set_text(res.stdout)
        Caller.main_loop.draw_screen()
 
    text_widget = urwid.Text("")
    popup = cPopup(
        urwid.Filler(text_widget, valign="top"),
        title="Slurm job info (using myjobs)",
        footer=("footer", "q: Exit widget | Q: Exit app"),
        height=("relative", 100),
    )
    Caller.enter_widget(popup)
    Caller.main_loop.draw_screen()
    threading.Thread(target=read_process_output, args=(text_widget, job_id)).start()

def cancel_slurm_job(size, key):
    sim = api.get_entry_in_focus()
 
    def _cancel() -> None:
        slurm_dict = sim.metadata.get("_slurm", None)
        if slurm_dict is None:
            api.Database().set_footer(
                urwid.Text(("footer", f"No slurm job found for <{sim.uid}>"))
            )
            return
 
        job_id = slurm_dict.get("jobId", None)
        env = os.environ.copy()
        subprocess.run(["scancel", job_id], env=env)
        api.Database().set_footer(
            urwid.Text(("footer", f"Slurm job <{job_id}> canceled for <{sim.uid}>"))
        )
 
    ui = cConfirmDialog(
        Caller.main_loop, f"Are you sure you want to cancel {id}?", callback=_cancel
    )
    Caller.enter_widget(ui)