abe_setup_py/abe_setup.py

#!/usr/bin/env python3
#
# A script to setup a Working tree from abe git submodules

DEBUG = 1

def main():
    global DEBUG
    p = parser()
    args = p.parse_args()
    DEBUG = args.debug
    DEBUG = DEBUG - (1 if args.quiet else 0)

    print(args) if DEBUG>1 else None

    if args.source_url and args.source_url.find('://') < 0 and args.source_url.find( ':' ) > 0:
        args.source_url = 'ssh://' + args.source_url.replace( ':', '/', 1)

    for ret in recurse_abe_submodules( args.output_dir, args.source_url, args.tag, clone_repo_cached(cache_dir=args.cache_dir)):
        pass


def parser():
    import argparse
    p = argparse.ArgumentParser()
    p.add_argument('-d', '--debug', action='count', default=1)
    p.add_argument('-q', '--quiet', action='store_true')
    p.add_argument('-o', '--output_dir', default='.')
    p.add_argument('-c', '--cache_dir', default=None)
    p.add_argument('-t', '--tag', default='master')
    p.add_argument('-s', '--source_url', default=None)
    return p


def recurse_abe_submodules( parent, func=None):
    if func != None:
        yield func( parent )

    subm = get_abe_submodules( parent.subdir )

    import os
    import urllib.parse
    for su in subm:
        print(f"Repo: \"{parent.path}\" has Submodule: \"{su.subdir}\" Type: \"{su.subtype}\" From: \"{newurl}\"") if DEBUG else None
        for ret in recurse_abe_submodules( subm.full_path(), newurl, su.ref, func):
            yield ret

def cache_path(cache, remote):
    import os.path
    for url in remote:
        sane_url = str(url).replace('/','_').replace( '@', '_').replace(':', '_')
        return os.path.join( cache, sane_url)

def real_relpath(dest, source='.'):
    import os.path
    real_dest = os.path.realpath(dest)
    real_source = os.path.realpath(source)
    return os.path.relpath(real_dest, real_source)

from git import RemoteProgress
class ProgressPrinter(RemoteProgress):
    def update(self,op_code,cur_count,max_count=None,message=''):
        cur_count_int = round(cur_count)
        max_count_int = round(max_count or 100)
        print(f"\033[2K{cur_count_int}/{max_count_int}", str(round(100*cur_count / (max_count or 100)))+'%', end="\r")

def clone_repo_cached( cache_dir=None, bare=False):
    def clone_repo(path, remote, ref=None, cache_dir=cache_dir, bare=bare):
        print('Repo:', path) if DEBUG>1 else None
        from git import Repo
        repo = Repo.init(path, bare=bare)
        if 'origin' in repo.remotes:
            origin = repo.remotes['origin']
        else:
            origin = repo.create_remote('origin', remote)

        if cache_dir != None:
            cache_repo_path = cache_path(cache_dir, origin.urls)
            print(f"Cacheing from: {cache_repo_path}") if DEBUG else None
            clone_repo_cached(cache_dir=None, bare=True)( cache_repo_path, remote, ref)
            if 'cache' in repo.remotes:
                cache = repo.remotes['cache']
            else:
                cache = repo.create_remote('cache', real_relpath( cache_repo_path, path))
                cache.set_url("no_push" , '--push')
            cache.fetch(refspec="+refs/remotes/origin/*:refs/remotes/origin/*",progress=ProgressPrinter())
            print()
        else:
            origin.fetch(progress=ProgressPrinter())
            print()

        if not bare:
            print('Refs:', origin.refs) if DEBUG>1 else None
            print('Heads:', repo.heads) if DEBUG>1 else None

            if ref in repo.tags:
                tracking_branch_name = 'local_tag_branch/'+ref
                tracking_ref = repo.tags[ref]
                if tracking_branch_name in repo.heads:
                    active_branch = repo.heads[tracking_branch_name]
                else:
                    active_branch = repo.create_head('local_tag_branch/'+ref, tracking_ref)
            elif ref in repo.heads:
                tracking_ref = origin.refs[ref]
                active_branch = repo.heads[ref]
                active_branch.set_tracking_branch(tracking_ref)
            elif ref in origin.refs:
                tracking_ref = origin.refs[ref]
                active_branch = repo.create_head(ref, tracking_ref)
                active_branch.set_tracking_branch(tracking_ref)
            elif ref in origin.refs:
                tracking_ref = origin.refs[ref]
                active_branch = repo.create_head(ref, tracking_ref)
                active_branch.set_tracking_branch(tracking_ref)
            else:
                try:
                    tracking_ref = ref
                    tracking_branch_name = 'local_commit_branch/'+ref
                    active_branch = repo.create_head(tracking_branch_name, tracking_ref)
                except Exception:
                    raise Exception(f"Branch/Tag/Commit \"{ref}\" not found")
            print('Active Branch:', active_branch) if DEBUG else None
            print('Tracking Ref:', tracking_ref, '\n') if DEBUG else None
            active_branch.checkout()
            repo.head.reset( tracking_ref, index=True, working_tree=False)
        return repo
    return clone_repo


from enum import Enum
class AbeSubType(Enum):
    SUBMODULE = 1
    BOOTLOADER = 2
    BSP = 3


from urllib.parse import urlparse
class AbeSubmodule():
    subtype = AbeSubType.SUBMODULE
    def __init__(self, sub_info, subtype=None):
        if subtype != None:
            self.subtype = subtype
        self.subdir = sub_info[0]
        self.remote = sub_info[1]
        self.ref    = sub_info[2]
        if self.subtype == AbeSubType.BSP:
            self.remote = 'bsp/' + self.remote
        if self.subtype == AbeSubType.BOOTLOADER:
            self.remote = 'bootloader/' + self.remote

    def __repr__(self):
        return f"AbeSubmodule(['{self.subdir}','{self.remote}','{self.ref}'], {self.subtype})"

    def full_url(self, parent=None):
        selfurl = urlparse(self.remote)
        if not parent:
            return selfurl.geturl()
        purl = urlparse(parent.remote)
        newurl = purl._replace( path = selfurl.path )
        if selfurl.netloc:
            newurl = url._replace( netloc = selfurl.netloc )
        if selfurl.scheme:
            newurl = url._replace( scheme = selfurl.scheme )
        return newurl

    def full_path(self, parent=None):
        if not parent:
            return self.subdir
        import os.path
        return os.path.normpath(
            os.path.join( parent.full_path(),self.subdir)
        )



def get_abe_submodules(repo_dir):
    import os
    import itertools
    subfile_generators = []
    subfile_path = os.path.join(repo_dir, '.abe', 'submodules')
    if os.path.isfile(subfile_path):
        subfile_generators.append(parse_abe_subfile(subfile_path, subtype=AbeSubType.SUBMODULE))
    bspfile_path = os.path.join(repo_dir, '.abe', 'bsps')
    if os.path.isfile(bspfile_path):
        subfile_generators.append(parse_abe_subfile(bspfile_path, subtype=AbeSubType.BSP))
    bootloaderfile_path = os.path.join(repo_dir, '.abe', 'bootloaders')
    if os.path.isfile(bootloaderfile_path):
        subfile_generators.append(parse_abe_subfile(bootloaderfile_path, subtype=AbeSubType.BOOTLOADER))
    return itertools.chain(*subfile_generators)


def parse_abe_subfile(subfile_path, subtype=AbeSubType.SUBMODULE):
    with open(subfile_path) as subfile:
        for line in subfile.readlines():
            strline = line.strip()
            if strline.startswith('#'):
                continue
            sline = strline.split()
            if len(sline) < 3:
                continue
            print('Submodule:', sline) if DEBUG>1 else None
            sub = AbeSubmodule(sline, subtype)
            yield sub


if __name__=='__main__':
    main()