ota-flasher/app/release_checker.py

import json
import os
import requests
import wx
from typing import Callable
from datetime import datetime, timedelta

from app.utils import keep_latest_versions

CACHE_FILE = 'firmware/cache.json'
CACHE_DURATION = timedelta(minutes=30)


class ReleaseChecker:
    '''Release Checker for firmware updates'''
    release_name = ""
    commit_hash = ""

    def __init__(self):
        self.progress_callback: Callable[[int], None] = None

    def load_cache(self):
        '''Load cached data from file'''
        if os.path.exists(CACHE_FILE):
            with open(CACHE_FILE, 'r') as f:
                return json.load(f)
        return {}

    def save_cache(self, cache_data):
        '''Save cache data to file'''
        with open(CACHE_FILE, 'w') as f:
            json.dump(cache_data, f)

    def fetch_latest_release(self):
        '''Fetch latest firmware release from GitHub'''
        repo = "btclock/btclock_v3"
        cache = self.load_cache()
        now = datetime.now()

        if not os.path.exists("firmware"):
            os.makedirs("firmware")

        if 'latest_release' in cache and (now - datetime.fromisoformat(cache['latest_release']['timestamp'])) < CACHE_DURATION:
            latest_release = cache['latest_release']['data']
        else:
            url = f"https://api.github.com/repos/{repo}/releases/latest"
            try:
                response = requests.get(url)
                response.raise_for_status()
                latest_release = response.json()
                cache['latest_release'] = {
                    'data': latest_release,
                    'timestamp': now.isoformat()
                }
                self.save_cache(cache)
            except requests.RequestException as e:
                raise ReleaseCheckerException(
                    f"Error fetching release: {e}") from e

        release_name = latest_release['tag_name']
        self.release_name = release_name

        filenames_to_download = ["lolin_s3_mini_213epd_firmware.bin",
                                 "btclock_rev_b_213epd_firmware.bin", "littlefs.bin"]

        asset_urls = [asset['browser_download_url']
                      for asset in latest_release['assets'] if asset['name'] in filenames_to_download]

        if asset_urls:
            for asset_url in asset_urls:
                self.download_file(asset_url, release_name)

            ref_url = f"https://api.github.com/repos/{
                repo}/git/ref/tags/{release_name}"
            if ref_url in cache and (now - datetime.fromisoformat(cache[ref_url]['timestamp'])) < CACHE_DURATION:
                commit_hash = cache[ref_url]['data']

            else:
                response = requests.get(ref_url)
                response.raise_for_status()
                ref_info = response.json()
                if ref_info["object"]["type"] == "commit":
                    commit_hash = ref_info["object"]["sha"]
                else:
                    tag_url = f"https://api.github.com/repos/{
                        repo}/git/tags/{ref_info['object']['sha']}"
                    response = requests.get(tag_url)
                    response.raise_for_status()
                    tag_info = response.json()
                    commit_hash = tag_info["object"]["sha"]
                cache[ref_url] = {
                    'data': commit_hash,
                    'timestamp': now.isoformat()
                }
                self.save_cache(cache)

            self.commit_hash = commit_hash

            return self.release_name
        else:
            raise ReleaseCheckerException(
                f"File {filenames_to_download} not found in latest release")

    def download_file(self, url, release_name):
        '''Downloads Fimware Files'''
        local_filename = f"{release_name}_{url.split('/')[-1]}"
        response = requests.get(url, stream=True)
        total_length = response.headers.get('content-length')
        if not os.path.exists("firmware"):
            os.makedirs("firmware")
        if os.path.exists(f"firmware/{local_filename}"):
            return

        keep_latest_versions('firmware', 2)

        if total_length is None:
            raise ReleaseCheckerException("No content length header")
        else:
            total_length = int(total_length)
            chunk_size = 1024
            num_chunks = total_length // chunk_size
            with open(f"firmware/{local_filename}", 'wb') as f:
                for i, chunk in enumerate(response.iter_content(chunk_size=chunk_size)):
                    if chunk:
                        f.write(chunk)
                        f.flush()
                        progress = int((i / num_chunks) * 100)
                        if callable(self.progress_callback):
                            self.progress_callback(progress)

                if callable(self.progress_callback):
                    self.progress_callback(100)


class ReleaseCheckerException(Exception):
    pass