ahamilton/eris
1
0
Fork 0
Code Releases Activity

Coding style.

Browse source
This commit is contained in:
Andrew Hamilton 2016-10-21 18:12:33 +02:00
parent c695479499
commit 29edc33337
2 changed files with 59 additions and 59 deletions
Download patch file Download diff file Expand all files Collapse all files

108
vigil
View file

@ -84,12 +84,12 @@ KEYS_DOC = """Keys:
"""
_LOG_PATH = os.path.join(os.getcwd(), "vigil.log")
LOG_PATH = os.path.join(os.getcwd(), "vigil.log")
def _log_error(message=None):
def log_error(message=None):
message = traceback.format_exc() if message is None else message + "\n"
with open(_LOG_PATH, "a") as log_file:
with open(LOG_PATH, "a") as log_file:
log_file.write(message)
@ -135,27 +135,27 @@ class Entry(collections.UserList):
return appearance
def _is_path_excluded(path):
def is_path_excluded(path):
return any(part.startswith(".") for part in path.split(os.path.sep))
def _codebase_files(path, skip_hidden_directories=True):
def codebase_files(path, skip_hidden_directories=True):
for (dirpath, dirnames, filenames) in os.walk(path):
if skip_hidden_directories:
filtered_dirnames = [dirname for dirname in dirnames
if not _is_path_excluded(dirname)]
if not is_path_excluded(dirname)]
dirnames[:] = filtered_dirnames
for filename in filenames:
if not _is_path_excluded(filename):
if not is_path_excluded(filename):
yield os.path.join(dirpath, filename)
def _fix_paths(root_path, paths):
def fix_paths(root_path, paths):
return [os.path.join(".", os.path.relpath(path, root_path))
for path in paths]
def _change_background(str_, new_background):
def change_background(str_, new_background):
def change_background_style(style):
new_bg = (new_background if style.bg_color == termstr.Color.black
@ -165,24 +165,24 @@ def _change_background(str_, new_background):
return termstr.TermStr(str_).transform_style(change_background_style)
def _in_green(str_):
def in_green(str_):
return termstr.TermStr(str_, termstr.CharStyle(termstr.Color.green))
_UP, _DOWN, _LEFT, _RIGHT = (0, -1), (0, 1), (-1, 0), (1, 0)
def _directory_sort(path):
def directory_sort(path):
return (os.path.dirname(path), tools.splitext(path)[1],
os.path.basename(path))
def _type_sort(path):
def type_sort(path):
return (tools.splitext(path)[1], os.path.dirname(path),
os.path.basename(path))
def _log_filesystem_changed(log, added, removed, modified):
def log_filesystem_changed(log, added, removed, modified):
def part(stat, text, color):
return termstr.TermStr("%2s %s." % (stat, text)).fg_color(
termstr.Color.grey_100 if stat == 0 else color)
@ -192,7 +192,7 @@ def _log_filesystem_changed(log, added, removed, modified):
log.log_message("Filesystem changed: " + fill3.join(" ", parts))
def _get_diff_stats(old_files, new_files):
def get_diff_stats(old_files, new_files):
old_names = set(name for name, ctime in old_files)
new_names = set(name for name, ctime in new_files)
added_count = len(new_names - old_names)
@ -236,10 +236,8 @@ class Summary:
old_path = None
new_column = fill3.Column([])
new_cache = {}
paths = _fix_paths(self._root_path,
_codebase_files(self._root_path))
paths.sort(key=_directory_sort if self.is_directory_sort
else _type_sort)
paths = fix_paths(self._root_path, codebase_files(self._root_path))
paths.sort(key=directory_sort if self.is_directory_sort else type_sort)
jobs_added = False
new_cursor_position = (0, 0)
row_index = 0
@ -276,10 +274,10 @@ class Summary:
max_path_length = max(len(path) for path in paths) - len("./")
deleted_results = self._all_results - all_results
if log is not None:
stats = _get_diff_stats(
stats = get_diff_stats(
set(self._cache.keys()), set(new_cache.keys()))
if sum(stats) != 0:
_log_filesystem_changed(log, *stats)
log_filesystem_changed(log, *stats)
self._column, self._cache, self._cursor_position, self.result_total, \
self.completed_total, self._max_width, self._max_path_length, \
self.closest_placeholder_generator, self._all_results = (
@ -336,7 +334,7 @@ class Summary:
scroll_y = (screen_y // height) * height
self._view_widget.position = ((screen_x // width) * width, scroll_y)
appearance = self._view_widget.appearance(dimensions)
appearance[screen_y - scroll_y] = _change_background(
appearance[screen_y - scroll_y] = change_background(
appearance[screen_y - scroll_y], termstr.Color.grey_50)
return appearance
@ -412,9 +410,9 @@ class Summary:
tool_name = tools.tool_name_colored(
selection.tool, selection.path)
path_colored = tools.path_colored(selection.path)
log.log_message([_in_green("Refreshing "), tool_name,
_in_green(" result of "), path_colored,
_in_green("...")])
log.log_message([in_green("Refreshing "), tool_name,
in_green(" result of "), path_colored,
in_green("...")])
selection.reset()
self.closest_placeholder_generator = None
self._jobs_added_event.set()
@ -425,7 +423,7 @@ class Log:
_GREY_BOLD_STYLE = termstr.CharStyle(termstr.Color.grey_100, is_bold=True)
_GREEN_STYLE = termstr.CharStyle(termstr.Color.green)
_LOG_PATH = os.path.join(tools.CACHE_PATH, "log")
LOG_PATH = os.path.join(tools.CACHE_PATH, "log")
def __init__(self, appearance_changed_event):
self._appearance_changed_event = appearance_changed_event
@ -444,7 +442,7 @@ class Log:
if timestamp is None else timestamp)
line = termstr.TermStr(timestamp, Log._GREY_BOLD_STYLE) + " " + message
self.widget.append(fill3.Text(line))
with open(Log._LOG_PATH, "a") as log_file:
with open(Log.LOG_PATH, "a") as log_file:
print(line, file=log_file)
self.widget.widgets = self.widget[-200:]
self._appearance_cache = None
@ -455,7 +453,7 @@ class Log:
def delete_log_file(self):
with contextlib.suppress(FileNotFoundError):
os.remove(Log._LOG_PATH)
os.remove(Log.LOG_PATH)
def appearance_min(self):
appearance = self._appearance_cache
@ -470,20 +468,20 @@ class Log:
return self.portal.appearance(dimensions)
def _highlight_chars(str_, style, marker="*"):
def highlight_chars(str_, style, marker="*"):
parts = str_.split(marker)
highlighted_parts = [termstr.TermStr(part[0], style) + part[1:]
for part in parts[1:] if part != ""]
return fill3.join("", [parts[0]] + highlighted_parts)
def _get_status_help():
def get_status_help():
return fill3.join("\n", ["Statuses:"] +
[" " + tools.status_to_str(status) + " " + meaning
for status, meaning in tools.STATUS_MEANINGS])
def _make_key_map(key_data):
def make_key_map(key_data):
key_map = {}
for keys, action in key_data:
for key in keys:
@ -496,12 +494,12 @@ class Help:
def __init__(self, summary, screen):
self.summary = summary
self.screen = screen
keys_doc = _highlight_chars(KEYS_DOC, Log._GREEN_STYLE)
help_text = fill3.join("\n", [__doc__, keys_doc, _get_status_help()])
keys_doc = highlight_chars(KEYS_DOC, Log._GREEN_STYLE)
help_text = fill3.join("\n", [__doc__, keys_doc, get_status_help()])
self.view = fill3.View.from_widget(fill3.Text(help_text))
self.widget = fill3.Border(self.view, title="Help")
portal = self.view.portal
self.key_map = _make_key_map([
self.key_map = make_key_map([
({"h"}, self._exit_help), ({"d", "up"}, portal.scroll_up),
({"c", "down"}, portal.scroll_down),
({"j", "left"}, portal.scroll_left),
@ -558,7 +556,7 @@ class Screen:
self._is_help_visible = False
self._is_paused = False
self._make_widgets()
self._key_map = _make_key_map(Screen._KEY_DATA)
self._key_map = make_key_map(Screen._KEY_DATA)
def make_workers(self, worker_count, sandbox, is_being_tested):
workers = []
@ -684,8 +682,8 @@ class Screen:
else:
path = self._summary.get_selection().path
path_colored = tools.path_colored(path)
self._log.log_message([_in_green("Editing "), path_colored,
_in_green(' with command: "%s"...'
self._log.log_message([in_green("Editing "), path_colored,
in_green(' with command: "%s"...'
% self.editor_command)])
subprocess.Popen("%s %s" % (self.editor_command, path), shell=True,
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
@ -766,7 +764,7 @@ class Screen:
action(self)
self._appearance_changed_event.set()
_STATUS_BAR = _highlight_chars(
_STATUS_BAR = highlight_chars(
" *help *quit *d,*c,*j,*k,*f,*v:navigate *turn *log *edit *next *pause"
" *order *refresh", Log._GREEN_STYLE)
@ -822,7 +820,7 @@ class Screen:
({"p"}, toggle_pause), ({"r"}, refresh)]
def _add_watch_manager_to_mainloop(root_path, mainloop, on_filesystem_change,
def add_watch_manager_to_mainloop(root_path, mainloop, on_filesystem_change,
exclude_filter):
watch_manager = pyinotify.WatchManager()
event_mask = (pyinotify.IN_CREATE | pyinotify.IN_DELETE |
@ -853,14 +851,14 @@ def make_sandbox():
return sandbox
def _remove_sandbox(sandbox):
def remove_sandbox(sandbox):
cache_path = os.path.join(os.getcwd(), tools.CACHE_PATH)
subprocess.check_call(["sudo", "umount", sandbox.mount_point + cache_path])
sandbox.umount()
os.rmdir(sandbox.mount_point)
def _load_state(pickle_path, jobs_added_event, appearance_changed_event,
def load_state(pickle_path, jobs_added_event, appearance_changed_event,
root_path, loop):
is_first_run = True
try:
@ -883,7 +881,7 @@ def _load_state(pickle_path, jobs_added_event, appearance_changed_event,
return summary, screen, log, is_first_run
def _save_state(summary, screen, log, pickle_path):
def save_state(pickle_path, summary, screen, log):
# Cannot pickle generators, locks, sockets or events.
(summary.closest_placeholder_generator, summary._lock,
summary._jobs_added_event, screen._appearance_changed_event,
@ -900,8 +898,9 @@ def main(root_path, loop, worker_count=None, is_sandboxed=True,
pickle_path = os.path.join(tools.CACHE_PATH, "summary.pickle")
jobs_added_event = asyncio.Event()
appearance_changed_event = threading.Event()
summary, screen, log, is_first_run = _load_state(pickle_path,
jobs_added_event, appearance_changed_event, root_path, loop)
summary, screen, log, is_first_run = load_state(
pickle_path, jobs_added_event, appearance_changed_event, root_path,
loop)
screen.editor_command = editor_command
log.delete_log_file()
log.log_message("Program started.")
@ -912,14 +911,15 @@ def main(root_path, loop, worker_count=None, is_sandboxed=True,
def on_filesystem_change():
summary.sync_with_filesystem(log)
appearance_changed_event.set()
watch_manager_fd = _add_watch_manager_to_mainloop(
root_path, loop, on_filesystem_change, _is_path_excluded)
watch_manager_fd = add_watch_manager_to_mainloop(
root_path, loop, on_filesystem_change, is_path_excluded)
if is_sandboxed:
log.log_message("Making sandbox...")
sandbox = make_sandbox() if is_sandboxed else None
try:
log.log_message("Starting workers (%s) ..." % worker_count)
screen.make_workers(worker_count, sandbox, is_being_tested)
def exit_loop():
log.log_command("Exiting...")
time.sleep(0.05)
@ -929,13 +929,13 @@ def main(root_path, loop, worker_count=None, is_sandboxed=True,
log.log_message("Program stopped.")
finally:
if is_sandboxed:
_remove_sandbox(sandbox)
remove_sandbox(sandbox)
loop.remove_reader(watch_manager_fd)
_save_state(summary, screen, log, pickle_path)
save_state(pickle_path, summary, screen, log)
@contextlib.contextmanager
def _chdir(path):
def chdir(path):
old_cwd = os.getcwd()
os.chdir(path)
try:
@ -944,7 +944,7 @@ def _chdir(path):
os.chdir(old_cwd)
def _manage_cache(root_path):
def manage_cache(root_path):
cache_path = os.path.join(root_path, tools.CACHE_PATH)
timestamp_path = os.path.join(cache_path, "creation_time")
if os.path.exists(cache_path) and \
@ -957,12 +957,12 @@ def _manage_cache(root_path):
open(timestamp_path, "w").close()
def _check_arguments():
def check_arguments():
cmdline_help = __doc__ + USAGE.replace("*", "")
arguments = docopt.docopt(cmdline_help, help=False)
if arguments["--help"]:
print(cmdline_help)
print(_get_status_help())
print(get_status_help())
sys.exit(0)
worker_count = None
try:
@ -991,12 +991,12 @@ def _check_arguments():
if __name__ == "__main__":
root_path, worker_count, is_sandboxed, editor_command = _check_arguments()
root_path, worker_count, is_sandboxed, editor_command = check_arguments()
subprocess.call(["sudo", "-p", "Vigil uses sudo... "
"[sudo] password for %u: ", "true"])
with terminal.console_title("vigil: " + os.path.basename(root_path)):
_manage_cache(root_path)
with _chdir(root_path): # FIX: Don't change directory if possible.
manage_cache(root_path)
with chdir(root_path): # FIX: Don't change directory if possible.
loop = asyncio.get_event_loop()
main(root_path, loop, worker_count, is_sandboxed, editor_command)
loop.close()

4
vigil_test.py
View file

@ -225,8 +225,8 @@ class MainTestCase(unittest.TestCase):
# processes = _all_processes()
foo_path = os.path.join(root_path, "foo")
open(foo_path, "w").close()
vigil._manage_cache(root_path)
with vigil._chdir(root_path):
vigil.manage_cache(root_path)
with vigil.chdir(root_path):
with contextlib.redirect_stdout(io.StringIO()):
vigil.main(root_path, loop, worker_count=2,
is_sandboxed=True, is_being_tested=True)