mubert_txt2music.ipynb

{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/olaviinha/NeuralTextToMusic/blob/main/mubert_txt2music.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "wCskRiNym_LW"
      },
      "source": [
        "#<font face=\"Trebuchet MS\" size=\"6\">Mubert <font color=\"#999\" size=\"4\">&nbsp;&nbsp;&nbsp;|&nbsp;&nbsp;&nbsp;</font><font color=\"#999\" size=\"4\">text2music</font><font color=\"#999\" size=\"4\">&nbsp;&nbsp;&nbsp;|&nbsp;&nbsp;&nbsp;</font><a href=\"https://github.com/olaviinha/NeuralTextToMusic\" target=\"_blank\"><font color=\"#999\" size=\"4\">Github</font></a>\n",
        "\n",
        "This is a modified version of Mubert [Text-to-Music notebook](https://colab.research.google.com/github/ferluht/Mubert-Text-to-Music/blob/main/Mubert_Text_to_Music.ipynb). It uses [Mubert](https://mubert.com/) [API](https://mubert2.docs.apiary.io/#) to generate tracks based on text prompt and/or tags.\n",
        "\n",
        "**LICENSING INFORMATION and other details can be found from [the original MubertAI Github](https://github.com/MubertAI/Mubert-Text-to-Music).**\n",
        "\n",
        "### Tips\n",
        "- At the time of writing this, Mubert API endpoint used by this notebook does not appear extremely reliable; one minute it may work like a dream, the next minute every generation attempt will result in an error.\n",
        "\n",
        "- Use **either** prompt(s) or tag(s). If you fill `use_tags`, prompt(s) will be ignored altogether.\n",
        "\n",
        "- You may generate multiple tracks in one go by separating either prompts or tags by semicolon (`;`).\n",
        "\n",
        "- Tags should be separated with comma (`,`). To generate two tracks by different tags in one go, `use_tags` value should look something like `neurofunk, dubstep; future pop, joyful`.\n",
        "\n",
        "- If you have drive mounted and enter an `output_dir`, the notebook will attempt to auto-save all files there (prone to failures). Enter path relative to your Google Drive root, e.g. `music/mubert` if your Drive contains a directory called _music_, with a subdirectory called _mubert_."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "cellView": "form",
        "id": "A6-8KXU6m8TM"
      },
      "outputs": [],
      "source": [
        "#@title #Setup\n",
        "#@markdown This cell needs to be run only once. It will mount your Google Drive and setup prerequisites.<br>\n",
        "#@markdown <small>Mounting Drive will enable this notebook to save outputs directly to your Drive. Otherwise you will need to copy/download them manually from this notebook.</small>\n",
        "\n",
        "force_setup = False\n",
        "repositories = []\n",
        "pip_packages = ''\n",
        "apt_packages = ''\n",
        "mount_drive = True #@param {type:\"boolean\"}\n",
        "skip_setup = False #@ param {type:\"boolean\"}\n",
        "verbose = False\n",
        "\n",
        "mubert_tags_string = 'tribal,action,kids,neo-classic,run 130,pumped,jazz / funk,ethnic,dubtechno,reggae,acid jazz,liquidfunk,funk,witch house,tech house,underground,artists,mystical,disco,sensorium,r&b,agender,psychedelic trance / psytrance,peaceful,run 140,piano,run 160,setting,meditation,christmas,ambient,horror,cinematic,electro house,idm,bass,minimal,underscore,drums,glitchy,beautiful,technology,tribal house,country pop,jazz & funk,documentary,space,classical,valentines,chillstep,experimental,trap,new jack swing,drama,post-rock,tense,corporate,neutral,happy,analog,funky,spiritual,sberzvuk special,chill hop,dramatic,catchy,holidays,fitness 90,optimistic,orchestra,acid techno,energizing,romantic,minimal house,breaks,hyper pop,warm up,dreamy,dark,urban,microfunk,dub,nu disco,vogue,keys,hardcore,aggressive,indie,electro funk,beauty,relaxing,trance,pop,hiphop,soft,acoustic,chillrave / ethno-house,deep techno,angry,dance,fun,dubstep,tropical,latin pop,heroic,world music,inspirational,uplifting,atmosphere,art,epic,advertising,chillout,scary,spooky,slow ballad,saxophone,summer,erotic,jazzy,energy 100,kara mar,xmas,atmospheric,indie pop,hip-hop,yoga,reggaeton,lounge,travel,running,folk,chillrave & ethno-house,detective,darkambient,chill,fantasy,minimal techno,special,night,tropical house,downtempo,lullaby,meditative,upbeat,glitch hop,fitness,neurofunk,sexual,indie rock,future pop,jazz,cyberpunk,melancholic,happy hardcore,family / kids,synths,electric guitar,comedy,psychedelic trance & psytrance,edm,psychedelic rock,calm,zen,bells,podcast,melodic house,ethnic percussion,nature,heavy,bassline,indie dance,techno,drumnbass,synth pop,vaporwave,sad,8-bit,chillgressive,deep,orchestral,futuristic,hardtechno,nostalgic,big room,sci-fi,tutorial,joyful,pads,minimal 170,drill,ethnic 108,amusing,sleepy ambient,psychill,italo disco,lofi,house,acoustic guitar,bassline house,rock,k-pop,synthwave,deep house,electronica,gabber,nightlife,sport & fitness,road trip,celebration,electro,disco house,electronic'\n",
        "\n",
        "# additional_tags = 'extreme,sentimental'\n",
        "# mubert_tags_string = mubert_tags_string+','+additional_tags\n",
        "\n",
        "# Download the repo from Github\n",
        "import os\n",
        "from google.colab import output\n",
        "import warnings\n",
        "warnings.filterwarnings('ignore')\n",
        "%cd /content/\n",
        "\n",
        "# inhagcutils\n",
        "if not os.path.isfile('/content/inhagcutils.ipynb') and force_setup == False:\n",
        "  !pip -q install import-ipynb {pip_packages}\n",
        "  if apt_packages != '':\n",
        "    !apt-get update && apt-get install {apt_packages}\n",
        "  !curl -s -O https://raw.githubusercontent.com/olaviinha/inhagcutils/master/inhagcutils.ipynb\n",
        "import import_ipynb\n",
        "from inhagcutils import *\n",
        "\n",
        "# Mount Drive\n",
        "if mount_drive is True:\n",
        "  if not os.path.isdir('/content/drive'):\n",
        "    from google.colab import drive\n",
        "    drive.mount('/content/drive')\n",
        "    drive_root = '/content/drive/My Drive'\n",
        "  if not os.path.isdir('/content/mydrive'):\n",
        "    os.symlink('/content/drive/My Drive', '/content/mydrive')\n",
        "    drive_root = '/content/mydrive/'\n",
        "  drive_root_set = True\n",
        "else:\n",
        "  create_dirs(['/content/faux_drive'])\n",
        "  drive_root = '/content/faux_drive/'\n",
        "\n",
        "if len(repositories) > 0 and skip_setup == False:\n",
        "  for repo in repositories:\n",
        "    %cd /content/\n",
        "    install_dir = fix_path('/content/'+path_leaf(repo).replace('.git', ''))\n",
        "    repo = repo if '.git' in repo else repo+'.git'\n",
        "    !git clone {repo}\n",
        "    if os.path.isfile(install_dir+'setup.py') or os.path.isfile(install_dir+'setup.cfg'):\n",
        "      !pip install -e ./{install_dir}\n",
        "    if os.path.isfile(install_dir+'requirements.txt'):\n",
        "      !pip install -r {install_dir}/requirements.txt\n",
        "\n",
        "if len(repositories) == 1:\n",
        "  %cd {install_dir}\n",
        "\n",
        "dir_tmp = '/content/tmp/'\n",
        "create_dirs([dir_tmp])\n",
        "\n",
        "import time, sys\n",
        "from datetime import timedelta\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "import subprocess, time\n",
        "print(\"Setting up environment...\")\n",
        "start_time = time.time()\n",
        "all_process = [\n",
        "    ['pip', 'install', 'torch==1.12.1+cu113', 'torchvision==0.13.1+cu113', '--extra-index-url', 'https://download.pytorch.org/whl/cu113'],\n",
        "    ['pip', 'install', '-U', 'sentence-transformers'],\n",
        "    ['pip', 'install', 'httpx'],\n",
        "]\n",
        "for process in all_process:\n",
        "    running = subprocess.run(process,stdout=subprocess.PIPE).stdout.decode('utf-8')\n",
        "\n",
        "end_time = time.time()\n",
        "print(f\"Environment set up in {end_time-start_time:.0f} seconds\")\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "import numpy as np\n",
        "from sentence_transformers import SentenceTransformer\n",
        "minilm = SentenceTransformer('all-MiniLM-L6-v2')\n",
        "\n",
        "mubert_tags = np.array(mubert_tags_string.split(','))\n",
        "mubert_tags_embeddings = minilm.encode(mubert_tags)\n",
        "\n",
        "from IPython.display import Audio, display\n",
        "import httpx\n",
        "import json\n",
        "\n",
        "def get_track_by_prompt(prompt, pat, duration, maxit=20, autoplay=False, loop=False, intensity=\"High\", format=\"MP3\"):\n",
        "  job_status = 1\n",
        "  if loop:\n",
        "    mode = \"loop\"\n",
        "  else:\n",
        "    mode = \"track\"\n",
        "  r = httpx.post('https://api-b2b.mubert.com/v2/TTMRecordTrack', \n",
        "      json={\n",
        "          \"method\":\"RecordTrackTTM\",\n",
        "          \"params\": {\n",
        "              \"pat\": pat, \n",
        "              \"duration\": duration,\n",
        "              \"format\": format,\n",
        "              \"bitrate\": 192,\n",
        "              \"intensity\": intensity,\n",
        "              \"text\": prompt,\n",
        "              \"mode\": mode,\n",
        "          }\n",
        "      })\n",
        "\n",
        "  rdata = json.loads(r.text)\n",
        "  print( rdata )\n",
        "  assert rdata['status'] == 1, rdata['error']['text']\n",
        "  trackurl = rdata['data']['tasks'][0]['download_link']\n",
        "        \n",
        "  for i in range(maxit):\n",
        "    r = httpx.post('https://api-b2b.mubert.com/v2/TrackStatus', \n",
        "        json={\n",
        "          \"method\":\"TrackStatus\",\n",
        "          \"params\":\n",
        "            {\n",
        "                \"pat\": pat\n",
        "            }\n",
        "        })\n",
        "    rdata = json.loads(r.text)\n",
        "    print( rdata )\n",
        "    job_status = rdata['data']['tasks'][0]['task_status_code']\n",
        "    # print( rdata )\n",
        "    # print( job_status )\n",
        "    if job_status == 2:\n",
        "      time.sleep(5)\n",
        "      display(Audio(trackurl, autoplay=autoplay))\n",
        "      break\n",
        "    time.sleep(1)\n",
        "\n",
        "def get_track_by_tags(tags, pat, duration, maxit=20, autoplay=False, loop=False, intensity=\"high\", format=\"mp33\"):\n",
        "  job_status = 1\n",
        "  if loop:\n",
        "    mode = \"loop\"\n",
        "  else:\n",
        "    mode = \"track\"\n",
        "  r = httpx.post('https://api-b2b.mubert.com/v2/RecordTrackTTM', \n",
        "      json={\n",
        "          \"method\":\"RecordTrackTTM\",\n",
        "          \"params\": {\n",
        "              \"pat\": pat, \n",
        "              \"duration\": duration,\n",
        "              \"format\": format,\n",
        "              \"bitrate\": 192,\n",
        "              \"intensity\": intensity,\n",
        "              \"tags\": tags,\n",
        "              \"mode\": mode,\n",
        "          }\n",
        "      })\n",
        "\n",
        "  rdata = json.loads(r.text)\n",
        "  if verbose is True: print(rdata)\n",
        "  assert rdata['status'] == 1, rdata['error']['text']\n",
        "  trackurl = rdata['data']['tasks'][0]['download_link']\n",
        "  task_id = rdata['data']['tasks'][0]['task_id']\n",
        "        \n",
        "  for i in range(maxit):\n",
        "    r = httpx.post('https://api-b2b.mubert.com/v2/TrackStatus', \n",
        "        json={\n",
        "          \"method\":\"TrackStatus\",\n",
        "          \"params\":\n",
        "            {\n",
        "                \"pat\": pat\n",
        "            }\n",
        "        })\n",
        "    rdata = json.loads(r.text)\n",
        "    job_status = 1\n",
        "    if verbose is True: print(rdata)\n",
        "    for task in rdata['data']['tasks']:\n",
        "      if verbose is True: print('Task', task)\n",
        "      if verbose is True: print(task['task_id'], 'vs', task_id)\n",
        "      if task['task_id'] == task_id:\n",
        "        job_status = int(task['task_status_code'])\n",
        "    # job_status = rdata['data']['tasks'][0]['task_status_code']\n",
        "    # print( rdata )\n",
        "    if verbose is True: print( 'Task status:', job_status )\n",
        "    if job_status == 2:\n",
        "      time.sleep(1)\n",
        "      # display(Audio(trackurl, autoplay=autoplay))\n",
        "      break\n",
        "    time.sleep(1)\n",
        "  \n",
        "  if verbose is True: print('Status polling done.')\n",
        "  if verbose is True: print('Now poll for DL link', url)\n",
        "  for i in range(maxit):\n",
        "    r = httpx.get(trackurl)\n",
        "    if verbose is True: print(r)\n",
        "    if r.status_code == 200:\n",
        "      # time.sleep(3)\n",
        "      # display(Audio(trackurl, autoplay=autoplay))\n",
        "      if (format == 'flac' or format == 'wav') and duration > 120:\n",
        "        op(c.okb, 'Unable to provide audio player for a large file, skipping...')\n",
        "      else:\n",
        "        display(Audio(trackurl, autoplay=autoplay))\n",
        "      break\n",
        "    time.sleep(1)\n",
        "  if verbose is True: print('All done, return', job_status, trackurl)\n",
        "  return job_status, trackurl\n",
        "\n",
        "def find_similar(em, embeddings, method='cosine'):\n",
        "    scores = []\n",
        "    for ref in embeddings:\n",
        "        if method == 'cosine': \n",
        "            scores.append(1 - np.dot(ref, em)/(np.linalg.norm(ref)*np.linalg.norm(em)))\n",
        "        if method == 'norm': \n",
        "            scores.append(np.linalg.norm(ref - em))\n",
        "    return np.array(scores), np.argsort(scores)\n",
        "\n",
        "def get_tags_for_prompts(prompts, top_n=3, debug=False):\n",
        "    prompts_embeddings = minilm.encode(prompts)\n",
        "    ret = []\n",
        "    for i, pe in enumerate(prompts_embeddings):\n",
        "        scores, idxs = find_similar(pe, mubert_tags_embeddings)\n",
        "        top_tags = mubert_tags[idxs[:top_n]]\n",
        "        top_prob = 1 - scores[idxs[:top_n]]\n",
        "        if debug:\n",
        "            print(f\"Prompt: {prompts[i]}\\nTags: {', '.join(top_tags)}\\nScores: {top_prob}\\n\\n\\n\")\n",
        "        ret.append((prompts[i], list(top_tags)))\n",
        "    return ret\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "# email = \"email@domain.com\" #@param {type:\"string\"}\n",
        "email = gen_id()+'@'+gen_id()+'.com'\n",
        "\n",
        "r = httpx.post('https://api-b2b.mubert.com/v2/GetServiceAccess', \n",
        "    json={\n",
        "        \"method\":\"GetServiceAccess\",\n",
        "        \"params\": {\n",
        "            \"email\": email,\n",
        "            \"license\":\"ttmmubertlicense#f0acYBenRcfeFpNT4wpYGaTQIyDI4mJGv5MfIhBFz97NXDwDNFHmMRsBSzmGsJwbTpP1A6i07AXcIeAHo5\",\n",
        "            \"token\":\"4951f6428e83172a4f39de05d5b3ab10d58560b8\",\n",
        "            \"mode\": \"loop\"\n",
        "        }\n",
        "    })\n",
        "\n",
        "rdata = json.loads(r.text)\n",
        "assert rdata['status'] == 1, \"probably incorrect e-mail\"\n",
        "pat = rdata['data']['pat']\n",
        "print(f'Got token: {pat}')\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n",
        "output.clear()\n",
        "# !nvidia-smi\n",
        "op(c.ok, 'Setup finished.')\n",
        "\n",
        "def divide_chunks(l, n):\n",
        "    for i in range(0, len(l), n):\n",
        "        yield l[i:i + n]\n",
        "\n",
        "available_tags = mubert_tags_string.split(',')\n",
        "available_tags.sort() \n",
        "available_tag_chunks = list(divide_chunks(available_tags, 8))\n",
        "print()\n",
        "op(c.title, 'Available tags (for use_tags field):')\n",
        "for chunk in available_tag_chunks:\n",
        "  print( ', '.join(chunk) )"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "cellView": "form",
        "id": "H2z1os4Bocdy"
      },
      "outputs": [],
      "source": [
        "#@title # Generate tracks\n",
        "prompt = \"Grandma's space balls; Grandma's blueberry pie in style of country western ballad; Enter the Void\" #@param {type:\"string\"}\n",
        "\n",
        "#@markdown <small>If you use tags, prompt(s) will be ignored.\n",
        "use_tags = \"\" #@param {type:\"string\"}\n",
        "\n",
        "duration = \"60\" #@param {type:\"string\"}\n",
        "music_intensity = \"high\" #@param [\"low\", \"medium\", \"high\"]\n",
        "format = \"mp3\" #@param [\"mp3\", \"flac\", \"wav\"]\n",
        "output_dir = \"\" #@param {type:\"string\"}\n",
        "\n",
        "\n",
        "\n",
        "end_session_when_done = False #@ param {type: \"boolean\"}\n",
        "\n",
        "\n",
        "uniq_id = gen_id()\n",
        "tags_used = False\n",
        "intensity = music_intensity\n",
        "\n",
        "if use_tags == '':\n",
        "  if ';' in prompt:\n",
        "    prompts = [x.strip() for x in prompt.split(';')]\n",
        "  else:\n",
        "    prompts = [prompt]\n",
        "else:\n",
        "  tags_used = True\n",
        "  tag_sets_no = 1\n",
        "  if ';' in use_tags:\n",
        "    tag_sets_no = len(use_tags.split(';'))\n",
        "  prompts = ['tags-only'] * tag_sets_no\n",
        "\n",
        "if ':' in duration:\n",
        "  m, s = duration.split(':')\n",
        "  duration = int(m) * 60 + int(s)\n",
        "elif 'm' in duration:\n",
        "  m, s = duration.split('m')\n",
        "  if s == '': s = 0\n",
        "  duration = (int(m) * 60) + int(s.replace('s', ''))\n",
        "else:\n",
        "  duration = int(duration.replace('s', ''))\n",
        "\n",
        "# Output\n",
        "if output_dir == '':\n",
        "  dir_out = dir_tmp\n",
        "else:\n",
        "  if not os.path.isdir(drive_root+output_dir):\n",
        "    os.mkdir(drive_root+output_dir)\n",
        "  dir_out = drive_root+fix_path(output_dir)\n",
        "  \n",
        "timer_start = time.time()\n",
        "total = len(prompts)\n",
        "\n",
        "# -- DO THINGS --\n",
        "tags = []\n",
        "if tags_used is True:\n",
        "  if ';' in use_tags:\n",
        "    tag_sets = [x.strip() for x in use_tags.split(';')]\n",
        "  else:\n",
        "    tag_sets = [use_tags]\n",
        "  for utags in tag_sets:\n",
        "    if ',' in utags: \n",
        "      utags = [x.strip() for x in utags.split(',')]\n",
        "    else:\n",
        "      utags = [utags]\n",
        "    tags.append([prompts[0], utags])\n",
        "else:\n",
        "  tags = get_tags_for_prompts(prompts)\n",
        "\n",
        "retry_urls = []\n",
        "\n",
        "for i, tag in enumerate(tags, 1):\n",
        "  \n",
        "  used_prompt = tag[0]\n",
        "  used_tags = tag[1]\n",
        "  \n",
        "  if tags_used is True:\n",
        "    op(c.title, str(i)+'/'+str(total)+' Tags: '+', '.join(used_tags), time=True)\n",
        "    thing = ', '.join(used_tags)\n",
        "  else:\n",
        "    op(c.title, str(i)+'/'+str(total)+' Prompt: '+used_prompt, time=True)\n",
        "    op(c.okb, 'Tags: '+', '.join(used_tags), time=True)\n",
        "    thing = used_prompt\n",
        "  print()\n",
        "  try:\n",
        "    \n",
        "    status, url = get_track_by_tags(used_tags, pat, duration, maxit=30, autoplay=False, intensity=intensity, format=format)\n",
        "    if mount_drive is True and output_dir != '':\n",
        "      time.sleep(1)\n",
        "      if tags_used is True:\n",
        "        file_out = dir_out+slug('_'.join(used_tags))+'__'+uniq_id+'-'+str(i)+'.'+format\n",
        "      else:\n",
        "        file_out = dir_out+slug(used_prompt)+'__'+uniq_id+'-'+str(i)+'.'+format\n",
        "      print()\n",
        "      if status == 2:\n",
        "        !wget -q -O {file_out} {url}\n",
        "        if os.path.isfile(file_out) and os.path.getsize(file_out) > 0:\n",
        "          op(c.ok, 'Saved as', file_out.replace(drive_root, ''), time=True)\n",
        "        elif os.path.isfile(file_out) and os.path.getsize(file_out) == 0:\n",
        "          os.remove(file_out)\n",
        "          op(c.fail, 'Received a file of 0 bytes, skipping auto-save for now.', time=True)\n",
        "          # op(c.fail, 'Received a file of 0 bytes: file not auto-saved.')\n",
        "          if total == 1: op(c.fail, 'You can try to download the file manually from this link:', url)\n",
        "          retry_urls.append([thing, file_out, url])\n",
        "        else:\n",
        "          op(c.fail, 'ERROR saving file', file_out.replace(drive_root, ''), time=True)\n",
        "          # op(c.fail, 'You can try to download the file manually from this link:', url)\n",
        "      else:\n",
        "        op(c.fail, 'Generation is taking too long, skipping auto-save for now.', time=True)\n",
        "        if total==1: op(c.fail, 'You can try to download the file manually from this link:', url)\n",
        "        retry_urls.append([thing, file_out, url])\n",
        "    else:\n",
        "      op(c.title, 'Track download link:', url, time=True)\n",
        "    print()\n",
        "  except Exception as e:\n",
        "    print(str(e))\n",
        "  print()\n",
        "\n",
        "\n",
        "if len(retry_urls) > 0 and total > 1:\n",
        "  print()\n",
        "  op(c.warn, '- - - - - - - - - - - - - - - - -  - - - - - - - - - -')\n",
        "  op(c.warn, 'Checking if previously skipped tracks are ready yet...')\n",
        "  print()\n",
        "  time.sleep(3)\n",
        "  for retry_url in retry_urls:\n",
        "    thing, file_out, url = retry_url\n",
        "    op(c.title, thing, time=True)\n",
        "    print()\n",
        "    !wget -q -O {file_out} {url}\n",
        "    if os.path.isfile(file_out) and os.path.getsize(file_out) > 0:\n",
        "      display(Audio(file_out, autoplay=False))\n",
        "      op(c.ok, 'Saved as', file_out.replace(drive_root, ''), time=True)\n",
        "    elif os.path.isfile(file_out) and os.path.getsize(file_out) == 0:\n",
        "      os.remove(file_out)\n",
        "      op(c.fail, 'Received a file of 0 bytes again, file still not saved.', time=True)\n",
        "      op(c.fail, 'You can try to download the file manually from this link:', url, time=True)\n",
        "    else:\n",
        "      op(c.fail, 'ERROR saving file', file_out.replace(drive_root, ''), time=True)\n",
        "      op(c.fail, 'You can try to download the file manually from this link:', url, time=True)\n",
        "    print()\n",
        "\n",
        "# -- END THINGS --\n",
        "\n",
        "timer_end = time.time()\n",
        "\n",
        "print()\n",
        "op(c.okb, 'Elapsed', timedelta(seconds=timer_end-timer_start), time=True)\n",
        "op(c.ok, 'FIN.')\n",
        "\n",
        "if end_session_when_done is True: end_session()"
      ]
    }
  ],
  "metadata": {
    "colab": {
      "provenance": [],
      "authorship_tag": "ABX9TyMuyA3xtsUULUnsvg3VnaS/",
      "include_colab_link": true
    },
    "kernelspec": {
      "display_name": "Python 3",
      "name": "python3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}