mirror of
https://github.com/Stone-Red-Code/dvc.git
synced 2026-11-02 07:18:25 +01:00
dvc/dagascii: Use less instead of AsciiCanvas._do_draw
Uses less CLI tool to paging the output in the interactive mode while doing e.g. `dvc pipeline show ..`. Fixes #2807
This commit is contained in:
Tymoteusz Jankowski
+48
-93
@@ -2,8 +2,12 @@
|
||||
from __future__ import print_function
|
||||
from __future__ import unicode_literals
|
||||
|
||||
import logging
|
||||
import math
|
||||
import os
|
||||
import pydoc
|
||||
import sys
|
||||
from distutils.spawn import find_executable
|
||||
|
||||
from grandalf.graphs import Edge
|
||||
from grandalf.graphs import Graph
|
||||
@@ -12,6 +16,44 @@ from grandalf.layouts import SugiyamaLayout
|
||||
from grandalf.routing import EdgeViewer
|
||||
from grandalf.routing import route_with_lines
|
||||
|
||||
from dvc.utils import boxify
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
DEFAULT_PAGER = "less"
|
||||
DEFAULT_PAGER_FORMATTED = "{} --chop-long-lines --clear-screen".format(
|
||||
DEFAULT_PAGER
|
||||
)
|
||||
DVC_PAGER_VAR_NAME = "DVC_PAGER"
|
||||
DVC_PAGER_INFO = """\
|
||||
Less command line tool is missing.
|
||||
Install & add less tool to your PATH env. var. to automatically send output \
|
||||
to pager.
|
||||
Also, you can override default pager via {} env. var.
|
||||
""".format(
|
||||
DVC_PAGER_VAR_NAME
|
||||
)
|
||||
|
||||
|
||||
def find_pager(pager_cmd, pager_cmd_with_format, pager_env_name):
|
||||
if sys.stdout.isatty():
|
||||
less_found = find_executable(pager_cmd) is not None
|
||||
if less_found:
|
||||
pager_cmd = os.getenv(pager_env_name, pager_cmd_with_format)
|
||||
|
||||
def less_pager(text):
|
||||
return pydoc.tempfilepager(pydoc.plain(text), pager_cmd)
|
||||
|
||||
pager = less_pager
|
||||
else:
|
||||
logger.info(boxify(DVC_PAGER_INFO, border_color="yellow"))
|
||||
pager = pydoc.plainpager
|
||||
else:
|
||||
pager = pydoc.plainpager
|
||||
return pager
|
||||
|
||||
|
||||
class VertexViewer(object):
|
||||
"""Class to define vertex box boundaries that will be accounted for during
|
||||
@@ -60,99 +102,12 @@ class AsciiCanvas(object):
|
||||
|
||||
def draw(self):
|
||||
"""Draws ASCII canvas on the screen."""
|
||||
if sys.stdout.isatty(): # pragma: no cover
|
||||
from asciimatics.screen import Screen
|
||||
|
||||
Screen.wrapper(self._do_draw)
|
||||
else:
|
||||
for line in self.canvas:
|
||||
print("".join(line))
|
||||
|
||||
def _do_draw(self, screen): # pragma: no cover
|
||||
# pylint: disable=too-many-locals
|
||||
# pylint: disable=too-many-branches, too-many-statements
|
||||
from dvc.system import System
|
||||
from asciimatics.event import KeyboardEvent
|
||||
|
||||
offset_x = 0
|
||||
offset_y = 0
|
||||
smaxrow, smaxcol = screen.dimensions
|
||||
assert smaxrow > 1
|
||||
assert smaxcol > 1
|
||||
smaxrow -= 1
|
||||
smaxcol -= 1
|
||||
|
||||
if self.lines + 1 > smaxrow:
|
||||
max_y = self.lines + 1 - smaxrow
|
||||
else:
|
||||
max_y = 0
|
||||
|
||||
if self.cols + 1 > smaxcol:
|
||||
max_x = self.cols + 1 - smaxcol
|
||||
else:
|
||||
max_x = 0
|
||||
|
||||
while True:
|
||||
for y in range(smaxrow + 1):
|
||||
y_index = offset_y + y
|
||||
line = []
|
||||
for x in range(smaxcol + 1):
|
||||
x_index = offset_x + x
|
||||
if (
|
||||
len(self.canvas) > y_index
|
||||
and len(self.canvas[y_index]) > x_index
|
||||
):
|
||||
line.append(self.canvas[y_index][x_index])
|
||||
else:
|
||||
line.append(" ")
|
||||
assert len(line) == (smaxcol + 1)
|
||||
screen.print_at("".join(line), 0, y)
|
||||
|
||||
screen.refresh()
|
||||
|
||||
# NOTE: get_event() doesn't block by itself,
|
||||
# so we have to do the blocking ourselves.
|
||||
#
|
||||
# NOTE: using this workaround while waiting for PR [1]
|
||||
# to get merged and released. After that need to adjust
|
||||
# asciimatics version requirements.
|
||||
#
|
||||
# [1] https://github.com/peterbrittain/asciimatics/pull/188
|
||||
System.wait_for_input(self.TIMEOUT)
|
||||
|
||||
event = screen.get_event()
|
||||
if not isinstance(event, KeyboardEvent):
|
||||
continue
|
||||
|
||||
k = event.key_code
|
||||
if k == screen.KEY_DOWN or k == ord("s"):
|
||||
offset_y += 1
|
||||
elif k == screen.KEY_PAGE_DOWN or k == ord("S"):
|
||||
offset_y += smaxrow
|
||||
elif k == screen.KEY_UP or k == ord("w"):
|
||||
offset_y -= 1
|
||||
elif k == screen.KEY_PAGE_UP or k == ord("W"):
|
||||
offset_y -= smaxrow
|
||||
elif k == screen.KEY_RIGHT or k == ord("d"):
|
||||
offset_x += 1
|
||||
elif k == ord("D"):
|
||||
offset_x += smaxcol
|
||||
elif k == screen.KEY_LEFT or k == ord("a"):
|
||||
offset_x -= 1
|
||||
elif k == ord("A"):
|
||||
offset_x -= smaxcol
|
||||
elif k == ord("q") or k == ord("Q"):
|
||||
break
|
||||
|
||||
if offset_y > max_y:
|
||||
offset_y = max_y
|
||||
elif offset_y < 0:
|
||||
offset_y = 0
|
||||
|
||||
if offset_x > max_x:
|
||||
offset_x = max_x
|
||||
elif offset_x < 0:
|
||||
offset_x = 0
|
||||
pager = find_pager(
|
||||
DEFAULT_PAGER, DEFAULT_PAGER_FORMATTED, DVC_PAGER_VAR_NAME
|
||||
)
|
||||
lines = map("".join, self.canvas)
|
||||
joined_lines = os.linesep.join(lines)
|
||||
pager(joined_lines)
|
||||
|
||||
def point(self, x, y, char):
|
||||
"""Create a point on ASCII canvas.
|
||||
|
||||
@@ -218,41 +218,6 @@ class System(object):
|
||||
assert inode < 2 ** 64
|
||||
return inode
|
||||
|
||||
@staticmethod
|
||||
def _wait_for_input_windows(timeout):
|
||||
import sys
|
||||
import ctypes
|
||||
import msvcrt
|
||||
from ctypes.wintypes import DWORD, HANDLE
|
||||
|
||||
# https://docs.microsoft.com/en-us/windows/desktop/api/synchapi/nf-synchapi-waitforsingleobject
|
||||
from win32event import WAIT_OBJECT_0, WAIT_TIMEOUT
|
||||
|
||||
func = ctypes.windll.kernel32.WaitForSingleObject
|
||||
func.argtypes = [HANDLE, DWORD]
|
||||
func.restype = DWORD
|
||||
|
||||
rc = func(msvcrt.get_osfhandle(sys.stdin.fileno()), timeout * 1000)
|
||||
if rc not in [WAIT_OBJECT_0, WAIT_TIMEOUT]:
|
||||
raise RuntimeError(rc)
|
||||
|
||||
@staticmethod
|
||||
def _wait_for_input_posix(timeout):
|
||||
import sys
|
||||
import select
|
||||
|
||||
try:
|
||||
select.select([sys.stdin], [], [], timeout)
|
||||
except select.error:
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
def wait_for_input(timeout):
|
||||
if System.is_unix():
|
||||
return System._wait_for_input_posix(timeout)
|
||||
else:
|
||||
return System._wait_for_input_windows(timeout)
|
||||
|
||||
@staticmethod
|
||||
def is_symlink(path):
|
||||
path = fspath(path)
|
||||
|
||||
Reference in New Issue
Block a user