eris/tools.py

897 lines
29 KiB
Python
Raw Normal View History

2015-12-14 18:03:11 +00:00
# -*- coding: utf-8 -*-
2017-02-05 18:29:09 +01:00
# Copyright (C) 2015-2017 Andrew Hamilton. All rights reserved.
2015-12-14 18:03:11 +00:00
# Licensed under the Artistic License 2.0.
import ast
2016-01-21 23:22:42 +00:00
import contextlib
2015-12-14 18:03:11 +00:00
import dis
import enum
2015-12-14 18:03:11 +00:00
import functools
import gzip
2015-12-14 18:03:11 +00:00
import hashlib
import io
import math
import os
import os.path
import pickle
import pwd
import stat
import subprocess
import tempfile
import time
2016-02-19 18:27:20 +00:00
import traceback
2015-12-14 18:03:11 +00:00
import pygments
import pygments.lexers
import pygments.styles
import fill3
2016-01-01 17:55:43 +00:00
import gut
2016-02-19 18:27:20 +00:00
import lscolors
2015-12-14 18:03:11 +00:00
import termstr
2016-02-24 01:45:31 +00:00
CACHE_PATH = ".vigil"
2016-11-14 19:38:13 +01:00
if "PYGMENT_STYLE" not in os.environ:
os.environ["PYGMENT_STYLE"] = "native"
class Status(enum.IntEnum):
2015-12-14 18:03:11 +00:00
ok = 1
problem = 2
normal = 3
2015-12-14 18:03:11 +00:00
error = 4
not_applicable = 5
2015-12-14 18:03:11 +00:00
running = 6
pending = 7
2015-12-14 18:03:11 +00:00
paused = 8
2016-02-01 22:25:54 +00:00
timed_out = 9
2015-12-14 18:03:11 +00:00
_STATUS_COLORS = {Status.ok: termstr.Color.green,
Status.problem: termstr.Color.red,
Status.normal: termstr.Color.white,
Status.not_applicable: termstr.Color.grey_100,
Status.running: termstr.Color.light_blue,
Status.paused: termstr.Color.yellow,
Status.timed_out: termstr.Color.purple}
2015-12-14 18:03:11 +00:00
STATUS_MEANINGS = [
(Status.normal, "Normal"), (Status.ok, "Ok"),
(Status.problem, "Problem"), (Status.not_applicable, "Not applicable"),
(Status.running, "Running"), (Status.paused, "Paused"),
2016-02-01 22:25:54 +00:00
(Status.timed_out, "Timed out"), (Status.pending, "Pending"),
(Status.error, "Error")
]
2015-12-14 18:03:11 +00:00
_STATUS_TO_TERMSTR = {
status: termstr.TermStr(" ", termstr.CharStyle(bg_color=color))
for status, color in _STATUS_COLORS.items()}
_STATUS_TO_TERMSTR[Status.error] = termstr.TermStr(
2015-12-14 18:03:11 +00:00
"E", termstr.CharStyle(bg_color=termstr.Color.red))
_STATUS_TO_TERMSTR[Status.pending] = "."
2016-10-24 22:06:27 +02:00
STATUS_CURSOR_COLORS = {Status.ok: termstr.Color.black,
Status.problem: termstr.Color.white,
Status.normal: termstr.Color.black,
Status.not_applicable: termstr.Color.white,
Status.running: termstr.Color.white,
Status.paused: termstr.Color.black,
Status.timed_out: termstr.Color.white}
2015-12-14 18:03:11 +00:00
def get_ls_color_codes():
if "LS_COLORS" not in os.environ:
script = os.path.join(os.path.dirname(__file__), "LS_COLORS.sh")
with open(script) as file_:
codes = file_.readline().strip()[len("LS_COLORS='"):-len("';")]
os.environ["LS_COLORS"] = codes
return lscolors.get_color_codes(os.environ)
_LS_COLOR_CODES = get_ls_color_codes()
TIMEOUT = 60
2015-12-14 18:03:11 +00:00
def _printable(text):
return "".join(char if ord(char) > 31 or char in ["\n", "\t"] else "#"
for char in text)
def _fix_input(input_):
2015-12-14 18:03:11 +00:00
input_str = input_.decode("utf-8") if isinstance(input_, bytes) else input_
return _printable(input_str).expandtabs(tabsize=4)
2015-12-14 18:03:11 +00:00
def _do_command(command, timeout=None, **kwargs):
2015-12-14 18:03:11 +00:00
stdout, stderr = "", ""
2016-01-21 23:22:42 +00:00
with contextlib.suppress(subprocess.CalledProcessError):
2015-12-29 23:11:47 +00:00
process = subprocess.Popen(command, stdout=subprocess.PIPE,
2016-01-21 23:22:42 +00:00
stderr=subprocess.PIPE, **kwargs)
try:
stdout, stderr = process.communicate(timeout=timeout)
except subprocess.TimeoutExpired:
process.kill()
raise
return _fix_input(stdout), _fix_input(stderr), process.returncode
2015-12-14 18:03:11 +00:00
def _run_command(command, status_text=Status.ok):
2015-12-14 18:03:11 +00:00
status, output = status_text, ""
try:
2015-12-29 23:11:47 +00:00
process = subprocess.Popen(command, stdout=subprocess.PIPE,
2015-12-14 18:03:11 +00:00
stderr=subprocess.PIPE)
stdout, stderr = process.communicate()
output = stdout + stderr
except subprocess.CalledProcessError:
status = Status.problem
2015-12-14 18:03:11 +00:00
if process.returncode != 0:
status = Status.problem
return status, fill3.Text(_fix_input(output))
2015-12-14 18:03:11 +00:00
def _syntax_highlight(text, lexer, style):
def _parse_rgb(hex_rgb):
if hex_rgb.startswith("#"):
hex_rgb = hex_rgb[1:]
return tuple(eval("0x"+hex_rgb[index:index+2]) for index in [0, 2, 4])
2016-02-18 22:34:46 +00:00
def _char_style_for_token_type(token_type, default_bg_color):
token_style = style.style_for_token(token_type)
fg_color = (termstr.Color.black if token_style["color"] is None
else _parse_rgb(token_style["color"]))
bg_color = (default_bg_color if token_style["bgcolor"] is None
else _parse_rgb(token_style["bgcolor"]))
return termstr.CharStyle(fg_color, bg_color, token_style["bold"],
token_style["italic"],
token_style["underline"])
default_bg_color = _parse_rgb(style.background_color)
2016-02-18 22:34:46 +00:00
text = fill3.join(
"", [termstr.TermStr(text, _char_style_for_token_type(
token_type, default_bg_color))
for token_type, text in pygments.lex(text, lexer)])
return fill3.Text(text, pad_char=termstr.TermStr(" ").bg_color(
default_bg_color))
def _syntax_highlight_using_path(text, path):
2015-12-14 18:03:11 +00:00
lexer = pygments.lexers.get_lexer_for_filename(path, text)
style = pygments.styles.get_style_by_name(os.environ["PYGMENT_STYLE"])
return _syntax_highlight(text, lexer, style)
2015-12-14 18:03:11 +00:00
def pygments_(path):
with open(path) as file_:
try:
text = file_.read()
except UnicodeDecodeError:
return Status.not_applicable, fill3.Text("Not unicode")
2015-12-14 18:03:11 +00:00
else:
try:
source_widget = _syntax_highlight_using_path(_fix_input(text),
path)
2015-12-14 18:03:11 +00:00
except pygments.util.ClassNotFound:
return Status.normal, fill3.Text(text)
return Status.normal, source_widget
2015-12-14 18:03:11 +00:00
def linguist(path):
# Dep: ruby?, ruby-dev, libicu-dev, cmake, "gem install github-linguist"
return _run_command(["linguist", path], Status.normal)
2015-12-14 18:03:11 +00:00
def _permissions_in_octal(permissions):
result = []
for part_index in range(3):
index = part_index * 3 + 1
part = permissions[index:index+3]
digit = sum(2 ** (2 - index) for index, element in enumerate(part)
if element != "-")
result.append(str(digit))
return "".join(result)
def _pretty_bytes(bytes):
if bytes == 0:
return "0 B"
units = ("B", "KB", "MB", "GB", "TB", "PB", "EB", "ZB", "YB")
unit_index = int(math.floor(math.log(bytes, 1024)))
power = math.pow(1024, unit_index)
conversion = round(bytes/power, 2)
return "%s %s" % (conversion, units[unit_index])
def _md5(path):
2015-12-14 18:03:11 +00:00
with open(path, "rb") as file:
return hashlib.md5(file.read()).hexdigest()
def metadata(path):
2015-12-14 18:03:11 +00:00
def detail(value, unit):
2016-01-27 22:19:53 +00:00
result = (" (%s)" % value if unit is None else " (%s %s)" %
(value, unit))
return termstr.TermStr(result).fg_color(termstr.Color.grey_100)
2015-12-14 18:03:11 +00:00
is_symlink = "yes" if os.path.islink(path) else "no"
stat_result = os.stat(path)
permissions = stat.filemode(stat_result.st_mode)
2015-12-14 18:03:11 +00:00
hardlinks = str(stat_result.st_nlink)
group = [pwd.getpwuid(stat_result.st_gid).pw_name,
detail(stat_result.st_gid, "gid")]
2015-12-14 18:03:11 +00:00
owner = [pwd.getpwuid(stat_result.st_uid).pw_name,
detail(stat_result.st_uid, "uid")]
2015-12-14 18:03:11 +00:00
modified, created, access = [
[time.asctime(time.gmtime(seconds)), detail(int(seconds), "secs")]
2015-12-14 18:03:11 +00:00
for seconds in (stat_result.st_mtime, stat_result.st_ctime,
stat_result.st_atime)]
size = [_pretty_bytes(stat_result.st_size),
detail(stat_result.st_size, "bytes")]
2016-01-21 23:22:42 +00:00
stdout, *rest = _do_command(
2015-12-14 18:03:11 +00:00
["file", "--dereference", "--brief", "--uncompress", "--mime", path])
mime_type = stdout
2016-01-21 23:22:42 +00:00
stdout, *rest = _do_command(
2015-12-14 18:03:11 +00:00
["file", "--dereference", "--brief", "--uncompress", path])
file_type = stdout
md5sum = _md5(path)
2016-01-21 23:22:42 +00:00
stdout, *rest = _do_command(["sha1sum", path])
2015-12-14 18:03:11 +00:00
sha1sum = stdout.split()[0]
permissions_value = [permissions,
detail(_permissions_in_octal(permissions), None)]
2015-12-14 18:03:11 +00:00
text = []
for line in [
("size", size), ("permissions", permissions_value), None,
("modified time", modified), ("creation time", created),
("access time", access), None,
("owner", owner), ("group", group), None,
("hardlinks", hardlinks), ("symlink", is_symlink), None,
("md5", md5sum), ("sha1", sha1sum), None,
("mime type", mime_type.strip()),
("file type", file_type.strip())]:
if line is None:
text.append("\n")
else:
name, value = line
2016-01-27 22:19:53 +00:00
name = termstr.TermStr(name + ":").fg_color(
termstr.Color.light_blue).ljust(16)
text.append(name + fill3.join("", value) + "\n")
return (Status.normal, fill3.Text(fill3.join("", text)))
metadata.dependencies = {"file", "coreutils"}
2015-12-14 18:03:11 +00:00
def contents(path):
root, ext = splitext(path)
if ext == "":
with open(path) as file_:
return Status.normal, fill3.Text(_fix_input(file_.read()))
else:
return pygments_(path)
contents.dependencies = {"python3-pygments"}
def _is_python_syntax_correct(path, python_version):
if python_version == "python":
stdin, stdout, returncode = _do_command(
["python", "-c",
"__import__('compiler').parse(open('%s').read())" % path])
return returncode == 0
else: # python3
with open(path) as f:
source = f.read()
try:
ast.parse(source)
except:
return False
return True
def _python_version(path): # Need a better hueristic
for version in ["python3", "python"]:
if _is_python_syntax_correct(path, version):
return version
return "python3"
def python_syntax(path):
python_version = _python_version(path)
2016-01-10 18:29:52 +00:00
return _run_command([python_version, "-m", "py_compile", path])
python_syntax.dependencies = {"python", "python3"}
python_syntax.url = "https://en.wikipedia.org/wiki/Python_syntax_and_semantics"
2015-12-14 18:03:11 +00:00
def _has_shebang_line(path):
with open(path, "rb") as file_:
return file_.read(2) == "#!"
def _is_python_test_file(path):
path = str(os.path.basename(path))
return path.endswith("_test.py") or path.startswith("test_")
2015-12-14 18:03:11 +00:00
def python_unittests(path):
if _is_python_test_file(path):
2016-02-19 18:27:20 +00:00
command = ([path] if _has_shebang_line(path)
else [_python_version(path), path])
stdout, stderr, returncode = _do_command(command, timeout=TIMEOUT)
status = Status.ok if returncode == 0 else Status.problem
return status, fill3.Text(stdout + "\n" + stderr)
2015-12-14 18:03:11 +00:00
else:
return Status.not_applicable, fill3.Text("No tests.")
python_unittests.dependencies = {"python", "python3"}
python_unittests.url = "https://docs.python.org/3/library/unittest.html"
2015-12-14 18:03:11 +00:00
def pydoc(path):
pydoc_exe = "pydoc3" if _python_version(path) == "python3" else "pydoc"
status, output = Status.normal, ""
2015-12-14 18:03:11 +00:00
try:
output = subprocess.check_output([pydoc_exe, path], timeout=TIMEOUT)
output = _fix_input(output)
2015-12-14 18:03:11 +00:00
except subprocess.CalledProcessError:
status = Status.not_applicable
2015-12-14 18:03:11 +00:00
if not output.startswith("Help on module"):
status = Status.not_applicable
2015-12-14 18:03:11 +00:00
return status, fill3.Text(output)
pydoc.dependencies = {"python", "python3"}
pydoc.url = "https://docs.python.org/3/library/pydoc.html"
2015-12-14 18:03:11 +00:00
2016-10-15 18:45:32 +02:00
def mypy(path):
stdout, stderr, returncode = _do_command(["mypy", path], timeout=TIMEOUT)
status = Status.ok if returncode == 0 else Status.normal
return status, fill3.Text(stdout)
mypy.dependencies = {"mypy"}
mypy.url = "mypy"
2016-10-15 18:45:32 +02:00
def _colorize_coverage_report(text):
line_color = {"> ": termstr.Color.green, "! ": termstr.Color.red,
" ": None}
return fill3.join("", [termstr.TermStr(line).fg_color(line_color[line[:2]])
for line in text.splitlines(keepends=True)])
def python_coverage(path):
# FIX: Also use test_*.py files.
test_path = path[:-(len(".py"))] + "_test.py"
if os.path.exists(test_path):
with tempfile.TemporaryDirectory() as temp_dir:
python_exe = "%s-coverage" % _python_version(path)
coverage_path = os.path.join(temp_dir, "coverage")
env = os.environ.copy()
env["COVERAGE_FILE"] = coverage_path
2016-01-21 23:22:42 +00:00
stdout, *rest = _do_command(
[python_exe, "run", test_path], env=env, timeout=TIMEOUT)
path = os.path.normpath(path)
stdout, *rest = _do_command([python_exe, "annotate", "--directory",
temp_dir, path], env=env)
flat_path = path.replace("/", "_")
with open(os.path.join(temp_dir, flat_path + ",cover"), "r") as f:
stdout = f.read()
return Status.normal, fill3.Text(_colorize_coverage_report(stdout))
else:
return Status.not_applicable, fill3.Text(
"No corresponding test file: " + os.path.normpath(test_path))
python_coverage.dependencies = {"python-coverage", "python3-coverage"}
python_coverage.url = "python3-coverage"
def python_profile(path):
stdout, *rest = _do_command([_python_version(path), "-m", "cProfile",
"--sort=cumulative", path], timeout=TIMEOUT)
return Status.normal, fill3.Text(stdout)
python_profile.dependencies = {"python", "python3"}
python_profile.url = "https://docs.python.org/3/library/profile.html"
def pycodestyle(path):
cmd = (["pycodestyle"] if _python_version(path) == "python"
else ["python3", "-m", "pycodestyle"])
2016-09-27 13:30:08 +02:00
return _run_command(cmd + [path])
pycodestyle.dependencies = {"pycodestyle", "python3-pycodestyle"}
pycodestyle.url = "pycodestyle"
def pyflakes(path):
2016-01-10 18:29:52 +00:00
return _run_command([_python_version(path), "-m", "pyflakes", path])
pyflakes.dependencies = {"pyflakes"}
pyflakes.url = "pyflakes"
def pylint(path):
2016-01-10 18:29:52 +00:00
return _run_command([_python_version(path), "-m", "pylint",
"--errors-only", path])
pylint.dependencies = {"pylint", "pylint3"}
pylint.url = "pylint3"
def python_gut(path):
with open(path) as module_file:
output = gut.gut_module(module_file.read())
source_widget = _syntax_highlight_using_path(_fix_input(output), path)
return Status.normal, source_widget
python_gut.dependencies = set()
python_gut.url = "https://github.com/ahamilton/vigil/blob/master/gut.py"
def python_modulefinder(path):
2016-01-10 18:29:52 +00:00
return _run_command([_python_version(path), "-m", "modulefinder", path],
Status.normal)
python_modulefinder.dependencies = {"python", "python3"}
python_modulefinder.url = "https://docs.python.org/3/library/modulefinder.html"
2015-12-14 18:03:11 +00:00
def _get_mccabe_line_score(line, python_version):
position, function_name, score = line.split()
return int(score if python_version == "python3" else score[:-1])
def _colorize_mccabe(text, python_version):
return fill3.join("", [
termstr.TermStr(line).fg_color(termstr.Color.yellow)
if _get_mccabe_line_score(line, python_version) > 10 else line
for line in text.splitlines(keepends=True)])
def python_mccabe(path):
python_version = _python_version(path)
2016-01-21 23:22:42 +00:00
stdout, *rest = _do_command([python_version, "-m", "mccabe", path])
max_score = 0
2016-01-21 23:22:42 +00:00
with contextlib.suppress(ValueError): # When there are no lines
max_score = max(_get_mccabe_line_score(line, python_version)
for line in stdout.splitlines())
status = Status.problem if max_score > 10 else Status.ok
return status, fill3.Text(_colorize_mccabe(stdout, python_version))
python_mccabe.dependencies = {"python-mccabe", "python3-mccabe"}
python_mccabe.url = "python3-mccabe"
def python_tidy(path): # Deps: found on internet?
2016-01-21 23:22:42 +00:00
stdout, *rest = _do_command(["python", "python-tidy.py", path])
return Status.normal, _syntax_highlight_using_path(stdout, path)
2015-12-14 18:03:11 +00:00
def disassemble_pyc(path):
2016-02-10 19:30:30 +00:00
with open(path, "rb") as file_:
bytecode = file_.read()
2015-12-14 18:03:11 +00:00
stringio = io.StringIO()
dis.dis(bytecode, file=stringio)
stringio.seek(0)
return Status.normal, fill3.Text(stringio.read())
disassemble_pyc.dependencies = set()
disassemble_pyc.url = "https://docs.python.org/3/library/dis.html"
2015-12-14 18:03:11 +00:00
2016-10-25 22:48:20 +02:00
def bandit(path):
python_version = _python_version(path)
stdout, stderr, returncode = _do_command(
[python_version, "-m", "bandit.cli.main", "-f", "txt", path],
timeout=TIMEOUT)
status = Status.ok if returncode == 0 else Status.normal
text = stdout if python_version == "python" else _fix_input(eval(stdout))
text_without_timestamp = "".join(text.splitlines(keepends=True)[2:])
return status, fill3.Text(text_without_timestamp)
bandit.dependencies = {"python-bandit", "python3-bandit"}
bandit.url = "python3-bandit"
2016-10-25 22:48:20 +02:00
def _perl_version(path):
# stdout, stderr, returncode = _do_command(["perl", "-c", path])
# return "perl6" if "Perl v6.0.0 required" in stderr else "perl"
return "perl"
def perl_syntax(path):
return _run_command([_perl_version(path), "-c", path])
# perl_syntax.dependencies = {"perl", "rakudo"}
perl_syntax.dependencies = {"perl"}
perl_syntax.url = "https://en.wikipedia.org/wiki/Perl"
2015-12-14 18:03:11 +00:00
def perldoc(path):
2016-01-22 17:06:43 +00:00
stdout, stderr, returncode = _do_command(["perldoc", "-t", path])
return ((Status.normal, fill3.Text(stdout)) if returncode == 0
else (Status.not_applicable, fill3.Text(stderr)))
2015-12-14 18:03:11 +00:00
perldoc.dependencies = {"perl-doc"}
perldoc.url = "http://perldoc.perl.org/"
2015-12-14 18:03:11 +00:00
def perltidy(path):
2016-01-21 23:22:42 +00:00
stdout, *rest = _do_command(["perltidy", "-st", path])
return Status.normal, _syntax_highlight_using_path(stdout, path)
perltidy.dependencies = {"perltidy"}
perltidy.url = "http://perltidy.sourceforge.net/"
2015-12-14 18:03:11 +00:00
# def perl6_syntax(path):
# return _run_command(["perl6", "-c", path])
# perl6_syntax.dependencies = {"rakudo"}
def c_syntax_gcc(path):
2016-11-21 11:59:56 +01:00
return _run_command(["gcc", "-fsyntax-only", path])
c_syntax_gcc.dependencies = {"gcc"}
c_syntax_gcc.url = "https://gcc.gnu.org/"
2016-11-21 11:59:56 +01:00
2016-11-27 14:02:35 +01:00
def c_syntax_clang(path):
return _run_command(["clang", "-fsyntax-only", path])
c_syntax_clang.dependencies = {"clang"}
c_syntax_clang.url = "http://clang.llvm.org/"
def splint(path):
stdout, stderr, returncode = _do_command(["splint", "-preproc", path])
status = Status.ok if returncode == 0 else Status.problem
return status, fill3.Text(stdout + stderr)
splint.dependencies = {"splint"}
splint.url = "splint"
2015-12-14 18:03:11 +00:00
_OBJDUMP_URL = "https://en.wikipedia.org/wiki/Objdump"
2015-12-14 18:03:11 +00:00
def objdump_headers(path):
return _run_command(["objdump", "--all-headers", path], Status.normal)
2015-12-14 18:03:11 +00:00
objdump_headers.dependencies = {"binutils"}
objdump_headers.url = _OBJDUMP_URL
2015-12-14 18:03:11 +00:00
def objdump_disassemble(path):
return _run_command(
["objdump", "--disassemble", "--reloc", "--dynamic-reloc", path],
Status.normal)
2015-12-14 18:03:11 +00:00
objdump_disassemble.dependencies = {"binutils"}
objdump_disassemble.url = _OBJDUMP_URL
2015-12-14 18:03:11 +00:00
def readelf(path):
return _run_command(["readelf", "--all", path], Status.normal)
2015-12-14 18:03:11 +00:00
readelf.dependencies = {"binutils"}
readelf.url = _OBJDUMP_URL
2015-12-14 18:03:11 +00:00
def unzip(path):
return _run_command(["unzip", "-l", path], Status.normal)
2015-12-14 18:03:11 +00:00
unzip.dependencies = {"unzip"}
unzip.url = "unzip"
2015-12-14 18:03:11 +00:00
_TAR_URL = "http://www.gnu.org/software/tar/manual/tar.html"
2015-12-14 18:03:11 +00:00
def tar_gz(path):
return _run_command(["tar", "ztvf", path], Status.normal)
2015-12-14 18:03:11 +00:00
tar_gz.dependencies = {"tar"}
tar_gz.url = _TAR_URL
2015-12-14 18:03:11 +00:00
def tar_bz2(path):
return _run_command(["tar", "jtvf", path], Status.normal)
2015-12-14 18:03:11 +00:00
tar_bz2.dependencies = {"tar"}
tar_bz2.url = _TAR_URL
2015-12-14 18:03:11 +00:00
def nm(path):
return _run_command(["nm", "--demangle", path], Status.normal)
2015-12-14 18:03:11 +00:00
nm.dependencies = {"binutils"}
nm.url = "https://linux.die.net/man/1/nm"
2015-12-14 18:03:11 +00:00
def pdf2txt(path):
return _run_command(["pdf2txt", path], Status.normal)
2015-12-14 18:03:11 +00:00
pdf2txt.dependencies = {"python-pdfminer"}
pdf2txt.url = "python-pdfminer"
2015-12-14 18:03:11 +00:00
def html_syntax(path):
# Maybe only show errors
stdout, stderr, returncode = _do_command(["tidy", path])
status = Status.ok if returncode == 0 else Status.problem
2015-12-14 18:03:11 +00:00
return status, fill3.Text(stderr)
html_syntax.dependencies = {"tidy"}
html_syntax.url = "tidy"
2015-12-14 18:03:11 +00:00
def tidy(path):
2016-01-21 23:22:42 +00:00
stdout, *rest = _do_command(["tidy", path])
return Status.normal, fill3.Text(stdout)
2015-12-14 18:03:11 +00:00
tidy.dependencies = {"tidy"}
tidy.url = "tidy"
2015-12-14 18:03:11 +00:00
def html2text(path):
return _run_command(["html2text", path], Status.normal)
html2text.dependencies = {"html2text"}
html2text.url = "html2text"
def cpp_syntax_gcc(path):
2016-11-21 11:59:56 +01:00
return _run_command(["gcc", "-fsyntax-only", path])
cpp_syntax_gcc.dependencies = {"gcc"}
cpp_syntax_gcc.url = "https://gcc.gnu.org/"
2016-11-21 11:59:56 +01:00
2016-11-27 14:02:35 +01:00
def cpp_syntax_clang(path):
return _run_command(["clang", "-fsyntax-only", path])
cpp_syntax_clang.dependencies = {"clang"}
cpp_syntax_clang.url = "http://clang.llvm.org/"
2015-12-14 18:03:11 +00:00
def bcpp(path):
stdout, stderr, returncode = _do_command(["bcpp", "-fi", path])
status = Status.normal if returncode == 0 else Status.problem
return status, _syntax_highlight_using_path(stdout, path)
2015-12-14 18:03:11 +00:00
bcpp.dependencies = {"bcpp"}
def uncrustify(path):
with tempfile.TemporaryDirectory() as temp_dir:
config_path = os.path.join(temp_dir, "uncrustify.cfg")
stdout, stderr, returncode = _do_command(
["uncrustify", "--detect", "-f", path, "-o", config_path])
2016-02-08 21:55:22 +00:00
if returncode == 0:
stdout, stderr, returncode = _do_command(
["uncrustify", "-c", config_path, "-f", path])
status = Status.normal if returncode == 0 else Status.problem
return status, _syntax_highlight_using_path(stdout, path)
2015-12-14 18:03:11 +00:00
uncrustify.dependencies = {"uncrustify"}
uncrustify.url = "uncrustify"
2015-12-14 18:03:11 +00:00
def php5_syntax(path):
2016-01-10 18:29:52 +00:00
return _run_command(["php", "--syntax-check", path])
2016-09-27 13:30:08 +02:00
php5_syntax.dependencies = {"php"}
php5_syntax.url = "https://en.wikipedia.org/wiki/PHP"
2015-12-14 18:03:11 +00:00
2016-02-13 18:55:15 +00:00
#############################
2016-10-21 23:11:18 +02:00
LOG_PATH = os.path.join(os.getcwd(), "vigil.log")
def log_error(message=None):
message = traceback.format_exc() if message is None else message + "\n"
with open(LOG_PATH, "a") as log_file:
log_file.write(message)
def lru_cache_with_eviction(maxsize=128, typed=False):
versions = {}
make_key = functools._make_key
def evict(*args, **kwds):
key = make_key(args, kwds, typed)
if key in versions:
versions[key] += 1
def decorating_function(user_function):
def remove_version(*args, **kwds):
return user_function(*args[1:], **kwds)
new_func = functools.lru_cache(maxsize=maxsize, typed=typed)(
remove_version)
def add_version(*args, **kwds):
key = make_key(args, kwds, typed)
return new_func(*((versions.setdefault(key, 0),) + args), **kwds)
add_version.versions = versions
add_version.cache_info = new_func.cache_info
add_version.evict = evict
return functools.update_wrapper(add_version, user_function)
return decorating_function
def dump_pickle_safe(object_, path, protocol=pickle.HIGHEST_PROTOCOL,
open=open):
tmp_path = path + ".tmp"
try:
with open(tmp_path, "wb") as file_:
pickle.dump(object_, file_, protocol=protocol)
except (OSError, KeyboardInterrupt):
os.remove(tmp_path)
else:
os.rename(tmp_path, path)
def status_to_str(status):
return (_STATUS_TO_TERMSTR[status] if isinstance(status, enum.Enum)
else status)
class Result:
2016-10-21 22:24:52 +02:00
def __init__(self, path, tool):
self.path = path
self.tool = tool
2016-02-24 01:45:31 +00:00
self.pickle_path = os.path.join(CACHE_PATH, path + "-" + tool.__name__)
self.scroll_position = (0, 0)
self.is_completed = False
self.is_placeholder = True
self.status = Status.pending
@property
@lru_cache_with_eviction(maxsize=50)
def result(self):
unknown_label = fill3.Text("?")
if self.is_placeholder:
return unknown_label
try:
2016-10-21 22:24:52 +02:00
with gzip.open(self.pickle_path, "rb") as pickle_file:
return pickle.load(pickle_file)
except FileNotFoundError:
return unknown_label
@result.setter
def result(self, value):
os.makedirs(os.path.dirname(self.pickle_path), exist_ok=True)
2016-10-21 22:24:52 +02:00
dump_pickle_safe(value, self.pickle_path, open=gzip.open)
Result.result.fget.evict(self)
def set_status(self, status):
self.status = status
self.entry.appearance_cache = None
async def run(self, log, appearance_changed_event, runner):
self.is_placeholder = False
2016-02-24 01:45:31 +00:00
tool_name = tool_name_colored(self.tool, self.path)
path = path_colored(self.path)
log.log_message(["Running ", tool_name, " on ", path, "..."])
self.set_status(Status.running)
if runner.is_already_paused:
runner.is_already_paused = False
runner.pause()
appearance_changed_event.set()
start_time = time.time()
new_status = await runner.run_tool(self.path, self.tool)
Result.result.fget.evict(self)
end_time = time.time()
self.set_status(new_status)
appearance_changed_event.set()
self.is_completed = True
log.log_message(
2016-02-24 01:45:31 +00:00
["Finished running ", tool_name, " on ", path, ". ",
status_to_str(new_status), " %s secs" %
round(end_time - start_time, 2)])
def reset(self):
self.is_placeholder = True
self.set_status(Status.pending)
def appearance_min(self):
return [status_to_str(self.status)]
def _generic_tools():
return [contents, metadata]
2015-12-14 18:03:11 +00:00
2016-11-19 13:50:07 +01:00
TOOLS_FOR_EXTENSIONS = \
[
(["py"], [python_syntax, python_unittests, pydoc, mypy, python_coverage,
python_profile, pycodestyle, pyflakes, pylint, python_gut,
python_modulefinder, python_mccabe, bandit]),
(["pyc"], [disassemble_pyc]),
(["pl", "pm", "t"], [perl_syntax, perldoc, perltidy]),
# (["p6", "pm6"], [perl6_syntax, perldoc]),
(["pod", "pod6"], [perldoc]),
(["java"], [uncrustify]),
2016-11-27 14:02:35 +01:00
(["c", "h"], [c_syntax_gcc, c_syntax_clang, splint, uncrustify]),
2016-11-19 13:50:07 +01:00
(["o"], [objdump_headers, objdump_disassemble, readelf]),
(["cc", "cpp", "hpp"], [cpp_syntax_gcc, cpp_syntax_clang, bcpp,
uncrustify]),
2016-11-19 13:50:07 +01:00
(["pdf"], [pdf2txt]),
(["html"], [html_syntax, tidy, html2text]),
(["php"], [php5_syntax]),
(["zip"], [unzip]),
(["tar.gz", "tgz"], [tar_gz]),
(["tar.bz2"], [tar_bz2]),
(["a", "so"], [nm]),
]
@functools.lru_cache(maxsize=1)
def _tools_for_extension():
2016-11-19 13:50:07 +01:00
result = {}
for extensions, tools in TOOLS_FOR_EXTENSIONS:
for extension in extensions:
result[extension] = tools
return result
2015-12-14 18:03:11 +00:00
def tools_all():
tools_ = set(_generic_tools())
for tool_list in _tools_for_extension().values():
2015-12-14 18:03:11 +00:00
tools_.update(set(tool_list))
return tools_
def dependencies():
dependencies_all = set()
for tool in tools_all():
dependencies_all.update(tool.dependencies)
2015-12-14 18:03:11 +00:00
return dependencies_all
def splitext(path):
root, ext = os.path.splitext(path)
if "." in root:
for compound_ext in [".tar.gz", ".tar.bz2"]:
if path.endswith(compound_ext):
return path[:-len(compound_ext)], path[-len(compound_ext):]
return root, ext
def tools_for_path(path):
root, ext = splitext(path)
extra_tools = [] if ext == "" else _tools_for_extension().get(ext[1:], [])
return _generic_tools() + extra_tools
2015-12-14 18:03:11 +00:00
def run_tool_no_error(path, tool):
try:
status, result = tool(path)
except subprocess.TimeoutExpired:
2016-02-01 22:25:54 +00:00
status, result = Status.timed_out, fill3.Text("Timed out")
2015-12-14 18:03:11 +00:00
except:
status, result = Status.error, _syntax_highlight(
2016-02-24 01:45:31 +00:00
traceback.format_exc(), pygments.lexers.PythonTracebackLexer(),
pygments.styles.get_style_by_name(os.environ["PYGMENT_STYLE"]))
2015-12-14 18:03:11 +00:00
return status, result
2015-12-30 00:45:53 +00:00
def _convert_lscolor_code_to_charstyle(lscolor_code):
parts = lscolor_code.split(";")
if len(parts) == 1:
is_bold = parts[0] == "1"
fg_color = None
elif len(parts) == 2:
is_bold = False
fg_color = int(parts[1])
2015-12-30 00:45:53 +00:00
else:
is_bold = len(parts) == 4 and parts[3] == "1"
fg_color = int(parts[2])
return termstr.CharStyle(fg_color=fg_color, is_bold=is_bold)
def _charstyle_of_path(path):
color_code = lscolors.color_code_for_path(path, _LS_COLOR_CODES)
2015-12-30 00:45:53 +00:00
return (termstr.CharStyle() if color_code is None else
_convert_lscolor_code_to_charstyle(color_code))
2015-12-14 18:03:11 +00:00
@functools.lru_cache(maxsize=100)
2016-02-24 01:45:31 +00:00
def path_colored(path):
2015-12-30 00:45:53 +00:00
char_style = _charstyle_of_path(path)
2015-12-14 18:03:11 +00:00
path = path[2:]
dirname, basename = os.path.split(path)
if dirname == "":
return termstr.TermStr(basename, char_style)
else:
dirname = dirname + os.path.sep
2015-12-30 00:45:53 +00:00
return (termstr.TermStr(dirname, _charstyle_of_path(dirname)) +
2015-12-14 18:03:11 +00:00
termstr.TermStr(basename, char_style))
@functools.lru_cache(maxsize=100)
2016-02-24 01:45:31 +00:00
def tool_name_colored(tool, path):
char_style = (termstr.CharStyle(is_bold=True) if tool in _generic_tools()
2015-12-30 00:45:53 +00:00
else _charstyle_of_path(path))
2015-12-14 18:03:11 +00:00
return termstr.TermStr(tool.__name__, char_style)
@functools.lru_cache()
def get_homepage_of_package(package):
line = subprocess.getoutput("dpkg-query --status %s | grep Homepage" % package)
return line.split()[1]
def url_of_tool(tool):
try:
url = tool.url
return url if url.startswith("http") else get_homepage_of_package(url)
except AttributeError:
return None