Source code for qibocal.cli.upload

"""Upload report to server."""

import base64
import json
import pathlib
import shutil
import socket
import subprocess
import uuid
from urllib.parse import urljoin

import yaml
from qibo.config import log, raise_error

from .utils import META

# options for report upload
UPLOAD_HOST = (
    "qibocal@localhost"
    if socket.gethostname() == "saadiyat"
    else "qibocal@login.qrccluster.com"
)
TARGET_DIR = "qibocal-reports/"
ROOT_URL = "http://login.qrccluster.com:9000/"


[docs]def upload_report(path: pathlib.Path, tag: str, author: str): # load meta and update tag meta = yaml.safe_load((path / META).read_text()) meta["author"] = author if tag is not None: meta["tag"] = tag (path / META).write_text(json.dumps(meta, indent=4)) # check the rsync command exists. if not shutil.which("rsync"): raise_error( RuntimeError, "Could not find the rsync command. Please make sure it is installed.", ) # check that we can authentica with a certificate ssh_command_line = ( "ssh", "-o", "PreferredAuthentications=publickey", "-q", UPLOAD_HOST, "exit", ) str_line = " ".join(repr(ele) for ele in ssh_command_line) log.info(f"Checking SSH connection to {UPLOAD_HOST}.") try: subprocess.run(ssh_command_line, check=True) except subprocess.CalledProcessError as e: raise RuntimeError( ( "Could not validate the SSH key. " "The command\n%s\nreturned a non zero exit status. " "Please make sure that your public SSH key is on the server." ) % str_line ) from e except OSError as e: raise RuntimeError( "Could not run the command\n{}\n: {}".format(str_line, e) ) from e log.info("Connection seems OK.") # upload output randname = base64.urlsafe_b64encode(uuid.uuid4().bytes).decode() newdir = TARGET_DIR + randname rsync_command = ( "rsync", "-aLz", "--chmod=ug=rwx,o=rx", f"{path}/", f"{UPLOAD_HOST}:{newdir}", ) log.info(f"Uploading output ({path}) to {UPLOAD_HOST}") try: subprocess.run(rsync_command, check=True) except subprocess.CalledProcessError as e: msg = f"Failed to upload output: {e}" raise RuntimeError(msg) from e url = urljoin(ROOT_URL, randname) log.info(f"Upload completed. The result is available at:\n{url}") return url