Raymond_Bailie/TANK-GIT
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit

Browse Source
This commit is contained in:
Raymond Bailie
2026-06-17 07:11:08 +02:00
commit 5075e5f87b
19 changed files with 2239 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

755
lib/ugit.py Normal file
View File

@@ -0,0 +1,755 @@
# ugit v2.0
# MicroPython OTA update from GitHub
# Created by TURFPTAx for the OpenMuscle project
# https://openmuscle.org
#
# Install:
# import mip; mip.install("github:turfptax/ugit")
#
# Quick start (one-time setup via REPL or Thonny):
# import ugit
# ugit.create_config(ssid='YourWifi', password='YourPass',
# user='github_user', repository='your-repo')
#
# Then in your main.py (safe to commit — no secrets in code):
# import ugit
# ugit.pull_all()
#
# Credentials are stored in /config.json on the device only.
# This file is auto-ignored so ugit will never delete or overwrite it.
# NEVER commit config.json to your GitHub repository.
__version__ = '2.1.0'
import os
import urequests
import json
import hashlib
import binascii
import machine
import time
import network
# _GITHUB_API = 'https://git.ultiware.co.za/repos'
# _GITHUB_RAW = 'https://raw.githubusercontent.com'
_GITEA_API = 'https://git.ultiware.co.za/api/v1/repos'
_GITEA_RAW = 'https://git.ultiware.co.za'
_USER_AGENT = 'ugit-turfptax'
_CONFIG_PATH = '/config.json'
def _headers(token=''):
h = {'User-Agent': _USER_AGENT}
if token:
h['authorization'] = 'bearer %s' % token
return h
def _git_blob_hash(data):
"""Compute SHA1 the way GitHub does: sha1('blob {size}\\0{content}')"""
if isinstance(data, str):
data = data.encode('utf-8')
header = ('blob %d\0' % len(data)).encode('utf-8')
sha = hashlib.sha1(header + data)
return binascii.hexlify(sha.digest()).decode('utf-8')
def _local_file_hash(filepath):
"""Get the GitHub-compatible blob SHA1 of a local file."""
try:
f = open(filepath, 'rb')
data = f.read()
f.close()
except:
data = b''
return _git_blob_hash(data)
def _is_directory(path):
try:
# stat[0] has mode bits; 0x4000 is the directory flag
return os.stat(path)[0] & 0x4000 != 0
except:
return False
def _build_internal_tree(path='/'):
"""Recursively scan the device filesystem, return {'/path': 'sha1hash'}."""
tree = {}
os.chdir(path)
for item in os.listdir():
full = path + item if path.endswith('/') else path + '/' + item
if _is_directory(full):
if os.listdir(full):
tree.update(_build_internal_tree(full))
else:
try:
tree[full] = _local_file_hash(full)
except OSError:
pass
os.chdir('/')
return tree
def _get_storage_info():
"""Get filesystem storage info in bytes. Returns (total, free, used)."""
stat = os.statvfs('/')
block_size = stat[0]
total_blocks = stat[2]
free_blocks = stat[3]
total = block_size * total_blocks
free = block_size * free_blocks
used = total - free
return total, free, used
def _file_size(filepath):
"""Get size of a single file in bytes."""
try:
return os.stat(filepath)[6]
except:
return 0
def _is_usb_cdc():
"""Detect if this board likely uses native USB-CDC for serial.
USB-CDC boards (ESP32-S2, S3, C3, C6) lose their serial port on
machine.reset(). If boot.py/main.py crashes before USB re-enumerates,
the device becomes inaccessible and requires reflashing.
"""
try:
machine_str = os.uname().machine.upper()
for chip in ('ESP32S3', 'ESP32S2', 'ESP32C6', 'ESP32C3', 'ESP32H2'):
if chip in machine_str:
return True
except:
pass
return False
def _is_ignored(path, ignore):
"""Check if path matches any ignore entry (exact or directory prefix).
'/lib' in ignore matches '/lib/aioble/core.mpy' but not '/library/foo.py'.
"""
for entry in ignore:
e = entry.rstrip('/')
if path == e or path.startswith(e + '/'):
return True
return False
def _repo_download_size(git_tree, local_tree, ignore):
"""Estimate bytes that will be downloaded (only changed/new files)."""
download = 0
for item in git_tree['tree']:
if item['type'] != 'blob':
continue
path = item['path']
if not path.startswith('/'):
path = '/' + path
if _is_ignored(path, ignore):
continue
git_sha = item.get('sha', '')
local_sha = local_tree.get(path, '')
if not (git_sha and local_sha and git_sha == local_sha):
download += item.get('size', 0)
return download
def _fmt_size(b):
"""Format bytes as human-readable string."""
if b < 1024:
return '%d B' % b
elif b < 1024 * 1024:
return '%d KB' % (b // 1024)
else:
return '%d.%d MB' % (b // (1024 * 1024), (b % (1024 * 1024)) * 10 // (1024 * 1024))
def storage_info():
"""Print human-readable storage info for the device."""
total, free, used = _get_storage_info()
print('Storage: %s total, %s used, %s free (%d%% free)' % (
_fmt_size(total), _fmt_size(used), _fmt_size(free),
(free * 100) // total if total else 0))
return {'total': total, 'free': free, 'used': used}
def _ensure_ignore(ignore):
"""Make sure config.json and ugit.py are always in the ignore list."""
if ignore is None:
ignore = []
protected = ['/ugit.py', _CONFIG_PATH, '/ugit.backup', '/ugit_log.txt', '/lib', '/sd']
for p in protected:
if p not in ignore:
ignore.append(p)
return ignore
def _load_config():
"""Load config from /config.json. Returns dict or empty dict if not found."""
try:
f = open(_CONFIG_PATH, 'r')
cfg = json.loads(f.read())
f.close()
return cfg
except:
return {}
def _resolve_config(user=None, repository=None, branch=None, token=None,
ssid=None, password=None, ignore=None):
"""Merge explicit arguments with config.json, arguments take priority."""
cfg = _load_config()
return {
'user': user or cfg.get('user', ''),
'repository': repository or cfg.get('repository', ''),
'branch': branch or cfg.get('branch', 'main'),
'token': token if token is not None else cfg.get('token', ''),
'ssid': ssid or cfg.get('ssid', ''),
'password': password or cfg.get('password', ''),
'ignore': ignore if ignore is not None else cfg.get('ignore', []),
}
def create_config(ssid='', password='', user='', repository='',
branch='main', token='', ignore=None):
"""
Save credentials to /config.json on the device.
Run this once via REPL or Thonny to set up your device.
The config file stays on the device and is never synced to GitHub.
Example:
ugit.create_config(
ssid='MyWifi',
password='MyPassword',
user='turfptax',
repository='my-project',
token='ghp_xxxx' # optional, for private repos
)
"""
if ignore is None:
ignore = []
cfg = {
'ssid': ssid,
'password': password,
'user': user,
'repository': repository,
'branch': branch,
'token': token,
'ignore': ignore,
}
# MicroPython devices have no secure keychain; credentials must be stored
# on the local filesystem. config.json is auto-protected from sync and
# should never be committed to the GitHub repository.
f = open(_CONFIG_PATH, 'w')
f.write(json.dumps(cfg))
f.close()
# Print confirmation (field names only, never values)
field_names = [k for k in cfg if cfg[k]]
print('Config saved to %s' % _CONFIG_PATH)
print('Fields stored: %s' % ', '.join(field_names))
print('WARNING: config.json contains credentials. Never commit it to GitHub.')
def show_config():
"""Display current config (passwords are masked)."""
cfg = _load_config()
if not cfg:
print('No config found. Run ugit.create_config() to set up.')
return
for k, v in cfg.items():
if k in ('password', 'token') and v:
print(' %s: %s' % (k, v[:3] + '*' * (len(v) - 3)))
else:
print(' %s: %s' % (k, v))
def wificonnect(ssid=None, password=None):
"""Connect to WiFi. Returns the WLAN object.
If ssid/password not provided, reads from config.json."""
if not ssid or not password:
cfg = _load_config()
ssid = ssid or cfg.get('ssid', '')
password = password or cfg.get('password', '')
if not ssid or not password:
raise ValueError('No WiFi credentials. Pass ssid/password or run ugit.create_config()')
wlan = network.WLAN(network.STA_IF)
wlan.active(False)
wlan.active(True)
wlan.connect(ssid, password)
retries = 0
while not wlan.isconnected():
time.sleep(1)
retries += 1
if retries > 30:
raise OSError('WiFi connection timed out')
print('WiFi connected:', wlan.ifconfig()[0])
return wlan
def pull_git_tree(user, repository, branch='main', token=''):
"""Fetch the full recursive tree from GitHub API."""
url = '%s/%s/%s/git/trees/%s?recursive=1' % (_GITHUB_API, user, repository, branch)
r = urequests.get(url, headers=_headers(token))
data = json.loads(r.content.decode('utf-8'))
r.close()
if 'tree' not in data:
raise Exception('Branch "%s" not found for %s/%s' % (branch, user, repository))
return data
def pull(filepath, raw_url, token=''):
"""Download a single file from GitHub and write it to the device."""
r = urequests.get(raw_url, headers=_headers(token))
data = r.content
r.close()
# ensure parent directory exists
parts = filepath.split('/')
for i in range(1, len(parts)):
d = '/'.join(parts[:i])
if d:
try:
os.mkdir(d)
except:
pass
f = open(filepath, 'wb')
f.write(data)
f.close()
return data
def pull_all(user=None, repository=None, branch=None, token=None,
ssid=None, password=None, ignore=None,
isconnected=False, reset_after=False):
"""
Sync device filesystem with a GitHub repository.
Only downloads files whose SHA1 hash differs from the local copy.
Deletes local files not present in the repo (except ignored files).
All arguments are optional if config.json exists on the device.
Explicit arguments override config.json values.
Args:
user: GitHub username
repository: GitHub repository name
branch: Branch to pull from (default 'main')
token: Personal access token for private repos
ssid: WiFi SSID (skipped if isconnected=True)
password: WiFi password
ignore: Extra file paths to never touch
isconnected: Set True if already connected to WiFi
reset_after: Reset the device after update (default False).
Ignored on USB-CDC boards (ESP32-S2/S3/C3/C6) to
prevent bricking if boot code has errors.
"""
c = _resolve_config(user, repository, branch, token, ssid, password, ignore)
ignore = _ensure_ignore(c['ignore'])
if not c['user'] or not c['repository']:
raise ValueError('user and repository required. Pass them or run ugit.create_config()')
if not isconnected:
wificonnect(c['ssid'], c['password'])
os.chdir('/')
raw_base = '%s/%s/%s/%s/' % (_GITHUB_RAW, c['user'], c['repository'], c['branch'])
# fetch repo tree from GitHub
print('Fetching repository tree...')
git_tree = pull_git_tree(c['user'], c['repository'], c['branch'], c['token'])
# build local file tree with hashes
print('Scanning local files...')
local_tree = _build_internal_tree()
log = []
updated = 0
skipped = 0
deleted = 0
git_files = set()
for item in git_tree['tree']:
path = item['path']
# normalize to absolute path
if not path.startswith('/'):
path = '/' + path
if item['type'] == 'tree':
try:
os.mkdir(path)
except:
pass
continue
git_files.add(path)
if _is_ignored(path, ignore):
skipped += 1
continue
# compare hashes — only download if changed or new
git_sha = item.get('sha', '')
local_sha = local_tree.get(path, '')
if git_sha and local_sha and git_sha == local_sha:
skipped += 1
log.append(path + ' unchanged')
continue
# download the file
try:
pull(path, raw_base + item['path'], c['token'])
updated += 1
log.append(path + ' updated')
print(' updated:', path)
except Exception as e:
log.append(path + ' FAILED: ' + str(e))
print(' FAILED:', path, e)
# delete local files not in the repo (except ignored)
for local_path in local_tree:
if local_path not in git_files and not _is_ignored(local_path, ignore):
try:
os.remove(local_path)
deleted += 1
log.append(local_path + ' deleted')
print(' deleted:', local_path)
except:
log.append(local_path + ' delete failed')
# write log
summary = 'ugit: %d updated, %d skipped, %d deleted' % (updated, skipped, deleted)
print(summary)
log.insert(0, summary)
try:
f = open('/ugit_log.txt', 'w')
f.write('\n'.join(log))
f.close()
except:
pass
if reset_after:
if _is_usb_cdc():
print('WARNING: USB-CDC board detected (%s).' % os.uname().machine)
print('machine.reset() skipped — reset manually or power-cycle.')
print('Auto-reset can brick USB-CDC boards if boot.py/main.py crashes.')
else:
print('Resetting device in 5 seconds...')
time.sleep(5)
machine.reset()
return log
def check_for_updates(user=None, repository=None, branch=None, token=None,
ignore=None, isconnected=False,
ssid=None, password=None):
"""
Check if the repo has changes without downloading anything.
Returns a dict with 'new', 'changed', 'deleted' file lists.
All arguments are optional if config.json exists on the device.
"""
c = _resolve_config(user, repository, branch, token, ssid, password, ignore)
ignore = _ensure_ignore(c['ignore'])
if not c['user'] or not c['repository']:
raise ValueError('user and repository required. Pass them or run ugit.create_config()')
if not isconnected:
wificonnect(c['ssid'], c['password'])
os.chdir('/')
git_tree = pull_git_tree(c['user'], c['repository'], c['branch'], c['token'])
local_tree = _build_internal_tree()
git_files = set()
new = []
changed = []
deleted = []
for item in git_tree['tree']:
if item['type'] != 'blob':
continue
path = item['path']
if not path.startswith('/'):
path = '/' + path
git_files.add(path)
if _is_ignored(path, ignore):
continue
local_sha = local_tree.get(path, '')
if not local_sha:
new.append(path)
elif item.get('sha', '') != local_sha:
changed.append(path)
for local_path in local_tree:
if local_path not in git_files and not _is_ignored(local_path, ignore):
deleted.append(local_path)
return {'new': new, 'changed': changed, 'deleted': deleted}
def update(branch=None, token=None):
"""Update ugit.py itself from this repository.
Args:
branch: Branch to pull from (default: 'main'). Use this to test
development branches, e.g. ugit.update('fix/my-branch')
token: GitHub token for private repos (reads from config if None)
Detects where ugit was imported from (e.g. /lib/ugit.py vs /ugit.py)
and updates in-place so mip-installed copies are updated correctly.
"""
if token is None:
cfg = _load_config()
token = cfg.get('token', '')
if branch is None:
branch = 'main'
# Update in the same location ugit was imported from
try:
dest = __file__
except:
dest = '/ugit.py'
raw_url = '%s/turfptax/ugit/%s/ugit.py' % (_GITHUB_RAW, branch)
pull(dest, raw_url, token)
print('ugit updated at %s from branch %s. Reset device to use new version.' % (dest, branch))
def backup(ignore=None):
"""Backup all files on the device to /ugit.backup.
Returns the number of files backed up, or -1 if not enough space."""
ignore = _ensure_ignore(ignore)
local_tree = _build_internal_tree()
# estimate backup size: sum of file contents + metadata overhead
backup_size = 0
file_count = 0
for path in local_tree:
if not _is_ignored(path, ignore):
backup_size += _file_size(path) + len(path) + 80 # metadata per file
file_count += 1
_, free, _ = _get_storage_info()
if backup_size > free:
print('Not enough space for backup.')
print(' Need: %s, Free: %s' % (_fmt_size(backup_size), _fmt_size(free)))
return -1
f = open('/ugit.backup', 'w')
f.write('ugit backup v2\n')
for path, sha in local_tree.items():
if _is_ignored(path, ignore):
continue
f.write('FILE:%s SHA:%s\n' % (path, sha))
try:
df = open(path, 'rb')
content = df.read()
df.close()
try:
f.write('---\n')
f.write(content.decode('utf-8'))
f.write('\n---\n')
except:
f.write('---BINARY:%d---\n' % len(content))
except:
f.write('---ERROR---\n')
f.close()
print('Backup saved to /ugit.backup (%d files, ~%s)' % (
file_count, _fmt_size(backup_size)))
return file_count
def restore(ignore=None):
"""
Restore files from /ugit.backup.
Reads the backup file, recreates directories, and writes files back.
Use this if an OTA update went wrong.
Example:
ugit.restore()
"""
ignore = _ensure_ignore(ignore)
try:
f = open('/ugit.backup', 'r')
except:
print('No backup file found at /ugit.backup')
return False
header = f.readline().strip()
if not header.startswith('ugit backup'):
print('Invalid backup file format.')
f.close()
return False
restored = 0
skipped = 0
current_path = None
current_content = None
in_content = False
for line in f:
line_stripped = line.rstrip('\n')
if line_stripped.startswith('FILE:'):
# save previous file if we have one
if current_path and current_content is not None:
if not _is_ignored(current_path, ignore):
_restore_file(current_path, current_content)
restored += 1
else:
skipped += 1
# parse new file entry: FILE:/path SHA:hash
parts = line_stripped.split(' SHA:')
current_path = parts[0][5:] # strip 'FILE:'
current_content = None
in_content = False
elif line_stripped == '---' and not in_content:
in_content = True
current_content = ''
elif line_stripped == '---' and in_content:
in_content = False
elif line_stripped.startswith('---BINARY:') or line_stripped == '---ERROR---':
in_content = False
current_content = None
print(' skip (binary/error): %s' % current_path)
elif in_content:
if current_content:
current_content += '\n' + line_stripped
else:
current_content = line_stripped
# save last file
if current_path and current_content is not None:
if not _is_ignored(current_path, ignore):
_restore_file(current_path, current_content)
restored += 1
else:
skipped += 1
f.close()
print('Restore complete: %d files restored, %d skipped' % (restored, skipped))
return True
def _restore_file(filepath, content):
"""Write a file back to the device, creating directories as needed."""
parts = filepath.split('/')
for i in range(1, len(parts)):
d = '/'.join(parts[:i])
if d:
try:
os.mkdir(d)
except:
pass
f = open(filepath, 'w')
f.write(content)
f.close()
print(' restored:', filepath)
def safe_pull_all(user=None, repository=None, branch=None, token=None,
ssid=None, password=None, ignore=None,
isconnected=False, reset_after=False):
"""
Like pull_all(), but checks available storage first and creates a
backup before updating. If the update fails, the backup remains
on the device for manual restore via ugit.restore().
Pre-flight checks:
1. Estimates download size from GitHub tree
2. Estimates backup size from local files
3. Verifies enough free space for backup + downloads
4. Creates backup
5. Runs the update
6. If update fails, prints restore instructions
reset_after defaults to False. On USB-CDC boards (ESP32-S2/S3/C3/C6),
machine.reset() is always skipped to prevent bricking.
Example:
ugit.safe_pull_all()
"""
c = _resolve_config(user, repository, branch, token, ssid, password, ignore)
ignore = _ensure_ignore(c['ignore'])
if not c['user'] or not c['repository']:
raise ValueError('user and repository required. Pass them or run ugit.create_config()')
if not isconnected:
wificonnect(c['ssid'], c['password'])
os.chdir('/')
# step 1: gather info
print('Pre-flight check...')
total, free, used = _get_storage_info()
print(' Storage: %s free of %s' % (_fmt_size(free), _fmt_size(total)))
git_tree = pull_git_tree(c['user'], c['repository'], c['branch'], c['token'])
local_tree = _build_internal_tree()
# step 2: estimate sizes
download_size = _repo_download_size(git_tree, local_tree, ignore)
backup_size = 0
for path in local_tree:
if not _is_ignored(path, ignore):
backup_size += _file_size(path) + len(path) + 80
# we need space for: backup file + downloaded files + 4KB buffer
needed = backup_size + download_size + 4096
print(' Backup estimate: %s' % _fmt_size(backup_size))
print(' Download estimate: %s' % _fmt_size(download_size))
print(' Total needed: %s' % _fmt_size(needed))
if needed > free:
shortfall = needed - free
print('\nNot enough space! Short by %s.' % _fmt_size(shortfall))
print('Free up space or use pull_all() without backup.')
return None
# step 3: backup
print('\nCreating backup...')
result = backup(ignore)
if result == -1:
print('Backup failed. Aborting safe update.')
return None
# step 4: run the update
print('\nStarting update...')
try:
log = pull_all(
user=c['user'], repository=c['repository'],
branch=c['branch'], token=c['token'],
ignore=ignore, isconnected=True, reset_after=False
)
except Exception as e:
print('\nUpdate FAILED: %s' % str(e))
print('Your backup is at /ugit.backup')
print('Run ugit.restore() to roll back.')
return None
print('\nUpdate complete with backup at /ugit.backup')
if reset_after:
if _is_usb_cdc():
print('WARNING: USB-CDC board detected (%s).' % os.uname().machine)
print('machine.reset() skipped — reset manually or power-cycle.')
print('Auto-reset can brick USB-CDC boards if boot.py/main.py crashes.')
else:
print('Resetting device in 5 seconds...')
time.sleep(5)
machine.reset()
return log