vacuum-tube/vacuumtube/prepare.py

from vacuumtube.config import config
from datetime import datetime
from rich import print
import json
import subprocess


# https://stackoverflow.com/questions/1094841/get-human-readable-version-of-file-size
# except, the units match the output of `df -h` and `borg info`
def readable_size(num, suffix=""):
    for unit in ["B", "K", "M", "G", "T", "P", "E", "Z"]:
        if abs(num) < 1000.0:
            return f"{num:3.1f}{unit}{suffix}"
        num /= 1000.0
    return f"{num:.1f}Y{suffix}"


# setup
# get the repo information

if not config['repo']['path']:
    raise Exception('No repo path provided!')
if not config['disk']['partition']:
    raise Exception('No partition path provided!')

# check if a passphrase has been provided
if config['repo']['passphrase']:
    passphrase = 'BORG_PASSPHRASE=' + config['repo']['passphrase'] + ' '
else:
    passphrase = ''


borg_info_raw = subprocess.run(passphrase + 'borg info --json ' + config['repo']['path'], shell=True, capture_output=True)

borg_list = subprocess.run(passphrase + 'borg list --json ' + config['repo']['path'], shell=True, capture_output=True)

# get free disk space
df_avail_raw = subprocess.run(config['disk']['ssh'] + ' "df --block-size=1000 --output=avail ' + config['disk']['partition'] + ' | tail -1"', shell=True, capture_output=True)
df_avail = df_avail_raw.stdout.rstrip()

try:
    # parse repo info from json
    borg_info = json.loads(borg_info_raw.stdout)

    # parse repo info from json
    borg_list = json.loads(borg_list.stdout)
except json.decoder.JSONDecodeError:
    print('[bold red]Unable to reach repo!')

# try to get compressed + deduplicated backup size
try:
    stats = borg_info['cache']['stats']
    csize = int(stats['unique_csize'])
    readable_csize = readable_size(csize)
    last_archive = borg_list['archives'][-1]
    archive_num = len(borg_list['archives'])
except NameError:
    csize = 0
    readable_csize = last_archive = archive_num = '??'

# datetime format: https://borgbackup.readthedocs.io/en/stable/internals/frontends.html#standard-output
try:
    last_archive_time = datetime.strptime(last_archive['time'], '%Y-%m-%dT%H:%M:%S.%f')
    last_archive_time = last_archive_time.strftime('%d/%m/%Y %H:%M')
except TypeError:
    last_archive_time = 'Unknown'

if df_avail_raw.returncode == 0:
    df_avail_bytes = int(df_avail) * 1000
    df_avail_readable = readable_size(df_avail_bytes)
else:
    # placeholder to make the bar empty
    df_avail_bytes = 1000
    df_avail_readable = '??'