Release v0.2.0 (#13)

Release v0.2.0

README.md

@@ -5,7 +5,6 @@
 <h3 align="center">A Top-like Interface for Austin</h3>
 
 
-
 <p align="center">
  <a href="https://github.com/P403n1x87/austin-tui/actions?workflow=Tests">
  <img src="https://github.com/P403n1x87/austin-tui/workflows/Tests/badge.svg"

austin_tui/__init__.py

@@ -29,12 +29,12 @@ def write_exception_to_file(e):
  trace = "".join(tb.format_tb(e.__traceback__))
  trace += f"{type(e).__qualname__}: {e}"
  # print(trace)
- with open("/tmp/austin-tui.out", "a") as fout:
+ with open("./austin-tui.out", "a") as fout:
  fout.write(trace + "\n")
 
 
 def fp(text):
- with open("/tmp/austin-tui.out", "a") as fout:
+ with open("./austin-tui.out", "a") as fout:
  fout.write(str(text) + "\n")
 
 

austin_tui/__main__.py

@@ -22,7 +22,8 @@
 
 import asyncio
 import sys
-from typing import List, Optional
+from textwrap import wrap
+from typing import Any, List, Optional
 
 from austin import AustinError, AustinTerminated
 from austin.aio import AsyncAustin
@@ -40,11 +41,28 @@
  _get_all_tasks = asyncio.Task.all_tasks # Python 3.6 compatibility
 
 
+def _print(text: str) -> None:
+ for line in wrap(text, 78):
+ print(line)
+
+
 class AustinTUIArgumentParser(AustinArgumentParser):
  """Austin TUI implementation of the Austin argument parser."""
 
  def __init__(self) -> None:
- super().__init__(name="austin-tui", full=False, alt_format=False)
+ super().__init__(
+ name="austin-tui", full=False, alt_format=False, exposure=False
+ )
+
+ def parse_args(self) -> Any:
+ """Parse command line arguments and report any errors."""
+ try:
+ return super().parse_args()
+ except AustinCommandLineError as e:
+ reason, *code = e.args
+ if reason:
+ _print(reason)
+ exit(code[0] if code else -1)
 
 
 class AustinTUI(AsyncAustin):
@@ -53,14 +71,7 @@ class AustinTUI(AsyncAustin):
  def __init__(self) -> None:
  super().__init__()
 
- try:
- self._args = AustinTUIArgumentParser().parse_args()
- except AustinCommandLineError as e:
- reason, *code = e.args
- if reason:
- print(reason)
- AustinTUIArgumentParser().print_help()
- exit(code[0] if code else -1)
+ self._args = AustinTUIArgumentParser().parse_args()
 
  self._model = AustinModel()
  self._view: AustinView = ViewBuilder.from_resource(
@@ -69,17 +80,13 @@ def __init__(self) -> None:
  self._view.mode = (
  AustinProfileMode.MEMORY if self._args.memory else AustinProfileMode.TIME
  )
+ self._view.callback = self.on_view_event
 
  self._global_stats: Optional[str] = None
 
  def on_sample_received(self, sample: str) -> None:
  """Austin sample received callback."""
- try:
- self._model.update(sample)
- except Exception as e:
- from austin_tui import write_exception_to_file
-
- write_exception_to_file(e)
+ self._model.update(sample)
 
  @catch
  def on_ready(
@@ -104,58 +111,71 @@ def on_terminate(self, stats: str) -> None:
  self._global_stats = stats
  self._view.stop()
 
+ def on_view_event(self, event: AustinView.Event, data: Any = None) -> None:
+ """View events handler."""
+
+ def _unhandled() -> None:
+ raise RuntimeError(f"Unhandled view event: {event}")
+
+ {AustinView.Event.QUIT: self.shutdown}.get(event, _unhandled)()
+
  async def start(self, args: List[str]) -> None:
  """Start Austin and catch any exceptions."""
  try:
  await super().start(args)
  except AustinTerminated:
  pass
- except AustinError as e:
- raise KeyboardInterrupt("Failed to start") from e
+ except Exception as e:
+ self.shutdown(e)
 
  def run(self) -> None:
  """Run the TUI."""
  loop = asyncio.get_event_loop()
 
  try:
- print("🏁 Starting the Austin TUI ...")
- loop.create_task(self.start(AustinTUIArgumentParser.to_list(self._args)))
+ print("🏁 Starting the Austin TUI ...", end="\r")
+ austin_task = loop.create_task(
+ self.start(AustinTUIArgumentParser.to_list(self._args))
+ )
  loop.run_forever()
- except KeyboardInterrupt as e:
- if e.__cause__:
- print(
- "❌ Austin failed to start. Please make sure that the Austin binary is\n"
- "available from the PATH environment variable and that the command line\n"
- "arguments that you have provided are correct."
- )
- finally:
+ if not austin_task.done():
+ austin_task.cancel()
+ except AustinError as e:
+ self.shutdown(e)
+ except KeyboardInterrupt:
  self.shutdown()
 
- def shutdown(self) -> None:
+ def shutdown(self, exception: Optional[Exception] = None) -> None:
  """Shutdown the TUI."""
- self._view.close()
-
  try:
  self.terminate()
  except AustinError:
  pass
 
- for task in _get_all_tasks():
- task.cancel()
+ self._view.close()
+
+ if exception:
+ message = exception.args[0]
+ if message[0] == "(":
+ _, _, message = message.partition(") ")
+ _print(message)
+
+ loop = asyncio.get_event_loop()
+ loop.stop()
 
  pending = [task for task in _get_all_tasks() if not task.done()]
  if pending:
- done, _ = asyncio.get_event_loop().run_until_complete(asyncio.wait(pending))
- for t in done:
- try:
- res = t.result()
- if res:
- print(res)
- except (AustinError, asyncio.CancelledError):
- pass
-
- if self._global_stats:
- print(self._global_stats)
+ try:
+  done, _ = loop.run_until_complete(asyncio.wait(pending))
+ for t in done:
+ try:
+  res = t.result()
+ if res:
+  print(res)
+ except (AustinError, asyncio.CancelledError):
+ pass
+  except Exception:
+  pass
 
 
 def main() -> None:

austin_tui/view/__init__.py

@@ -23,7 +23,7 @@
 import asyncio
 import curses
 import sys
-from typing import Any, Callable, Dict, Optional, TextIO, Type
+from typing import Any, Callable, Dict, List, Optional, TextIO, Type
 
 from austin_tui.view.palette import Palette
 from austin_tui.widgets import Container, Widget
@@ -81,11 +81,15 @@ def _validate_ns(node: Element) -> None:
 
 
 class View:
- """View object."""
+ """View object.
+
+ All coroutines are collected and scheduled for execution when the view is
+ opened.
+ """
 
  def __init__(self, name: str) -> None:
- self._input_event = asyncio.Event()
- self._input_task = asyncio.get_event_loop().create_task(self._input_loop())
+ self._tasks = []
+
  self._event_handlers: Dict[str, EventHandler] = {}
 
  self._open = False
@@ -94,12 +98,21 @@ def __init__(self, name: str) -> None:
  self.palette = Palette()
  self.root_widget = None
 
+ def _create_tasks(self) -> List[asyncio.Task]:
+ loop = asyncio.get_event_loop()
+ self._tasks = [
+ loop.create_task(coro())
+ for coro in (
+ attr
+ for attr in (getattr(self, name) for name in dir(self))
+ if callable(attr) and asyncio.iscoroutinefunction(attr)
+ )
+ ]
+
  async def _input_loop(self) -> None:
  if not self.root_widget:
  raise RuntimeError("Missing root widget")
 
- await self._input_event.wait()
-
  while self._open:
  await asyncio.sleep(0.015)
 
@@ -112,12 +125,6 @@ async def _input_loop(self) -> None:
  except (KeyError, curses.error):
  pass
 
- except Exception as e:
- from austin_tui import write_exception_to_file
-
- write_exception_to_file(e)
- raise KeyboardInterrupt()
-
  def _build(self, node: Element) -> Widget:
  _validate_ns(node)
  widget_class = QName(node).localname
@@ -141,7 +148,11 @@ def markup(self, text: Any) -> AttrString:
  return markup(str(text), self.palette)
 
  def open(self) -> None:
- """Open the view."""
+ """Open the view.
+
+ Calling this method not only shows the TUI on screen, but also collects
+ and schedules all the coroutines on the instance with the event loop.
+ """
  if not self.root_widget:
  raise RuntimeError("View has no root widget")
 
@@ -153,15 +164,20 @@ def open(self) -> None:
  self.root_widget.refresh()
 
  self._open = True
- self._input_event.set()
+
+ self._create_tasks()
 
  def close(self) -> None:
  """Close the view."""
  if not self._open or not self.root_widget:
  return
 
  self.root_widget.hide()
- self._input_task.cancel()
+
+ for task in self._tasks:
+ task.cancel()
+
+ task = []
  self._open = False
 
  @property
@@ -237,43 +253,3 @@ def from_resource(module: str, resource: str) -> View:
  files(module).joinpath(resource).read_text(encoding="utf8").encode()
  )
  )
-
-
-# if __name__ == "__main__":
-#
-# class AustinView(View):
-# def on_quit(self, data=None):
-# raise KeyboardInterrupt("Quit event")
-#
-# def on_previous_thread(self, data=None):
-# title = self.title.get_text()
-# self.title.set_text(title[1:] + title[0])
-# self.title.refresh()
-#
-# def on_next_thread(self, data=None):
-# title = self.title.get_text()
-# self.title.set_text(title[-1] + title[:-1])
-# self.title.refresh()
-#
-# def on_full_mode_toggled(self, data=None):
-# pass
-#
-# def on_table_up(self, data=None):
-# return self.table.scroll_up()
-#
-# def on_table_down(self, data=None):
-# return self.table.scroll_down()
-#
-# view = ViewBuilder.from_resource("austin_tui.view", "tui.austinui")
-# exc = None
-# try:
-# view.open()
-#
-# loop = asyncio.get_event_loop()
-# loop.run_forever()
-# except Exception as e:
-# exc = e
-# finally:
-# view.close()
-# if exc:
-# raise exc