|
import json |
|
import logging |
|
import os |
|
import sys |
|
import threading |
|
import time |
|
import zipfile |
|
from urllib.parse import urlencode |
|
from urllib.request import urlopen, urlretrieve |
|
|
|
|
|
|
|
class Spinner: |
|
busy = False |
|
delay = 0.1 |
|
|
|
@staticmethod |
|
def spinning_cursor(): |
|
while 1: |
|
for cursor in "|/-\\": |
|
yield cursor |
|
|
|
def __init__(self, delay=None): |
|
self.spinner_generator = self.spinning_cursor() |
|
if delay and float(delay): |
|
self.delay = delay |
|
|
|
def spinner_task(self): |
|
while self.busy: |
|
sys.stdout.write(next(self.spinner_generator)) |
|
sys.stdout.flush() |
|
time.sleep(self.delay) |
|
sys.stdout.write("\b") |
|
sys.stdout.flush() |
|
|
|
def __enter__(self): |
|
self.busy = True |
|
threading.Thread(target=self.spinner_task).start() |
|
|
|
def __exit__(self, exception, value, tb): |
|
self.busy = False |
|
time.sleep(self.delay) |
|
if exception is not None: |
|
return False |
|
|
|
|
|
def download_cect_data(bundle_root: str = "."): |
|
base_url = "https://cloud-api.yandex.net/v1/disk/public/resources/download?" |
|
public_key = "https://disk.yandex.ru/d/pWEKt6D3qi3-aw" |
|
|
|
url = base_url + urlencode(dict(public_key=public_key)) |
|
response = urlopen(url) |
|
filepath = "AVUCTK_cases.zip" |
|
if not os.path.exists(os.path.join(bundle_root, filepath)): |
|
with urlopen(url) as response: |
|
code = response.getcode() |
|
if code == 200: |
|
download_url = json.loads(response.read())["href"] |
|
print("Downloading file...") |
|
with Spinner(): |
|
urlretrieve(download_url, filepath) |
|
else: |
|
raise RuntimeError( |
|
f"Download of file from {url} to {filepath} failed due to network issue or denied permission." |
|
) |
|
else: |
|
logging.info("zipfile exists, skipping download") |
|
with zipfile.ZipFile(os.path.join(bundle_root, filepath), "r") as zip_ref: |
|
zip_ref.extractall() |
|
|
