vaclavt/vdiff
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

type hints across vdiff

Browse Source
This commit is contained in:
Vaclav Tvrdik
2026-09-29 14:59:56 +02:00
parent 54c69484d5
commit d6f0ed3009
10 changed files with 339 additions and 230 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
TODO.txt
View File

@@ -16,4 +16,3 @@
## Code
- add type hints

4
tests/stubs/sublime.py
View File

@@ -36,6 +36,10 @@ class Region:
return f"Region({self.a}, {self.b})"
class Edit:
"""Opaque edit token; only ever passed through, so the annotations have something to name"""
class Settings(dict):
def get(self, key, default=None):
return dict.get(self, key, default)

143
vdiff/commands.py
View File

@@ -1,15 +1,21 @@
""" Commands: the two quick panels (changed files, recent commits) and the actions of a review view """
import bisect
import os
from typing import Callable, Dict, List, Optional, Sequence, Tuple
import sublime
import sublime_plugin
from . import git, parse, render, state
from .git import Commit, FileChange
from .state import Kept, ViewState
from .utils import file_count, run_async, settings, status
# what an ACTIONS entry does: act(view, window, state) for the review view it was fired in
Action = Callable[[sublime.View, sublime.Window, ViewState], None]
# quick-panel kind per status letter: colour + the letter; the description comes from git.STATUS_NAMES
STATUS_COLOURS = {
STATUS_COLOURS: Dict[str, int] = {
"M": sublime.KIND_ID_COLOR_ORANGISH,
"A": sublime.KIND_ID_COLOR_GREENISH,
"?": sublime.KIND_ID_COLOR_GREENISH,
@@ -30,20 +36,22 @@ MIN_SUBJECT_WIDTH = 24
PREVIEW_DELAY_MS = 60
# directory -> repository root; roots do not move, and resolving runs on the UI thread
_roots = {}
_roots: Dict[str, str] = {}
def resolve_repo(window):
def resolve_repo(window: sublime.Window) -> Optional[str]:
"""Repository of the review view or active file, else of the first project folder that is one"""
view = window.active_view()
candidates = []
candidates: List[str] = []
if view is not None:
if state.is_review_view(view):
st = state.get(view) or restore_view(view)
if st.repo:
return st.repo
elif view.file_name():
candidates.append(os.path.dirname(view.file_name()))
else:
name = view.file_name()
if name:
candidates.append(os.path.dirname(name))
candidates.extend(window.folders())
for directory in candidates:
root = _roots.get(directory)
@@ -56,14 +64,14 @@ def resolve_repo(window):
return None
def find_review_view(window):
def find_review_view(window: sublime.Window) -> Optional[sublime.View]:
for view in window.views():
if state.is_review_view(view):
return view
return None
def open_review(window):
def open_review(window: sublime.Window) -> Optional[Tuple[sublime.View, ViewState]]:
"""(view, state) of the window's single review view for the resolved repo, focused; None without a repo"""
repo = resolve_repo(window)
if not repo:
@@ -81,7 +89,7 @@ def open_review(window):
return view, st
def show_file(view, st, change):
def show_file(view: sublime.View, st: ViewState, change: FileChange) -> None:
"""Fetch and parse the file's diff on the worker thread, then render it"""
st.token += 1
# set before the worker starts: n / p count from here while the diff is still loading
@@ -89,14 +97,14 @@ def show_file(view, st, change):
token = st.token
repo, sha = st.repo, st.sha
def work():
def work() -> Optional[Tuple[parse.Diff, List[parse.Block]]]:
if token != st.token:
return None # superseded before it started: skip the subprocess
text = git.full_diff(repo, change.path, sha=sha, untracked=(change.status == "?"))
diff = parse.parse_unified(text)
return diff, parse.change_blocks(diff.lines)
def done(result):
def done(result: Optional[Tuple[parse.Diff, List[parse.Block]]]) -> None:
if result is None or token != st.token or not view.is_valid():
return
diff, blocks = result
@@ -108,8 +116,8 @@ def show_file(view, st, change):
run_async(work, done)
def file_items(files):
items = []
def file_items(files: Sequence[FileChange]) -> List[sublime.QuickPanelItem]:
items: List[sublime.QuickPanelItem] = []
for change in files:
description = git.STATUS_NAMES.get(change.status, "")
colour = STATUS_COLOURS.get(change.status)
@@ -125,9 +133,9 @@ def file_items(files):
return items
def commit_items(commits):
def commit_items(commits: Sequence[Commit]) -> List[sublime.QuickPanelItem]:
"""Quick-panel rows: subject on the left, its totals and the hash right-aligned"""
items = []
items: List[sublime.QuickPanelItem] = []
for c in commits:
annotation = c.short
if c.files is not None:
@@ -141,13 +149,15 @@ def commit_items(commits):
return items
def show_files_panel(window, view, st, files, kept, on_cancel=None, placeholder="Changed files"):
def show_files_panel(window: sublime.Window, view: sublime.View, st: ViewState, files: List[FileChange],
kept: Kept, on_cancel: Optional[Callable[[], None]] = None,
placeholder: str = "Changed files") -> None:
"""Quick panel over `files`; highlighting previews, Enter keeps, Esc restores `kept` (or runs on_cancel)"""
st.files = files
selected = max(0, st.index())
window.focus_view(view)
def on_select(index):
def on_select(index: int) -> None:
st.highlight += 1 # cancel a pending preview
if index == -1:
if on_cancel is not None:
@@ -164,17 +174,23 @@ def show_files_panel(window, view, st, files, kept, on_cancel=None, placeholder=
return # the preview already shows it
show_file(view, st, change)
def on_highlight(index):
def on_highlight(index: int) -> None:
if not 0 <= index < len(files):
return
st.highlight += 1
ticket = st.highlight
sublime.set_timeout(lambda: ticket == st.highlight and show_file(view, st, files[index]), PREVIEW_DELAY_MS)
def preview() -> None:
if ticket == st.highlight:
show_file(view, st, files[index])
sublime.set_timeout(preview, PREVIEW_DELAY_MS)
window.show_quick_panel(file_items(files), on_select, 0, selected, on_highlight, placeholder)
def refresh_listing(view, st, message=None, pick=None):
def refresh_listing(view: sublime.View, st: ViewState, message: Optional[str] = None,
pick: Optional[Callable[[List[FileChange], int], Optional[FileChange]]] = None) -> None:
"""Re-list the working tree changes and show the same file, or the next one when it is gone
`pick(files, index)` may name another file to show instead; `index` is where the shown file
@@ -185,7 +201,7 @@ def refresh_listing(view, st, message=None, pick=None):
previous_index = max(0, st.index())
path = st.change.path if st.change else None
def done(files):
def done(files: List[FileChange]) -> None:
if not view.is_valid():
return
st.files = files
@@ -202,7 +218,7 @@ def refresh_listing(view, st, message=None, pick=None):
run_async(lambda: git.changed_files(st.repo), done)
def refresh_if_changed(view, st):
def refresh_if_changed(view: sublime.View, st: ViewState) -> None:
"""Re-render a working-tree view when the file or the index changed since it was drawn"""
if st.in_commit or st.change is None:
return
@@ -213,14 +229,14 @@ def refresh_if_changed(view, st):
class VdiffFilesCommand(sublime_plugin.WindowCommand):
"""List uncommitted changes (staged, unstaged, untracked) in a quick panel"""
def run(self):
def run(self) -> None:
window = self.window
opened = open_review(window)
if not opened:
return
view, st = opened
def done(files):
def done(files: List[FileChange]) -> None:
if not view.is_valid():
return
kept = st.enter()
@@ -237,7 +253,7 @@ class VdiffFilesCommand(sublime_plugin.WindowCommand):
class VdiffCommitsCommand(sublime_plugin.WindowCommand):
"""List recent commits, then the files of the chosen commit; `sha` skips straight to one commit"""
def run(self, selected=0, sha=None):
def run(self, selected: int = 0, sha: Optional[str] = None) -> None:
opened = open_review(self.window)
if not opened:
return
@@ -247,7 +263,7 @@ class VdiffCommitsCommand(sublime_plugin.WindowCommand):
return
count = int(settings().get("commits_count", 30) or 30)
def done(commits):
def done(commits: List[Commit]) -> None:
if not view.is_valid():
return
if not commits:
@@ -257,23 +273,23 @@ class VdiffCommitsCommand(sublime_plugin.WindowCommand):
run_async(lambda: git.recent_commits(st.repo, count), done)
def show_commits(self, view, st, commits, selected):
def show_commits(self, view: sublime.View, st: ViewState, commits: Sequence[Commit], selected: int) -> None:
items = commit_items(commits)
for i, c in enumerate(commits):
if c.sha == st.sha:
selected = i
def on_select(index):
def on_select(index: int) -> None:
if index != -1:
self.open_commit(view, st, commits[index].sha, index)
self.window.show_quick_panel(items, on_select, 0, selected, None, "Recent commits")
def open_commit(self, view, st, sha, index):
def open_commit(self, view: sublime.View, st: ViewState, sha: str, index: Optional[int]) -> None:
"""Files panel of one commit; Esc goes back to the commit list when it came from there (index given)"""
window = self.window
def done(result):
def done(result: Tuple[git.CommitInfo, List[FileChange]]) -> None:
if not view.is_valid():
return
info, files = result
@@ -282,7 +298,7 @@ class VdiffCommitsCommand(sublime_plugin.WindowCommand):
render.render_message(view, st, f"Commit {info.short} touches no files")
return
def back():
def back() -> None:
window.run_command("vdiff_commits", {"selected": index})
placeholder = f"{info.short} · {info.subject}"
@@ -295,29 +311,31 @@ class VdiffCommitsCommand(sublime_plugin.WindowCommand):
# --- review view actions ----------------------------------------------------
def _act_files(view, window, st):
def _act_files(view: sublime.View, window: sublime.Window, st: ViewState) -> None:
window.run_command("vdiff_files")
def _act_commits(view, window, st):
def _act_commits(view: sublime.View, window: sublime.Window, st: ViewState) -> None:
window.run_command("vdiff_commits")
def _act_close(view, window, st):
def _act_close(view: sublime.View, window: sublime.Window, st: ViewState) -> None:
view.close()
def _act_help(view, window, st):
def _act_help(view: sublime.View, window: sublime.Window, st: ViewState) -> None:
render.show_help(view, st)
def _git_action(verb, done_verb, operation, confirm=None, pick=None):
def _git_action(verb: str, done_verb: str, operation: Callable[[str, FileChange], None],
confirm: Optional[Callable[[FileChange], bool]] = None,
pick: Optional[Callable[[List[FileChange], int], Optional[FileChange]]] = None) -> Action:
"""Action running `operation(repo, change)` on the shown working-tree file, then re-listing
`pick` is handed to refresh_listing to choose which file to show afterwards.
"""
def act(view, window, st):
def act(view: sublime.View, window: sublime.Window, st: ViewState) -> None:
if st.in_commit or st.change is None:
status(f"{verb} works on uncommitted changes only")
return
@@ -332,7 +350,7 @@ def _git_action(verb, done_verb, operation, confirm=None, pick=None):
return act
def _confirm_discard(change):
def _confirm_discard(change: FileChange) -> bool:
if not settings().get("confirm_discard", True):
return True
if change.status == "?":
@@ -345,7 +363,7 @@ def _confirm_discard(change):
return sublime.ok_cancel_dialog(message, "Discard")
def _next_unstaged(files, start):
def _next_unstaged(files: Sequence[FileChange], start: int) -> int:
"""Index of the first unstaged file from `start` on, wrapping round; -1 when every file is staged"""
for offset in range(len(files)):
index = (start + offset) % len(files)
@@ -354,7 +372,7 @@ def _next_unstaged(files, start):
return -1
def _stage_next(files, index):
def _stage_next(files: Sequence[FileChange], index: int) -> Optional[FileChange]:
"""After staging, the next unstaged file to show; None keeps the file the refresh would show"""
if not settings().get("stage_and_move_next_file", True):
return None
@@ -363,7 +381,7 @@ def _stage_next(files, index):
return files[following] if following >= 0 else None
def _act_refresh(view, window, st):
def _act_refresh(view: sublime.View, window: sublime.Window, st: ViewState) -> None:
if st.change is None:
return
if st.in_commit:
@@ -372,7 +390,7 @@ def _act_refresh(view, window, st):
refresh_listing(view, st)
def _act_open(view, window, st):
def _act_open(view: sublime.View, window: sublime.Window, st: ViewState) -> None:
if st.change is None:
return
path = os.path.join(st.repo, st.change.path)
@@ -383,8 +401,8 @@ def _act_open(view, window, st):
window.open_file(f"{path}:{line}", sublime.ENCODED_POSITION)
def _step_file(delta):
def act(view, window, st):
def _step_file(delta: int) -> Action:
def act(view: sublime.View, window: sublime.Window, st: ViewState) -> None:
if not st.files:
status("no file list, press f or c")
return
@@ -406,8 +424,8 @@ def _step_file(delta):
return act
def _step_change(delta):
def act(view, window, st):
def _step_change(delta: int) -> Action:
def act(view: sublime.View, window: sublime.Window, st: ViewState) -> None:
starts = [start for start, _ in st.blocks]
if not starts:
status("no changes to step through")
@@ -425,7 +443,7 @@ def _step_change(delta):
return act
def _act_commit(view, window, st):
def _act_commit(view: sublime.View, window: sublime.Window, st: ViewState) -> None:
"""Commit the staged files with a message typed into an input panel; Esc there cancels"""
if st.in_commit:
status("commit works on uncommitted changes only")
@@ -435,7 +453,7 @@ def _act_commit(view, window, st):
status("no staged files")
return
def on_done(message):
def on_done(message: str) -> None:
if not message.strip():
status("commit cancelled: empty message")
return
@@ -447,7 +465,7 @@ def _act_commit(view, window, st):
window.show_input_panel("Commit message", "", on_done, None, None)
def _act_amend(view, window, st):
def _act_amend(view: sublime.View, window: sublime.Window, st: ViewState) -> None:
"""Rewrite HEAD's message in an input panel, folding in whatever is staged; Esc there cancels
The panel is one line, so only the subject is offered; the rest of the message rides along
@@ -458,10 +476,10 @@ def _act_amend(view, window, st):
return
staged = [c for c in st.files if c.staged]
def prompt(message):
def prompt(message: str) -> None:
subject, newline, body = message.partition("\n")
def on_done(text):
def on_done(text: str) -> None:
if not text.strip():
status("amend cancelled: empty message")
return
@@ -482,14 +500,14 @@ def _act_amend(view, window, st):
run_async(lambda: git.head_message(st.repo), prompt)
def _move_line(forward):
def act(view, window, st):
def _move_line(forward: bool) -> Action:
def act(view: sublime.View, window: sublime.Window, st: ViewState) -> None:
view.run_command("move", {"by": "lines", "forward": forward})
return act
ACTIONS = {
ACTIONS: Dict[str, Action] = {
"files": _act_files,
"commits": _act_commits,
"close": _act_close,
@@ -513,10 +531,10 @@ ACTIONS = {
class VdiffActionCommand(sublime_plugin.TextCommand):
"""Every key/link action of a review view, dispatched by name through ACTIONS"""
def is_enabled(self, action=None):
def is_enabled(self, action: Optional[str] = None) -> bool:
return state.is_review_view(self.view)
def run(self, edit, action):
def run(self, edit: sublime.Edit, action: str) -> None:
act = ACTIONS.get(action)
if act is None:
status(f"unknown action {action}")
@@ -525,7 +543,7 @@ class VdiffActionCommand(sublime_plugin.TextCommand):
act(self.view, self.view.window(), st)
def restore_view(view):
def restore_view(view: sublime.View) -> ViewState:
"""Rebuild the state of a review view restored from a previous session; re-renders asynchronously"""
data = view.settings().get(state.VIEW_DATA) or {}
repo = data.get("repo")
@@ -542,16 +560,17 @@ def restore_view(view):
return st
change = git.FileChange(path, data.get("status", "M"), bool(data.get("staged")), None, None, None)
if st.in_commit:
sha = st.sha
if sha is not None:
def done(result):
def show_commit(result: Tuple[git.CommitInfo, List[FileChange]]) -> None:
st.commit, st.files = result
show_file(view, st, change)
run_async(lambda: git.commit_with_files(repo, st.sha), done)
run_async(lambda: git.commit_with_files(repo, sha), show_commit)
else:
def done(files):
def show_working_tree(files: List[FileChange]) -> None:
st.files = files
index = state.find_change(files, path)
if index >= 0:
@@ -559,5 +578,5 @@ def restore_view(view):
else:
render.render_message(view, st, f"{path} has no uncommitted changes any more")
run_async(lambda: git.changed_files(repo), done)
run_async(lambda: git.changed_files(repo), show_working_tree)
return st

109
vdiff/git.py
View File

@@ -2,13 +2,44 @@
import os
import re
import subprocess
from collections import namedtuple
from typing import Dict, List, Optional, NamedTuple, Sequence, Tuple
# path, one-letter status (M A D R ?), staged flag, numstat counts (None for binary), rename source
FileChange = namedtuple("FileChange", "path status staged insertions deletions old_path")
# shortstat totals; all three are None when git prints no diffstat (merges, empty commits)
Commit = namedtuple("Commit", "sha short author date subject files insertions deletions")
CommitInfo = namedtuple("CommitInfo", "sha short author date subject body")
class FileChange(NamedTuple):
"""path, one-letter status (M A D R ?), staged flag, numstat counts (None for binary), rename source"""
path: str
status: str
staged: bool
insertions: Optional[int]
deletions: Optional[int]
old_path: Optional[str]
class Commit(NamedTuple):
"""shortstat totals; all three are None when git prints no diffstat (merges, empty commits)"""
sha: str
short: str
author: str
date: str
subject: str
files: Optional[int]
insertions: Optional[int]
deletions: Optional[int]
class CommitInfo(NamedTuple):
sha: str
short: str
author: str
date: str
subject: str
body: str
# a numstat entry: insertions, deletions (both None for a binary file) and the rename source
Numstat = Tuple[Optional[int], Optional[int], Optional[str]]
# hash of the empty tree: the diff base for a repository without commits
EMPTY_TREE = "4b825dc642cb6eb9a060e54bf8d69288fbee4904"
@@ -17,9 +48,9 @@ EMPTY_TREE = "4b825dc642cb6eb9a060e54bf8d69288fbee4904"
FULL_CONTEXT = "-U999999"
# overridable from settings (sublime's PATH may not have a user-installed git)
GIT_BINARY = "git"
GIT_BINARY: str = "git"
STATUS_NAMES = {
STATUS_NAMES: Dict[str, str] = {
"M": "modified",
"A": "added",
"D": "deleted",
@@ -40,14 +71,14 @@ _SHORTSTAT = re.compile(r"\s*(\d+) files? changed(?:, (\d+) insertions?\(\+\))?(
class GitError(Exception):
"""git exited with an unexpected code; str() is the first stderr line, `stderr` the whole of it"""
def __init__(self, stderr, code):
def __init__(self, stderr: str, code: int) -> None:
self.stderr = stderr
self.code = code
first = stderr.strip().splitlines()
super().__init__(first[0] if first else f"git exited with {code}")
def run(repo, args, ok_codes=(0,)):
def run(repo: str, args: Sequence[str], ok_codes: Tuple[int, ...] = (0,)) -> str:
"""Run git in `repo`, return stdout as text"""
cmd = [GIT_BINARY, "-c", "core.quotepath=off", "--no-pager"] + list(args)
proc = subprocess.run(cmd, cwd=repo, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding="utf-8", errors="replace")
@@ -56,7 +87,7 @@ def run(repo, args, ok_codes=(0,)):
return proc.stdout
def repo_root(path):
def repo_root(path: Optional[str]) -> Optional[str]:
"""Top-level directory of the repository containing `path`, or None"""
if not path:
return None
@@ -70,7 +101,7 @@ def repo_root(path):
return root or None
def has_head(repo):
def has_head(repo: str) -> bool:
"""False in a repository without commits"""
try:
run(repo, ["rev-parse", "--verify", "-q", "HEAD"])
@@ -79,19 +110,19 @@ def has_head(repo):
return False
def base_ref(repo):
def base_ref(repo: str) -> str:
"""HEAD, or the empty tree when there is no commit yet"""
return "HEAD" if has_head(repo) else EMPTY_TREE
def _int_or_none(value):
def _int_or_none(value: str) -> Optional[int]:
"""A numstat count: "-" marks a binary file"""
return None if value == "-" else int(value)
def _parse_numstat(text):
def _parse_numstat(text: str) -> Dict[str, Numstat]:
"""`--numstat -z` output -> {path: (insertions, deletions, old_path)}"""
result = {}
result: Dict[str, Numstat] = {}
fields = iter(text.split("\0"))
for entry in fields:
parts = entry.split("\t")
@@ -107,9 +138,9 @@ def _parse_numstat(text):
return result
def _parse_name_status(text):
def _parse_name_status(text: str) -> List[Tuple[str, str, Optional[str]]]:
"""`--name-status -z` output -> [(status letter, path, old_path)]"""
result = []
result: List[Tuple[str, str, Optional[str]]] = []
fields = iter(text.split("\0"))
for status in fields:
if not status:
@@ -122,7 +153,7 @@ def _parse_name_status(text):
return result
def _count_lines(path):
def _count_lines(path: str) -> Optional[int]:
"""Line count of a working-tree file, None when binary or unreadable"""
try:
with open(path, "rb") as f:
@@ -144,7 +175,7 @@ def _count_lines(path):
return count + (0 if last == b"\n" else 1)
def changed_files(repo):
def changed_files(repo: str) -> List[FileChange]:
"""Working tree changes vs HEAD: staged, unstaged and untracked, sorted by path"""
status = run(repo, ["status", "--porcelain=v2", "-z", "--branch", "--untracked-files=all", "--no-renames"])
entries = [e for e in status.split("\0") if e]
@@ -152,7 +183,7 @@ def changed_files(repo):
base = EMPTY_TREE if "# branch.oid (initial)" in entries else "HEAD"
counts = _parse_numstat(run(repo, ["diff", "--numstat", "-z", "--no-renames", base]))
changes = []
changes: List[FileChange] = []
for entry in entries:
kind = entry[0]
if kind == "1":
@@ -177,26 +208,28 @@ def changed_files(repo):
return changes
def recent_commits(repo, count):
def recent_commits(repo: str, count: int) -> List[Commit]:
"""Last `count` commits, newest first; empty list without commits"""
if not has_head(repo):
return []
# --shortstat rides along on the same call: one subprocess, totals for free
text = run(repo, ["log", "-n", str(count), "--format=%H%x1f%h%x1f%an%x1f%ar%x1f%s", "--shortstat"])
commits = []
commits: List[Commit] = []
for line in text.split("\n"):
parts = line.split("\x1f")
if len(parts) == 5:
commits.append(Commit(*parts, files=None, insertions=None, deletions=None))
sha, short, author, date, subject = parts
commits.append(Commit(sha, short, author, date, subject, files=None, insertions=None, deletions=None))
continue
stat = commits and commits[-1].files is None and _SHORTSTAT.match(line)
if stat:
files, ins, dele = (int(g or 0) for g in stat.groups())
commits[-1] = commits[-1]._replace(files=files, insertions=ins, deletions=dele)
if commits and commits[-1].files is None:
stat = _SHORTSTAT.match(line)
if stat:
files, ins, dele = (int(g or 0) for g in stat.groups())
commits[-1] = commits[-1]._replace(files=files, insertions=ins, deletions=dele)
return commits
def commit_with_files(repo, sha):
def commit_with_files(repo: str, sha: str) -> Tuple[CommitInfo, List[FileChange]]:
"""(CommitInfo, [FileChange]) of one commit vs its first parent, files sorted by path; two git calls"""
show = ["show", "--first-parent", "-m", "-z"]
text = run(repo, show + [f"--format={_INFO_FORMAT}%x00", _DATE_FORMAT, "--name-status", sha])
@@ -209,7 +242,7 @@ def commit_with_files(repo, sha):
info = CommitInfo(*parts[:6])
counts = _parse_numstat(run(repo, show + ["--format=", "--numstat", sha]))
changes = []
changes: List[FileChange] = []
for status, path, old_path in _parse_name_status(files_text):
ins, dele, _ = counts.get(path, (None, None, None))
changes.append(FileChange(path, status, False, ins, dele, old_path))
@@ -217,7 +250,7 @@ def commit_with_files(repo, sha):
return info, changes
def _in_tree(repo, ref, path):
def _in_tree(repo: str, ref: str, path: str) -> bool:
"""Whether `path` exists in the tree of `ref`"""
try:
run(repo, ["cat-file", "-e", f"{ref}:{path}"])
@@ -226,7 +259,7 @@ def _in_tree(repo, ref, path):
return False
def full_diff(repo, path, sha=None, untracked=False):
def full_diff(repo: str, path: str, sha: Optional[str] = None, untracked: bool = False) -> str:
"""Unified diff of one file with the whole file as context
Working tree vs HEAD (staged and unstaged merged) by default; a commit vs its
@@ -240,22 +273,22 @@ def full_diff(repo, path, sha=None, untracked=False):
return run(repo, ["diff"] + common + [base_ref(repo), "--", path])
def stage(repo, change):
def stage(repo: str, change: FileChange) -> None:
"""Stage the whole file, including its deletion"""
run(repo, ["add", "-A", "--", change.path])
def commit(repo, message):
def commit(repo: str, message: str) -> None:
"""Commit what is staged; the message is one argv element, so it needs no quoting of its own"""
run(repo, ["commit", "-m", message])
def head_message(repo):
def head_message(repo: str) -> str:
"""HEAD's whole message, exactly as git stores it"""
return run(repo, ["log", "-1", "--format=%B"]).strip("\n")
def amend(repo, message, allow_empty=False):
def amend(repo: str, message: str, allow_empty: bool = False) -> None:
"""Rewrite HEAD with `message`, folding in whatever is staged
`allow_empty` for a pure reword: git otherwise refuses an amend whose result carries no change,
@@ -264,7 +297,7 @@ def amend(repo, message, allow_empty=False):
run(repo, ["commit", "--amend"] + (["--allow-empty"] if allow_empty else []) + ["-m", message])
def unstage(repo, change):
def unstage(repo: str, change: FileChange) -> None:
"""Move the file's staged changes back to the working tree"""
if has_head(repo):
run(repo, ["reset", "-q", "--", change.path])
@@ -272,7 +305,7 @@ def unstage(repo, change):
run(repo, ["rm", "-r", "-q", "--cached", "--", change.path])
def discard(repo, change):
def discard(repo: str, change: FileChange) -> None:
"""Throw away the file's changes, staged and unstaged, restoring the HEAD version"""
if change.status == "?":
run(repo, ["clean", "-f", "--", change.path])

6
vdiff/listeners.py
View File

@@ -7,15 +7,15 @@ from .commands import refresh_if_changed, restore_view
class VdiffEventListener(sublime_plugin.EventListener):
def on_pre_close(self, view):
def on_pre_close(self, view: sublime.View) -> None:
state.drop(view)
def on_activated(self, view):
def on_activated(self, view: sublime.View) -> None:
"""Re-render when the tab comes back and the file or index changed meanwhile (agent edits)"""
if state.is_review_view(view):
refresh_if_changed(view, state.get(view) or restore_view(view))
def on_post_save(self, view):
def on_post_save(self, view: sublime.View) -> None:
"""Review views showing the saved file re-render; the fingerprint decides which ones"""
for view_id, st in state.tracked():
review = sublime.View(view_id)

60
vdiff/palette.py
View File

@@ -11,6 +11,7 @@ probed and written in one pass rather than growing a colour at a time.
import json
import logging
import os
from typing import Any, Dict, List, Optional, Set, Tuple
import sublime
@@ -25,11 +26,11 @@ SCOPE_DEL_CHAR = "diff.deleted.char"
SCOPE_INS_CHAR = "diff.inserted.char"
# background kinds, and the tint each one draws over
KINDS = ("del", "ins", "del.char", "ins.char")
KIND_TINT = {"del": SCOPE_DEL, "ins": SCOPE_INS, "del.char": SCOPE_DEL_CHAR, "ins.char": SCOPE_INS_CHAR}
KINDS: Tuple[str, ...] = ("del", "ins", "del.char", "ins.char")
KIND_TINT: Dict[str, str] = {"del": SCOPE_DEL, "ins": SCOPE_INS, "del.char": SCOPE_DEL_CHAR, "ins.char": SCOPE_INS_CHAR}
# tints to provision when the active scheme defines none of its own (some third-party schemes)
FALLBACK_TINTS = {
FALLBACK_TINTS: Dict[str, str] = {
SCOPE_DEL: "hsla(357, 45%, 60%, 0.15)",
SCOPE_DEL_CHAR: "hsla(357, 60%, 60%, 0.30)",
SCOPE_INS: "hsla(114, 31%, 68%, 0.15)",
@@ -40,7 +41,7 @@ FALLBACK_TINTS = {
# one pass over them yields the working set in one write. There is no API that returns a scheme's own
# palette: style_for_scope resolves one scope at a time, and reading the scheme file would mean
# re-implementing sublime's merging of overrides and .tmTheme inheritance.
PALETTE_PROBES = (
PALETTE_PROBES: Tuple[str, ...] = (
"source",
"comment",
"string",
@@ -70,14 +71,14 @@ PALETTE_PROBES = (
)
# {override file name: {colour: index}} for every scheme provisioned this session
_palettes = {}
_palettes: Dict[str, Dict[str, int]] = {}
# the override name the last ensure() resolved, so scope_for knows which palette to read
_active = None
_active: Optional[str] = None
# schemes the reader has already been told need a restart
_notified = set()
_notified: Set[Optional[str]] = set()
def reset():
def reset() -> None:
"""Drop the caches, so the next ensure() re-reads and re-probes. Only the tests need this"""
global _active
_palettes.clear()
@@ -85,24 +86,24 @@ def reset():
_active = None
def claim_restart_notice(view):
def claim_restart_notice(view: sublime.View) -> bool:
"""Whether the reader still has to be told to restart, asked once per scheme
True only when a restart would actually help: the palette is provisioned but sublime has not
loaded it yet. A scheme we failed to provision at all is not something restarting fixes.
"""
if not _palettes.get(_active) or tokens_resolved(view) or _active in _notified:
if not (_active and _palettes.get(_active)) or tokens_resolved(view) or _active in _notified:
return False
_notified.add(_active)
return True
def scope_background(view, scope):
def scope_background(view: sublime.View, scope: str) -> Optional[str]:
"""The scheme's background for a scope, or None; style_for_scope can fail on a closing view"""
return _style(view, scope).get("background")
def _style(view, scope):
def _style(view: sublime.View, scope: str) -> Dict[str, Any]:
try:
return view.style_for_scope(scope)
except Exception:
@@ -110,47 +111,47 @@ def _style(view, scope):
return {}
def foreground(view, scope):
def foreground(view: sublime.View, scope: str) -> Optional[str]:
"""The scheme's foreground for a scope, normalised by sublime to #rrggbb"""
return _style(view, scope).get("foreground")
def tokens_resolved(view):
def tokens_resolved(view: sublime.View) -> bool:
"""Whether the generated (tint + colour) scopes are live
The token layer must never paint with a scope sublime does not know: it would draw that text in
the bright green it uses for an unknown scope. Sublime loads an override it has not seen before
only on a restart, so this stays false for the session that first wrote one.
"""
colours = _palettes.get(_active) or {}
colours = (_active and _palettes.get(_active)) or {}
if not colours:
return False
# they are all written by the same pass into the same file, so any one of them answers for all
return bool(scope_background(view, _scope_name(KINDS[0], next(iter(colours.values())))))
def scope_for(kind, colour):
def scope_for(kind: str, colour: str) -> Optional[str]:
"""The generated scope carrying `kind`'s background and `colour`, or None if never provisioned"""
index = _palettes.get(_active, {}).get(colour)
index = _palettes.get(_active or "", {}).get(colour)
return None if index is None else _scope_name(kind, index)
def token_key(scope):
def token_key(scope: str) -> str:
"""The add_regions key for a generated scope; one naming rule, spelled here only"""
return scope.replace("vdiff.", "vdiff.tok.", 1)
def token_keys():
def token_keys() -> List[str]:
"""Every region key the token layer can use, so a re-render can erase them all"""
colours = _palettes.get(_active, {})
colours = _palettes.get(_active or "", {})
return [token_key(_scope_name(kind, i)) for kind in KINDS for i in sorted(colours.values())]
def _scope_name(kind, index):
def _scope_name(kind: str, index: int) -> str:
return f"vdiff.{kind}.c{index}"
def ensure(view):
def ensure(view: sublime.View) -> None:
"""Provision the scopes for the active scheme, once per scheme per session
Whether the rules have taken effect is a separate question, asked by tokens_resolved(), because
@@ -163,7 +164,7 @@ def ensure(view):
return
path = os.path.join(sublime.packages_path(), "User", name)
try:
scheme = {}
scheme: Dict[str, Any] = {}
if os.path.exists(path):
with open(path, "r", encoding="utf-8") as f:
# a scheme the user has hand-edited may hold comments decode_value rejects; the raise
@@ -184,9 +185,9 @@ def ensure(view):
_palettes.setdefault(name, {})
def _probe_palette(view):
def _probe_palette(view: sublime.View) -> List[str]:
"""The distinct foregrounds the active scheme uses, in probe order so indices stay meaningful"""
seen = []
seen: List[str] = []
for scope in PALETTE_PROBES:
colour = foreground(view, scope)
if colour and colour not in seen:
@@ -194,13 +195,13 @@ def _probe_palette(view):
return seen
def _tints(view):
def _tints(view: sublime.View) -> Dict[str, str]:
"""{tint scope: background} for the scheme's diff colours, falling back to ours per colour pair
All or nothing per pair: a scheme that defines diff.deleted but not diff.deleted.char would
otherwise get its line tint next to our char tint, in a different hue.
"""
tints = {}
tints: Dict[str, str] = {}
for line_scope, char_scope in ((SCOPE_DEL, SCOPE_DEL_CHAR), (SCOPE_INS, SCOPE_INS_CHAR)):
line = scope_background(view, line_scope)
char = scope_background(view, char_scope)
@@ -209,7 +210,8 @@ def _tints(view):
return tints
def _merge_rules(rules, tints, colours, view):
def _merge_rules(rules: List[Dict[str, str]], tints: Dict[str, str], colours: Dict[str, int],
view: sublime.View) -> bool:
"""Bring the scheme's rules up to date, in place; True when anything changed
Each (background kind, palette colour) pair needs a rule carrying both, because add_regions draws
@@ -239,7 +241,7 @@ def _merge_rules(rules, tints, colours, view):
return changed
def override_name(view):
def override_name(view: sublime.View) -> Optional[str]:
"""File name in Packages/User that overrides the active scheme (same base name, .sublime-color-scheme)"""
try:
# already resolved through "auto" and dark/light settings by sublime

48
vdiff/parse.py
View File

@@ -1,12 +1,32 @@
""" Unified diff -> lines with old/new numbers, change blocks, word-level ranges; no sublime imports """
import difflib
import re
from collections import Counter, namedtuple
from collections import Counter
from itertools import groupby
from typing import Callable, Iterator, List, NamedTuple, Optional, Sequence, Tuple, TypeVar
# kind is " " (context), "-" (deleted) or "+" (inserted); the missing number is None
Line = namedtuple("Line", "kind old_no new_no text")
Diff = namedtuple("Diff", "lines binary")
class Line(NamedTuple):
"""kind is " " (context), "-" (deleted) or "+" (inserted); the missing number is None"""
kind: str
old_no: Optional[int]
new_no: Optional[int]
text: str
class Diff(NamedTuple):
lines: List[Line]
binary: bool
# the value `runs` groups by, so a caller gets its own key type back rather than `object`
Key = TypeVar("Key")
# an (start, end) index range of consecutive changed lines
Block = Tuple[int, int]
# a (start, end) character range inside one line's text
CharRange = Tuple[int, int]
# word-level marks only when the pair is mostly the same line, and never on huge lines (quadratic)
CHAR_DIFF_MIN_RATIO = 0.5
@@ -15,9 +35,9 @@ CHAR_DIFF_MAX_CHARS = 2000
_HUNK = re.compile(r"@@ -(\d+)(?:,\d+)? \+(\d+)")
def parse_unified(text):
def parse_unified(text: str) -> Diff:
"""Parse one file's unified diff (any number of hunks) into a Diff"""
lines = []
lines: List[Line] = []
binary = False
old_no = new_no = 0
in_hunk = False
@@ -52,25 +72,25 @@ def parse_unified(text):
return Diff(lines, binary)
def runs(items, key):
def runs(items: Sequence[Line], key: Callable[[Line], Key]) -> Iterator[Tuple[Key, int, int]]:
"""(key value, start, end) for each maximal run of consecutive items with the same key"""
for value, group in groupby(enumerate(items), key=lambda pair: key(pair[1])):
indexes = [i for i, _ in group]
yield value, indexes[0], indexes[-1] + 1
def change_blocks(lines):
def change_blocks(lines: Sequence[Line]) -> List[Block]:
"""[(start, end)) index ranges of consecutive changed lines"""
return [(start, end) for changed, start, end in runs(lines, lambda line: line.kind != " ") if changed]
def counts(lines):
def counts(lines: Sequence[Line]) -> Tuple[int, int]:
"""(insertions, deletions)"""
tally = Counter(line.kind for line in lines)
return tally["+"], tally["-"]
def paired_lines(lines, block):
def paired_lines(lines: Sequence[Line], block: Block) -> List[Tuple[int, int]]:
"""(deleted_index, inserted_index) pairs of a block with equal - and + counts"""
start, end = block
deleted = [i for i in range(start, end) if lines[i].kind == "-"]
@@ -80,15 +100,15 @@ def paired_lines(lines, block):
return list(zip(deleted, inserted))
def char_ranges(old, new):
def char_ranges(old: str, new: str) -> Tuple[List[CharRange], List[CharRange]]:
"""Character ranges that differ: ([(a, b) in old], [(a, b) in new]); empty when too different or too long"""
if len(old) + len(new) > CHAR_DIFF_MAX_CHARS:
return [], []
matcher = difflib.SequenceMatcher(None, old, new, autojunk=False)
if matcher.quick_ratio() < CHAR_DIFF_MIN_RATIO or matcher.ratio() < CHAR_DIFF_MIN_RATIO:
return [], []
old_ranges = []
new_ranges = []
old_ranges: List[CharRange] = []
new_ranges: List[CharRange] = []
for tag, i1, i2, j1, j2 in matcher.get_opcodes():
if tag in ("replace", "delete") and i2 > i1:
old_ranges.append((i1, i2))
@@ -97,7 +117,7 @@ def char_ranges(old, new):
return old_ranges, new_ranges
def line_for_open(lines, index):
def line_for_open(lines: Sequence[Line], index: int) -> int:
"""New-file line number to open for a row: a deleted row maps to the next surviving line"""
if not lines:
return 1

119
vdiff/render.py
View File

@@ -3,14 +3,15 @@ import bisect
import html
import logging
import os
from collections import namedtuple
from typing import Dict, List, NamedTuple, Optional, Sequence, Tuple
import sublime
import sublime_plugin
from . import palette, parse
from .git import STATUS_NAMES
from .state import VIEW_FLAG
from .git import FileChange, STATUS_NAMES
from .parse import Block, Line
from .state import VIEW_FLAG, ViewState
from .utils import file_count, settings, status
LOG = logging.getLogger("vdiff")
@@ -21,7 +22,7 @@ KEY_DEL_GUTTER = "vdiff.del.gutter"
KEY_INS_GUTTER = "vdiff.ins.gutter"
KEY_DEL_CHAR = "vdiff.del.char"
KEY_INS_CHAR = "vdiff.ins.char"
REGION_KEYS = (KEY_DEL, KEY_INS, KEY_DEL_GUTTER, KEY_INS_GUTTER, KEY_DEL_CHAR, KEY_INS_CHAR)
REGION_KEYS: Tuple[str, ...] = (KEY_DEL, KEY_INS, KEY_DEL_GUTTER, KEY_INS_GUTTER, KEY_DEL_CHAR, KEY_INS_CHAR)
PHANTOM_KEY = "vdiff.header"
STATUS_KEY = "vdiff"
@@ -51,9 +52,17 @@ NEEDS_RESTART = "restart Sublime Text to finish enabling syntax highlighting on
# one source for the status-bar hints and the ? popup.
# actions: the ACTIONS names the row covers; a single one makes the row a link. short: status-bar
# wording, "" keeps the row out of it. The group flag is False for groups needing a working-tree file.
Key = namedtuple("Key", "keys label short actions")
class Key(NamedTuple):
"""One row of the status-bar hints and the ? popup"""
KEY_GROUPS = [
keys: str
label: str
short: str
actions: Tuple[str, ...]
# (group title, applies in a commit, rows)
KEY_GROUPS: List[Tuple[str, bool, List[Key]]] = [
(
"Navigate",
True,
@@ -94,7 +103,7 @@ KEY_GROUPS = [
),
]
KEYS = [key for _, _, group in KEY_GROUPS for key in group]
KEYS: List[Key] = [key for _, _, group in KEY_GROUPS for key in group]
KEY_HINTS = " ".join(f"{k.keys} {k.short}" for k in KEYS if k.short)
@@ -137,11 +146,11 @@ a { text-decoration: none; color: color(var(--foreground) alpha(0.85)); }
class VdiffReplaceCommand(sublime_plugin.TextCommand):
"""Replace the whole buffer; the only writer of a review view"""
def run(self, edit, text):
def run(self, edit: sublime.Edit, text: str) -> None:
self.view.replace(edit, sublime.Region(0, self.view.size()), text)
def init_view(view):
def init_view(view: sublime.View) -> None:
"""Settings a review view needs, applied once when it is created or restored"""
s = view.settings()
s.set(VIEW_FLAG, True)
@@ -167,40 +176,40 @@ def init_view(view):
view.set_status(STATUS_KEY, KEY_HINTS)
def index_point(view, index, col=0):
def index_point(view: sublime.View, index: int, col: int = 0) -> int:
"""Text point of column `col` on the row showing st.lines[index]"""
return view.text_point(index + _HEADER_ROWS, col)
def caret_index(view):
def caret_index(view: sublime.View) -> int:
"""Index into st.lines of the caret's row (the header row maps to -1)"""
if not len(view.sel()):
return 0
return view.rowcol(view.sel()[0].begin())[0] - _HEADER_ROWS
def fingerprint(st):
def fingerprint(st: ViewState) -> Optional[Tuple[Optional[float], Optional[float]]]:
"""Cheap "did anything change" key for working-tree mode: mtimes of the file and of the git index"""
if st.in_commit or st.change is None:
return None
return (_mtime(os.path.join(st.repo, st.change.path)), _mtime(os.path.join(st.repo, ".git", "index")))
def _mtime(path):
def _mtime(path: str) -> Optional[float]:
try:
return os.stat(path).st_mtime
except OSError:
return None
def flash(view):
def flash(view: sublime.View) -> None:
"""Visual bell: tint the buffer for a moment, for feedback that has nothing to say in words"""
view.add_regions(BELL_KEY, [sublime.Region(0, view.size())], _bell_scope(view), flags=sublime.DRAW_NO_OUTLINE)
# unconditional erase: overlapping flashes may cut one short, but none can leave the tint stuck
sublime.set_timeout(lambda: view.is_valid() and view.erase_regions(BELL_KEY), BELL_MS)
def _bell_scope(view):
def _bell_scope(view: sublime.View) -> str:
"""region.yellowish, or a scope with a background we know of when a scheme lacks it"""
if palette.scope_background(view, SCOPE_BELL):
return SCOPE_BELL
@@ -208,7 +217,7 @@ def _bell_scope(view):
return palette.SCOPE_INS
def render_message(view, st, message):
def render_message(view: sublime.View, st: ViewState, message: str) -> None:
"""Show a one-line notice instead of a file (nothing to review, binary, ...)"""
st.change = None
st.lines = []
@@ -219,8 +228,11 @@ def render_message(view, st, message):
_finish(view, st, 0, 0, "vDiff")
def render(view, st):
def render(view: sublime.View, st: ViewState) -> None:
"""Draw st.lines for st.change into the view"""
change = st.change
# every caller with nothing to show goes to render_message instead, so there is a file here
assert change is not None
lines = st.lines
if st.binary:
text = "(binary file)"
@@ -230,7 +242,7 @@ def render(view, st):
text = "\n".join(line.text for line in lines)
_replace(view, text)
path = os.path.join(st.repo, st.change.path)
path = os.path.join(st.repo, change.path)
_apply_syntax(view, path if lines and not st.binary else None)
if lines and not st.binary:
@@ -242,13 +254,13 @@ def render(view, st):
status(NEEDS_RESTART)
ins, dele = parse.counts(lines)
_finish(view, st, ins, dele, f"⟐ {os.path.basename(st.change.path)}")
_finish(view, st, ins, dele, f"⟐ {os.path.basename(change.path)}")
view.sel().clear()
view.sel().add(sublime.Region(index_point(view, 0)))
view.set_viewport_position((0, 0), False)
def _paint_regions(view, st):
def _paint_regions(view: sublime.View, st: ViewState) -> None:
"""The line tint, the gutter dots, the word-diff marks, and the syntax colours on top of them"""
lines = st.lines
starts = _row_starts(lines)
@@ -260,7 +272,8 @@ def _paint_regions(view, st):
ins_marks = [sublime.Region(starts[i]) for i, line in enumerate(lines) if line.kind == "+"]
view.add_regions(KEY_DEL_GUTTER, del_marks, SCOPE_DEL_MARK, icon="dot", flags=mark_flags)
view.add_regions(KEY_INS_GUTTER, ins_marks, SCOPE_INS_MARK, icon="dot", flags=mark_flags)
del_chars, ins_chars = [], []
del_chars: List[sublime.Region] = []
ins_chars: List[sublime.Region] = []
if settings().get("word_diff", True):
del_chars, ins_chars = _char_regions(lines, st.blocks, starts)
view.add_regions(KEY_DEL_CHAR, del_chars, palette.SCOPE_DEL_CHAR, flags=sublime.DRAW_NO_OUTLINE)
@@ -268,7 +281,8 @@ def _paint_regions(view, st):
_paint_tokens(view, lines, starts, del_chars + ins_chars)
def _paint_tokens(view, lines, starts, char_regions):
def _paint_tokens(view: sublime.View, lines: Sequence[Line], starts: Sequence[int],
char_regions: Sequence[sublime.Region]) -> None:
"""Repaint each run of same-coloured tokens with a scope carrying the tint *and* that colour
add_regions draws the text it covers in the region scope's own foreground, so the tints above have
@@ -284,8 +298,8 @@ def _paint_tokens(view, lines, starts, char_regions):
stronger = sorted((r.a, r.b) for r in char_regions)
starts_of = [a for a, _ in stronger]
groups = {}
colours = {} # scope name -> colour, resolved once per render
groups: Dict[str, List[sublime.Region]] = {}
colours: Dict[str, Optional[str]] = {} # scope name -> colour, resolved once per render
for i in changed:
kind = "del" if lines[i].kind == "-" else "ins"
end = starts[i] + len(lines[i].text)
@@ -302,11 +316,11 @@ def _paint_tokens(view, lines, starts, char_regions):
view.add_regions(palette.token_key(scope), regions, scope, flags=sublime.DRAW_NO_OUTLINE)
def _split(region, intervals):
def _split(region: sublime.Region, intervals: Sequence[Tuple[int, int]]) -> List[Tuple[sublime.Region, bool]]:
"""(part, inside) covering `region`, cut wherever one of the sorted `intervals` starts or ends"""
if not intervals:
return [(region, False)]
parts = []
parts: List[Tuple[sublime.Region, bool]] = []
pos = region.a
for a, b in intervals:
a, b = max(a, region.a), min(b, region.b)
@@ -323,9 +337,10 @@ def _split(region, intervals):
return parts
def _colour_runs(view, start, text, colours):
def _colour_runs(view: sublime.View, start: int, text: str,
colours: Dict[str, Optional[str]]) -> List[Tuple[sublime.Region, str]]:
"""(region, colour) per run of same-coloured tokens on one line, adjacent equal colours merged"""
runs = []
runs: List[Tuple[sublime.Region, str]] = []
for region, scope in view.extract_tokens_with_scopes(sublime.Region(start, start + len(text))):
if scope not in colours:
colours[scope] = palette.foreground(view, scope)
@@ -339,7 +354,7 @@ def _colour_runs(view, start, text, colours):
return runs
def _finish(view, st, ins, dele, name):
def _finish(view: sublime.View, st: ViewState, ins: int, dele: int, name: str) -> None:
"""Header, tab name, status bar and bookkeeping shared by render and render_message"""
_header(view, st, ins, dele)
view.set_name(name)
@@ -348,7 +363,7 @@ def _finish(view, st, ins, dele, name):
st.fingerprint = fingerprint(st)
def _replace(view, text):
def _replace(view: sublime.View, text: str) -> None:
view.set_read_only(False)
try:
view.run_command("vdiff_replace", {"text": "\n" * _HEADER_ROWS + text})
@@ -359,7 +374,7 @@ def _replace(view, text):
view.erase_regions(key)
def _apply_syntax(view, path):
def _apply_syntax(view: sublime.View, path: Optional[str]) -> None:
"""The real file's syntax, so the diff is highlighted like the file itself"""
syntax = None
if path:
@@ -370,9 +385,9 @@ def _apply_syntax(view, path):
view.assign_syntax(syntax.path if syntax else "scope:text.plain")
def _row_starts(lines):
def _row_starts(lines: Sequence[Line]) -> List[int]:
"""Text point where each line starts; computed from the text we wrote, no view round-trips"""
starts = []
starts: List[int] = []
point = _HEADER_ROWS
for line in lines:
starts.append(point)
@@ -381,9 +396,10 @@ def _row_starts(lines):
return starts
def _line_regions(lines, starts):
def _line_regions(lines: Sequence[Line],
starts: Sequence[int]) -> Tuple[List[sublime.Region], List[sublime.Region]]:
"""Full-line regions of deleted and inserted rows, consecutive rows merged"""
regions = {"-": [], "+": []}
regions: Dict[str, List[sublime.Region]] = {"-": [], "+": []}
for kind, start, end in parse.runs(lines, lambda line: line.kind):
if kind in regions:
# starts has a sentinel at len(lines): the last line stops at the buffer end
@@ -391,10 +407,11 @@ def _line_regions(lines, starts):
return regions["-"], regions["+"]
def _char_regions(lines, blocks, starts):
def _char_regions(lines: Sequence[Line], blocks: Sequence[Block],
starts: Sequence[int]) -> Tuple[List[sublime.Region], List[sublime.Region]]:
"""Changed characters inside paired deleted/inserted lines"""
deleted = []
inserted = []
deleted: List[sublime.Region] = []
inserted: List[sublime.Region] = []
for block in blocks:
for del_row, ins_row in parse.paired_lines(lines, block):
old_ranges, new_ranges = parse.char_ranges(lines[del_row].text, lines[ins_row].text)
@@ -403,12 +420,12 @@ def _char_regions(lines, blocks, starts):
return deleted, inserted
def _totals(files):
def _totals(files: Sequence[FileChange]) -> Tuple[int, int]:
"""Summed numstat counts; binary files carry None counts and add nothing"""
return sum(c.insertions or 0 for c in files), sum(c.deletions or 0 for c in files)
def _summary_html(st):
def _summary_html(st: ViewState) -> str:
"""Line above the header: the whole commit on one line, or how much of the tree is staged"""
if not st.files and not st.commit:
return ""
@@ -425,10 +442,10 @@ def _summary_html(st):
return f'<div class="summary">{text}</div>'
def _commit_parts(st):
def _commit_parts(st: ViewState) -> List[str]:
"""hash, author, date, totals and message — everything about the reviewed commit, in that order"""
info = st.commit
parts = []
parts: List[str] = []
if info:
parts.append(f'<span class="hash">{html.escape(info.short)}</span>')
parts.append(html.escape(info.author))
@@ -445,7 +462,7 @@ def _commit_parts(st):
return parts
def header_html(st, ins, dele):
def header_html(st: ViewState, ins: int, dele: int) -> str:
"""The minihtml header: the summary line, then the shown file's path, counts, state and action links"""
change = st.change
if change is None:
@@ -460,7 +477,7 @@ def header_html(st, ins, dele):
if state:
title += f'&nbsp;&nbsp;<span class="state">{state}</span>'
actions = []
actions: List[str] = []
if change is not None and not st.in_commit:
actions.append('<a href="stage">Stage</a>')
actions.append('<a href="unstage">Unstage</a>')
@@ -477,15 +494,15 @@ def header_html(st, ins, dele):
)
def _run_action(view, href):
def _run_action(view: sublime.View, href: str) -> None:
"""Dispatch a header or popup link through the same command the keymap uses"""
view.run_command("vdiff_action", {"action": href})
def _header(view, st, ins, dele):
def _header(view: sublime.View, st: ViewState, ins: int, dele: int) -> None:
"""One block phantom above the text; redrawn from scratch, so no PhantomSet state to keep"""
def navigate(href):
def navigate(href: str) -> None:
_run_action(view, href)
view.erase_phantoms(PHANTOM_KEY)
@@ -495,9 +512,9 @@ def _header(view, st, ins, dele):
# --- key help popup ---------------------------------------------------------
def help_html(st):
def help_html(st: ViewState) -> str:
"""The minihtml key list; rows that need a working-tree file are dimmed while a commit is shown"""
parts = []
parts: List[str] = []
for title, applies, group in KEY_GROUPS:
dim = st.in_commit and not applies
heading = html.escape(title)
@@ -517,13 +534,13 @@ def help_html(st):
return f'<body id="vdiff-help"><style>{HELP_CSS}</style>{"".join(parts)}</body>'
def show_help(view, st):
def show_help(view: sublime.View, st: ViewState) -> None:
"""Toggle the key list popup at the caret"""
if view.is_popup_visible():
view.hide_popup()
return
def navigate(href):
def navigate(href: str) -> None:
view.hide_popup()
_run_action(view, href)

64
vdiff/state.py
View File

@@ -1,17 +1,30 @@
""" Per-view state of review views, keyed by view id; the essentials are mirrored into view settings for restore """
from collections import namedtuple
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, Dict, List, NamedTuple, Optional, Sequence, Tuple
if TYPE_CHECKING:
# annotation-only, so this module keeps importing neither sublime nor the rest of the package
import sublime
from .git import CommitInfo, FileChange
from .parse import Block, Line
# view setting marking a review view (keymap context "setting.vdiff_view")
VIEW_FLAG = "vdiff_view"
# view setting holding what the view shows, so a restored tab can re-render
VIEW_DATA = "vdiff"
# what a quick panel goes back to on Esc
Kept = namedtuple("Kept", "sha commit change files")
class Kept(NamedTuple):
"""What a quick panel goes back to on Esc"""
sha: Optional[str]
commit: Optional["CommitInfo"]
change: Optional["FileChange"]
files: List["FileChange"]
def find_change(files, path):
def find_change(files: Sequence["FileChange"], path: Optional[str]) -> int:
"""Index of the FileChange with `path`, or -1"""
for i, change in enumerate(files):
if change.path == path:
@@ -27,24 +40,25 @@ class ViewState:
"""
repo: str
sha: object = None # None in working-tree mode, the reviewed commit's hash otherwise
commit: object = None # git.CommitInfo when sha is set
change: object = None # git.FileChange currently shown
requested: object = None # file asked for, maybe still loading; navigation counts from here
files: list = field(default_factory=list) # git.FileChange list the user is browsing (next/prev file)
lines: list = field(default_factory=list) # parse.Line list currently rendered
blocks: list = field(default_factory=list) # parse.change_blocks(lines)
sha: Optional[str] = None # None in working-tree mode, the reviewed commit's hash otherwise
commit: Optional["CommitInfo"] = None # set when sha is
change: Optional["FileChange"] = None # file currently shown
requested: Optional["FileChange"] = None # file asked for, maybe still loading; navigation counts from here
files: List["FileChange"] = field(default_factory=list) # what the user is browsing (next/prev file)
lines: List["Line"] = field(default_factory=list) # currently rendered
blocks: List["Block"] = field(default_factory=list) # parse.change_blocks(lines)
binary: bool = False
token: int = 0 # bumped per show request; a stale async result is dropped
rendered_token: int = -1 # token of the request that produced the current render
highlight: int = 0 # bumped per quick-panel highlight, to debounce previews
fingerprint: object = None # (file mtime, index mtime) at render time, working-tree mode
# (file mtime, index mtime) at render time, working-tree mode; either mtime is None when unreadable
fingerprint: Optional[Tuple[Optional[float], Optional[float]]] = None
@property
def in_commit(self):
def in_commit(self) -> bool:
return self.sha is not None
def index(self):
def index(self) -> int:
"""Position in `files` of the file being shown or loaded, or -1
`requested` leads: a diff arrives asynchronously, so counting from `change` alone would make
@@ -53,10 +67,10 @@ class ViewState:
change = self.requested or self.change
return find_change(self.files, change.path) if change else -1
def kept(self):
def kept(self) -> Kept:
return Kept(self.sha, self.commit, self.change, list(self.files))
def enter(self, sha=None, commit=None):
def enter(self, sha: Optional[str] = None, commit: Optional["CommitInfo"] = None) -> Kept:
"""Switch to browsing a commit (or the working tree); returns what to restore on Esc"""
kept = self.kept()
if sha != self.sha:
@@ -64,13 +78,13 @@ class ViewState:
self.sha, self.commit = sha, commit
return kept
def restore(self, kept):
def restore(self, kept: Kept) -> None:
self.sha, self.commit, self.change, self.files = kept
self.requested = kept.change
def persist(self, view):
def persist(self, view: "sublime.View") -> None:
"""Mirror the essentials into view settings (survives a restart)"""
data = {"repo": self.repo, "sha": self.sha}
data: Dict[str, Any] = {"repo": self.repo, "sha": self.sha}
if self.change:
data["path"] = self.change.path
data["status"] = self.change.status
@@ -78,26 +92,26 @@ class ViewState:
view.settings().set(VIEW_DATA, data)
STATES = {}
STATES: Dict[int, "ViewState"] = {}
def get(view):
def get(view: "sublime.View") -> Optional[ViewState]:
return STATES.get(view.id())
def tracked():
def tracked() -> List[Tuple[int, ViewState]]:
"""Snapshot of (view id, state) for every tracked review view"""
return list(STATES.items())
def put(view, st):
def put(view: "sublime.View", st: ViewState) -> ViewState:
STATES[view.id()] = st
return st
def drop(view):
def drop(view: "sublime.View") -> None:
STATES.pop(view.id(), None)
def is_review_view(view):
def is_review_view(view: Optional["sublime.View"]) -> bool:
return bool(view and view.settings().get(VIEW_FLAG))

15
vdiff/utils.py
View File

@@ -1,5 +1,6 @@
""" Small helpers shared across vdiff modules """
import logging
from typing import Any, Callable, Optional
import sublime
@@ -9,10 +10,10 @@ SETTINGS_FILE = "vDiff.sublime-settings"
LOG = logging.getLogger("vdiff")
_settings = None
_settings: Optional[sublime.Settings] = None
def settings():
def settings() -> sublime.Settings:
"""Cached settings handle; a sublime Settings object stays live"""
global _settings
if _settings is None:
@@ -22,25 +23,25 @@ def settings():
return _settings
def apply_git_binary():
def apply_git_binary() -> None:
"""Push the configured git executable into the (sublime-free) git module"""
git.GIT_BINARY = settings().get("git_binary") or "git"
def file_count(count):
def file_count(count: int) -> str:
"""'1 file' / '3 files' for a message"""
return f'{count} {"file" if count == 1 else "files"}'
def status(message):
def status(message: str) -> None:
"""Status-bar message with the plugin prefix"""
sublime.status_message(f"vDiff: {message}")
def run_async(work, done):
def run_async(work: Callable[[], Any], done: Callable[[Any], None]) -> None:
"""Run `work()` on sublime's worker thread, hand the result to `done` on the main thread"""
def job():
def job() -> None:
try:
result = work()
except Exception as error: