Update space.py

space.py

@@ -1,43 +1,3 @@
-
-import gradio as gr
-from app import demo as app
-import os
-
-_docs = {'LogsView': {'description': 'Creates a component to visualize logs from a subprocess in real-time.', 'members': {'__init__': {'value': {'type': 'str | Callable | tuple[str] | None', 'default': 'None', 'description': 'Default value to show in the code editor. If callable, the function will be called whenever the app loads to set the initial value of the component.'}, 'every': {'type': 'float | None', 'default': 'None', 'description': "If `value` is a callable, run the function 'every' number of seconds while the client connection is open. Has no effect otherwise. The event can be accessed (e.g. to cancel it) via this component's .load_event attribute."}, 'lines': {'type': 'int', 'default': '5', 'description': None}, 'label': {'type': 'str | None', 'default': 'None', 'description': 'The label for this component. Appears above the component and is also used as the header if there are a table of examples for this component. If None and used in a `gr.Interface`, the label will be the name of the parameter this component is assigned to.'}, 'show_label': {'type': 'bool | None', 'default': 'None', 'description': 'if True, will display label.'}, 'container': {'type': 'bool', 'default': 'True', 'description': 'If True, will place the component in a container - providing some extra padding around the border.'}, 'scale': {'type': 'int | None', 'default': 'None', 'description': 'relative size compared to adjacent Components. For example if Components A and B are in a Row, and A has scale=2, and B has scale=1, A will be twice as wide as B. Should be an integer. scale applies in Rows, and to top-level Components in Blocks where fill_height=True.'}, 'min_width': {'type': 'int', 'default': '160', 'description': 'minimum pixel width, will wrap if not sufficient screen space to satisfy this value. If a certain scale value results in this Component being narrower than min_width, the min_width parameter will be respected first.'}, 'visible': {'type': 'bool', 'default': 'True', 'description': 'If False, component will be hidden.'}, 'elem_id': {'type': 'str | None', 'default': 'None', 'description': 'An optional string that is assigned as the id of this component in the HTML DOM. Can be used for targeting CSS styles.'}, 'elem_classes': {'type': 'list[str] | str | None', 'default': 'None', 'description': 'An optional list of strings that are assigned as the classes of this component in the HTML DOM. Can be used for targeting CSS styles.'}, 'render': {'type': 'bool', 'default': 'True', 'description': 'If False, component will not render be rendered in the Blocks context. Should be used if the intention is to assign event listeners now but render the component later.'}}, 'postprocess': {'value': {'type': 'list[Log]', 'description': 'Expects a list of `Log` logs.'}}, 'preprocess': {'return': {'type': 'LogsView', 'description': 'Passes the code entered as a `str`.'}, 'value': None}}, 'events': {'change': {'type': None, 'default': None, 'description': 'Triggered when the value of the LogsView changes either because of user input (e.g. a user types in a textbox) OR because of a function update (e.g. an image receives a value from the output of an event trigger). See `.input()` for a listener that is only triggered by user input.'}, 'input': {'type': None, 'default': None, 'description': 'This listener is triggered when the user changes the value of the LogsView.'}, 'focus': {'type': None, 'default': None, 'description': 'This listener is triggered when the LogsView is focused.'}, 'blur': {'type': None, 'default': None, 'description': 'This listener is triggered when the LogsView is unfocused/blurred.'}}}, '__meta__': {'additional_interfaces': {'Log': {'source': '@dataclass\nclass Log:\n    level: Literal[\n        "INFO", "DEBUG", "WARNING", "ERROR", "CRITICAL"\n    ]\n    message: str\n    timestamp: str'}, 'LogsView': {'source': 'class LogsView(Component):\n    EVENTS = [\n        Events.change,\n        Events.input,\n        Events.focus,\n        Events.blur,\n    ]\n\n    def __init__(\n        self,\n        value: str | Callable | tuple[str] | None = None,\n        *,\n        every: float | None = None,\n        lines: int = 5,\n        label: str | None = None,\n        show_label: bool | None = None,\n        container: bool = True,\n        scale: int | None = None,\n        min_width: int = 160,\n        visible: bool = True,\n        elem_id: str | None = None,\n        elem_classes: list[str] | str | None = None,\n        render: bool = True,\n    ):\n        self.language = "shell"\n        self.lines = lines\n        self.interactive = False\n        super().__init__(\n            label=label,\n            every=every,\n            show_label=show_label,\n            container=container,\n            scale=scale,\n            min_width=min_width,\n            visible=visible,\n            elem_id=elem_id,\n            elem_classes=elem_classes,\n            render=render,\n            value=value,\n        )\n\n    def preprocess(self, payload: str | None) -> "LogsView":\n        raise NotImplementedError(\n            "LogsView cannot be used as an input component."\n        )\n\n    def postprocess(self, value: List[Log]) -> List[Log]:\n        return value\n\n    def api_info(self) -> dict[str, Any]:\n        return {\n            "items": {\n                "level": "string",\n                "message": "string",\n                "timestamp": "number",\n            },\n            "title": "Logs",\n            "type": "array",\n        }\n\n    def example_payload(self) -> Any:\n        return [\n            Log(\n                "INFO",\n                "Hello World",\n                datetime.now().isoformat(),\n            )\n        ]\n\n    def example_value(self) -> Any:\n        return [\n            Log(\n                "INFO",\n                "Hello World",\n                datetime.now().isoformat(),\n            )\n        ]\n\n    @classmethod\n    def run_process(\n        cls,\n        command: List[str],\n        date_format: str = "%Y-%m-%d %H:%M:%S",\n    ) -> Generator[List[Log], None, None]:\n        process = subprocess.Popen(\n            command,\n            stdout=subprocess.PIPE,\n            stderr=subprocess.STDOUT,\n            text=True,\n        )\n\n        if process.stdout is None:\n            raise ValueError("stdout is None")\n\n        logs = []\n\n        def _log(level: str, message: str):\n            log = Log(\n                level=level,\n                message=message,\n                timestamp=datetime.now().strftime(\n                    date_format\n                ),\n            )\n            logs.append(log)\n            return logs\n\n        _log("INFO", f"Running {\' \'.join(command)}")\n        for line in process.stdout:\n            yield _log("INFO", line.strip())\n\n        # TODO: what if task is cancelled but process is still running?\n\n        process.stdout.close()\n        return_code = process.wait()\n        if return_code:\n            yield _log(\n                "ERROR",\n                f"Process exited with code {return_code}",\n            )\n        else:\n            yield _log(\n                "INFO", "Process exited successfully"\n            )\n\n    @classmethod\n    def run_thread(\n        cls,\n        fn: Callable,\n        log_level: int = logging.INFO,\n        logger_name: str | None = None,\n        date_format: str = "%Y-%m-%d %H:%M:%S",\n        **kwargs,\n    ) -> Generator[List[Log], None, None]:\n        logs = [\n            Log(\n                level="INFO",\n                message=f"Running {fn.__name__}({\', \'.join(f\'{k}={v}\' for k, v in kwargs.items())})",\n                timestamp=datetime.now().strftime(\n                    date_format\n                ),\n            )\n        ]\n        yield logs\n\n        thread = Thread(\n            target=non_failing_fn(fn), kwargs=kwargs\n        )\n\n        def _log(record: logging.LogRecord) -> bool:\n            if record.thread != thread.ident:\n                return False  # Skip if not from the thread\n            if logger_name and not record.name.startswith(\n                logger_name\n            ):\n                return False  # Skip if not from the logger\n            if record.levelno < log_level:\n                return False  # Skip if too verbose\n            log = Log(\n                level=record.levelname,\n                message=record.getMessage(),\n                timestamp=datetime.fromtimestamp(\n                    record.created\n                ).strftime(date_format),\n            )\n            logs.append(log)\n            return True\n\n        with capture_logging(log_level) as log_queue:\n            thread.start()\n\n            # Loop to capture and yield logs from the thread\n            while thread.is_alive():\n                while True:\n                    try:\n                        if _log(log_queue.get_nowait()):\n                            yield logs\n                    except queue.Empty:\n                        break\n                thread.join(\n                    timeout=0.1\n                )  # adjust the timeout as needed\n\n        # After the thread completes, yield any remaining logs\n        while True:\n            try:\n                if _log(log_queue.get_nowait()):\n                    yield logs\n            except queue.Empty:\n                break\n\n        logs.append(\n            Log(\n                level="INFO",\n                message="Thread completed successfully",\n                timestamp=datetime.now().strftime(\n                    date_format\n                ),\n            )\n        )'}}, 'user_fn_refs': {'LogsView': ['Log', 'LogsView']}}}
-
-abs_path = os.path.join(os.path.dirname(__file__), "css.css")
-
-with gr.Blocks(
-    css=abs_path,
-    theme=gr.themes.Default(
-        font_mono=[
-            gr.themes.GoogleFont("Inconsolata"),
-            "monospace",
-        ],
-    ),
-) as demo:
-    gr.Markdown(
-"""
-# `gradio_logsview`
-
-<div style="display: flex; gap: 7px;">
-<img alt="Static Badge" src="https://img.shields.io/badge/version%20-%200.0.1%20-%20orange">  
-</div>
-
-Visualize logs in your Gradio app
-""", elem_classes=["md-custom"], header_links=True)
-    app.render()
-    gr.Markdown(
-"""
-## Installation
-
-```bash
-pip install gradio_logsview
-```
-
-## Usage
-
-```python
 import logging
 import random
 import time
@@ -81,7 +41,7 @@ def fn_thread_failing():
     yield from LogsView.run_thread(random_values, log_level=logging.INFO, failing=True)
 
 
-markdown_top = \"\"\"
+markdown_top = """
 # LogsView Demo
 
 This demo shows how to use the `LogsView` component to display logs from a process or a thread in real-time.
@@ -89,10 +49,10 @@ This demo shows how to use the `LogsView` component to display logs from a proce
 Click on any button to launch a process or a thread and see the logs displayed in real-time.
 In the thread example, logs are generated randomly with different log levels.
 In the process example, logs are generated by a Python script but any command can be executed.
-\"\"\"
+"""
 
 
-markdown_bottom = \"\"\"
+markdown_bottom = """
 ## How to run in a thread?
 
 With `LogsView.run_thread`, you can run a function in a separate thread and capture logs in real-time.
@@ -130,7 +90,7 @@ with gr.Blocks() as demo:
     btn = gr.Button("Run process")
     btn.click(fn_process, outputs=logs)
 ```
-\"\"\"
+"""
 
 with gr.Blocks() as demo:
     gr.Markdown(markdown_top)
@@ -153,304 +113,3 @@ with gr.Blocks() as demo:
 
 if __name__ == "__main__":
     demo.launch()
-
-```
-""", elem_classes=["md-custom"], header_links=True)
-
-
-    gr.Markdown("""
-## `LogsView`
-
-### Initialization
-""", elem_classes=["md-custom"], header_links=True)
-
-    gr.ParamViewer(value=_docs["LogsView"]["members"]["__init__"], linkify=['Log', 'LogsView'])
-
-
-    gr.Markdown("### Events")
-    gr.ParamViewer(value=_docs["LogsView"]["events"], linkify=['Event'])
-
-
-
-
-    gr.Markdown("""
-
-### User function
-
-The impact on the users predict function varies depending on whether the component is used as an input or output for an event (or both).
-
-- When used as an Input, the component only impacts the input signature of the user function.
-- When used as an output, the component only impacts the return signature of the user function.
-
-The code snippet below is accurate in cases where the component is used as both an input and an output.
-
-- **As input:** Is passed, passes the code entered as a `str`.
-- **As output:** Should return, expects a list of `Log` logs.
-
- ```python
-def predict(
-    value: LogsView
-) -> list[Log]:
-    return value
-```
-""", elem_classes=["md-custom", "LogsView-user-fn"], header_links=True)
-
-
-
-
-    code_Log = gr.Markdown("""
-## `Log`
-```python
-@dataclass
-class Log:
-    level: Literal[
-        "INFO", "DEBUG", "WARNING", "ERROR", "CRITICAL"
-    ]
-    message: str
-    timestamp: str
-```""", elem_classes=["md-custom", "Log"], header_links=True)
-
-    code_LogsView = gr.Markdown("""
-## `LogsView`
-```python
-class LogsView(Component):
-    EVENTS = [
-        Events.change,
-        Events.input,
-        Events.focus,
-        Events.blur,
-    ]
-
-    def __init__(
-        self,
-        value: str | Callable | tuple[str] | None = None,
-        *,
-        every: float | None = None,
-        lines: int = 5,
-        label: str | None = None,
-        show_label: bool | None = None,
-        container: bool = True,
-        scale: int | None = None,
-        min_width: int = 160,
-        visible: bool = True,
-        elem_id: str | None = None,
-        elem_classes: list[str] | str | None = None,
-        render: bool = True,
-    ):
-        self.language = "shell"
-        self.lines = lines
-        self.interactive = False
-        super().__init__(
-            label=label,
-            every=every,
-            show_label=show_label,
-            container=container,
-            scale=scale,
-            min_width=min_width,
-            visible=visible,
-            elem_id=elem_id,
-            elem_classes=elem_classes,
-            render=render,
-            value=value,
-        )
-
-    def preprocess(self, payload: str | None) -> "LogsView":
-        raise NotImplementedError(
-            "LogsView cannot be used as an input component."
-        )
-
-    def postprocess(self, value: List[Log]) -> List[Log]:
-        return value
-
-    def api_info(self) -> dict[str, Any]:
-        return {
-            "items": {
-                "level": "string",
-                "message": "string",
-                "timestamp": "number",
-            },
-            "title": "Logs",
-            "type": "array",
-        }
-
-    def example_payload(self) -> Any:
-        return [
-            Log(
-                "INFO",
-                "Hello World",
-                datetime.now().isoformat(),
-            )
-        ]
-
-    def example_value(self) -> Any:
-        return [
-            Log(
-                "INFO",
-                "Hello World",
-                datetime.now().isoformat(),
-            )
-        ]
-
-    @classmethod
-    def run_process(
-        cls,
-        command: List[str],
-        date_format: str = "%Y-%m-%d %H:%M:%S",
-    ) -> Generator[List[Log], None, None]:
-        process = subprocess.Popen(
-            command,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.STDOUT,
-            text=True,
-        )
-
-        if process.stdout is None:
-            raise ValueError("stdout is None")
-
-        logs = []
-
-        def _log(level: str, message: str):
-            log = Log(
-                level=level,
-                message=message,
-                timestamp=datetime.now().strftime(
-                    date_format
-                ),
-            )
-            logs.append(log)
-            return logs
-
-        _log("INFO", f"Running {' '.join(command)}")
-        for line in process.stdout:
-            yield _log("INFO", line.strip())
-
-        # TODO: what if task is cancelled but process is still running?
-
-        process.stdout.close()
-        return_code = process.wait()
-        if return_code:
-            yield _log(
-                "ERROR",
-                f"Process exited with code {return_code}",
-            )
-        else:
-            yield _log(
-                "INFO", "Process exited successfully"
-            )
-
-    @classmethod
-    def run_thread(
-        cls,
-        fn: Callable,
-        log_level: int = logging.INFO,
-        logger_name: str | None = None,
-        date_format: str = "%Y-%m-%d %H:%M:%S",
-        **kwargs,
-    ) -> Generator[List[Log], None, None]:
-        logs = [
-            Log(
-                level="INFO",
-                message=f"Running {fn.__name__}({', '.join(f'{k}={v}' for k, v in kwargs.items())})",
-                timestamp=datetime.now().strftime(
-                    date_format
-                ),
-            )
-        ]
-        yield logs
-
-        thread = Thread(
-            target=non_failing_fn(fn), kwargs=kwargs
-        )
-
-        def _log(record: logging.LogRecord) -> bool:
-            if record.thread != thread.ident:
-                return False  # Skip if not from the thread
-            if logger_name and not record.name.startswith(
-                logger_name
-            ):
-                return False  # Skip if not from the logger
-            if record.levelno < log_level:
-                return False  # Skip if too verbose
-            log = Log(
-                level=record.levelname,
-                message=record.getMessage(),
-                timestamp=datetime.fromtimestamp(
-                    record.created
-                ).strftime(date_format),
-            )
-            logs.append(log)
-            return True
-
-        with capture_logging(log_level) as log_queue:
-            thread.start()
-
-            # Loop to capture and yield logs from the thread
-            while thread.is_alive():
-                while True:
-                    try:
-                        if _log(log_queue.get_nowait()):
-                            yield logs
-                    except queue.Empty:
-                        break
-                thread.join(
-                    timeout=0.1
-                )  # adjust the timeout as needed
-
-        # After the thread completes, yield any remaining logs
-        while True:
-            try:
-                if _log(log_queue.get_nowait()):
-                    yield logs
-            except queue.Empty:
-                break
-
-        logs.append(
-            Log(
-                level="INFO",
-                message="Thread completed successfully",
-                timestamp=datetime.now().strftime(
-                    date_format
-                ),
-            )
-        )
-```""", elem_classes=["md-custom", "LogsView"], header_links=True)
-
-    demo.load(None, js=r"""function() {
-    const refs = {
-            Log: [], 
-            LogsView: [], };
-    const user_fn_refs = {
-          LogsView: ['Log', 'LogsView'], };
-    requestAnimationFrame(() => {
-
-        Object.entries(user_fn_refs).forEach(([key, refs]) => {
-            if (refs.length > 0) {
-                const el = document.querySelector(`.${key}-user-fn`);
-                if (!el) return;
-                refs.forEach(ref => {
-                    el.innerHTML = el.innerHTML.replace(
-                        new RegExp("\\b"+ref+"\\b", "g"),
-                        `<a href="#h-${ref.toLowerCase()}">${ref}</a>`
-                    );
-                })
-            }
-        })
-
-        Object.entries(refs).forEach(([key, refs]) => {
-            if (refs.length > 0) {
-                const el = document.querySelector(`.${key}`);
-                if (!el) return;
-                refs.forEach(ref => {
-                    el.innerHTML = el.innerHTML.replace(
-                        new RegExp("\\b"+ref+"\\b", "g"),
-                        `<a href="#h-${ref.toLowerCase()}">${ref}</a>`
-                    );
-                })
-            }
-        })
-    })
-}
-
-""")
-
-demo.launch()