import re
import pathlib
import subprocess
import threading
import requests
import bpy
from .. utility import addon
# Upstream publishes version TAGS only — that repo carries no addon code, so a new
# tag means "a vendor drop exists, go download it", never "we can fetch it".
update_url = 'https://api.github.com/repos/teamcsharp/HardOps-updater/tags'
# This install is a git checkout of the fork. Remotes are tried in order and the
# first that answers wins; punch:// is P2P over Hyperswarm and falls back to
# HTTPS on its own, so it is worth trying first.
FORK_REMOTES = ('p2p', 'web', 'forge')
FORK_BRANCH = 'main'
GIT_TIMEOUT = 20
FORK_TIMEOUT = 120 # punch:// peer discovery is slower than HTTPS
version_regex = re.compile(r"[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+")
# The check runs off the main thread, so it may not touch bpy data. It parks a
# result here and _drain applies it from a timer, which is on the main thread.
_result = None
_checking = False
def addon_path() -> pathlib.Path:
'''Root of the addon, which is also the fork's working tree'''
return pathlib.Path(__file__).resolve().parent.parent
def _git(*args, timeout=GIT_TIMEOUT):
'''Run git against the addon directory. Returns (ok, stdout, stderr)'''
try:
proc = subprocess.run(
('git', '-C', str(addon_path())) + args,
capture_output=True, text=True, timeout=timeout)
except (OSError, subprocess.SubprocessError) as error:
return False, '', str(error)
return proc.returncode == 0, proc.stdout.strip(), proc.stderr.strip()
def is_fork_checkout() -> bool:
'''True if this install is a git working tree we can update in place'''
ok, out, _ = _git('rev-parse', '--is-inside-work-tree')
return ok and out == 'true'
def remote_head(remote) -> str:
'''Branch tip on a remote without downloading anything, '' if unreachable'''
ok, out, _ = _git('ls-remote', remote, FORK_BRANCH, timeout=FORK_TIMEOUT)
if not ok: return ''
for line in out.splitlines():
parts = line.split()
if len(parts) == 2 and parts[1].endswith(FORK_BRANCH):
return parts[0]
return ''
def have_object(sha) -> bool:
'''True if the commit is already in the local object store'''
ok, _, _ = _git('cat-file', '-e', f'{sha}^{{commit}}')
return ok
def check_fork():
'''(remote, sha) for the first remote holding a commit we don't have.\n
Returns (None, '') when up to date, unreachable, or when we are AHEAD of the
remote — a differing sha we already hold means our own commits simply aren't
pushed yet, which is not an update.'''
if not is_fork_checkout(): return None, ''
for remote in FORK_REMOTES:
head = remote_head(remote)
if not head: continue
if have_object(head): return None, '' # up to date, or ahead
return remote, head
return None, ''
def check_vendor(current_version) -> str:
'''Latest upstream tag if it is newer than ours, else '' '''
response = requests.get(update_url, timeout=6)
if response.status_code != 200:
raise Exception
versions = list()
for tag in response.json():
name:str = tag['name']
# pad 3 number version
if name.count('.') < 4: name = f'{name}.0'
if len(current_version) < 4: current_version = (*current_version, 0)
if version_regex.match(name):
versions.append(name)
if not versions: return ''
latest_version_str = sorted(versions, reverse=True)[0]
current_version_str = '.'.join(map(str, list(current_version)))
return latest_version_str if str(current_version_str) != str(latest_version_str) else ''
def _worker(current_version):
'''Off the main thread — must not touch bpy data, only park a result'''
global _result
try:
remote, _sha = check_fork()
if remote:
_result = (f'Fork update available — pull from {remote}', True, remote)
return
except Exception:
pass
try:
latest = check_vendor(current_version)
_result = (f'HardOps {latest} available!', False, '') if latest else ('', False, '')
except Exception:
_result = ('Connection Failed', False, '')
def _drain():
'''Main thread — copy the worker's result into preferences'''
global _result, _checking
if _result is None:
return 1.0 # not finished, poll again
message, fork_update, remote = _result
_result = None
_checking = False
try:
preference = addon.preference()
preference.needs_update = message
preference.fork_update = fork_update
preference.fork_remote = remote
if message: print(message)
except Exception:
pass # addon torn down while the check was in flight
return None
def check_for_update(current_version):
global _checking
if not addon.preference().check_update: return
if _checking: return
# TODO: this should be stored in transient storage, like window_manager, so
# that we don't need to clear it here
addon.preference().needs_update = ""
addon.preference().fork_update = False
_checking = True
threading.Thread(target=_worker, args=(tuple(current_version),), daemon=True).start()
bpy.app.timers.register(_drain, first_interval=1.0)
def pull(remote):
'''Fast-forward this install from a fork remote. Returns (ok, message)'''
if not is_fork_checkout():
return False, 'This install is not a git checkout of the fork'
ok, out, err = _git('status', '--porcelain')
if not ok:
return False, err or 'git status failed'
if out:
return False, 'Local changes present — commit or stash them first'
ok, out, err = _git('pull', '--ff-only', remote, FORK_BRANCH, timeout=FORK_TIMEOUT)
if not ok:
return False, (err or out or 'pull failed').splitlines()[-1]
if 'Already up to date' in out:
return True, 'Already up to date'
return True, 'Updated — restart Blender to load the new version'
class HOPS_OT_ForkUpdate(bpy.types.Operator):
bl_idname = 'hops.fork_update'
bl_label = 'Update From Fork'
bl_description = 'Pull the latest commits from the fork. Blender must be restarted for them to take effect'
remote: bpy.props.StringProperty()
def execute(self, context):
remote = self.remote or addon.preference().fork_remote or FORK_REMOTES[0]
if context.window:
context.window.cursor_set('WAIT')
try:
ok, message = pull(remote)
finally:
if context.window:
context.window.cursor_set('DEFAULT')
self.report({'INFO'} if ok else {'WARNING'}, message)
if ok:
addon.preference().needs_update = message
addon.preference().fork_update = False
return {'FINISHED'}