app.py

"""Talk to spaces VM via subprocess.check_output."""
# pylint: disable=wrong-import-position

from typing import Optional, Tuple
import sys
from pathlib import Path
if "." not in sys.path:
    sys.path.insert(0, ".")

# import httpx
import subprocess as sp
from shlex import split
import pandas as pd

import matplotlib
import matplotlib.pyplot as plt
import seaborn as sns

# from textwrap import dedent
from inspect import cleandoc
import gradio as gr
import logzero
from logzero import logger

from gradiobee.seg_text import seg_text

matplotlib.use("Agg")  # non-interactive for plt.savefig
sns.set()
sns.set_style("darkgrid")
logzero.loglevel()  # default to 10


# def greet(command):
def process(command) -> Tuple[str, Optional[str]]:
    """Probe vm."""
    # single line: command
    # logger.info("input:\n\t %s", command)
    # print("print input:\n\t %s" % command)
    # print("print input*:\n\t %s" % "*".join(command.splitlines()))

    logger.info("input: %s", command)

    is_command = True
    is_command = command.strip().splitlines().__len__() < 2 and len(command.strip()) < 500

    if is_command:
        try:
            # out = sp.check_output(split(command), encoding="utf8", stderr=STDOUT)
            proc = sp.Popen(
                split(command), encoding="utf8", stdout=-1, stderr=-1
            )  # sp.PIPE: -1
            out, err = proc.communicate()
            # success = True
            return f"[{out}], [{err}]", None
        except Exception as e:
            out, err = "", str(e)
            # success = False
            # return f"[{out}]", f"[{err}]"
            return f"[{out}], [{err}]", None

    # quick test altair altair-save tooltip
    # from PIL import Image

    # import altair as alt
    # from altair_saver import save
    # alt.renderers.enable('altair_saver', fmts=['vega-lite', 'png'])

    df_ = pd.DataFrame(data={'x': [1, 2], 'y': [3, 4], "cos": [0.1, 0.5]})
    # chart_df = alt.Chart(df_).mark_circle(size=60).encode(
        # x='x',
        # y='y',
        # color='cos',
        # tooltip=['x', 'y', 'cos', ]
    # )
    # .interactive()

    # save(chart_df, "chart_df.html")
    # chart_df_html = Path("chart_df.html").read_text("utf")
    # save(chart_df, "chart_df.png")
    # chart_df_png = Path("chart_df.png").read_bytes()

    # chart_df_png = Image.open("chart_df.png")
    # chart_df_png = "chart_df.png"

    # scatter_plot.save('simple_scatter_plot_with_altairchart.html')
    # chart_df.save("chart_df.html")  # does not work, constains js
    # chart_df_html = Path("chart_df.html").read_text("utf")

    # chart_df.save("chart_df.png")  #
    # chart_df_png = "chart_df.png"

    # not is_command or not flag: text, do seg_text
    _ = "\n\n".join(seg_text(command.strip()))

    logger.debug(_)
    # logger.debug(chart_df_html)
    # print(_)
    # print(chart_df_html)

    # _ = seg_text(command.strip())
    _ = cleandoc(
        f"""seg_text output (segmented sents):
        {_}
        """
    ).strip()

    # return _, chart_df_html

    plt.figure()
    sns.scatterplot(data=df_, x='x', y='y')
    Path("img").mkdir(exist_ok=True)
    plt.savefig("img/df.png")
    plt.close()
    df_png = "img/df.png"

    # return _, chart_df_png
    return _, df_png


iface = gr.Interface(
    # fn=greet,
    # inputs="text",
    fn=process,
    # inputs="text",
    inputs=gr.inputs.Textbox(
        lines=5,
        placeholder="Type or paste input here then click 'Submit'",
        default="python -m site",
        label="command or multiline text",
    ),
    # outputs="text",
    # outputs=["text",],
    # outputs=["text", "html"],
    outputs=[
        "text",
        gr.outputs.Image("auto"),
    ],
    examples=[
        "cat /proc/version",
        "free  # show free memory",
        "uname -m",
        "df -h .",
        "cat /proc/cpuinfo",
        """python -c "from psutil import virtual_memory; print(virtual_memory())" """,
    ],
    title="probe the system",
    description="Talk to the system via subprocess.check_output ",
    layout="vertical",
)

# iface.launch(share=True, debug=True)
import platform
logger.info("platform.node(): %s", platform.node())
iface.launch(share=False, debug=True)
# iface.launch()