cappath_config_writer() - Code Metrics - Inspection of "move configparser base to new module, refactor acc..." - thurask/bbarchivist - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( d56c20...c7f92e )

by John

created 2016-09-30 06:51 UTC

cappath_config_writer() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	2
c	1
b	0
f	0
dl	0
loc	13
rs	9.4285

#!/usr/bin/env python3
"""This module is used for miscellaneous utilities."""

import os  # path work
import argparse  # argument parser for filters
import platform  # platform info
import glob  # cap grabbing
import threading  # get thread for spinner
import time  # spinner delay
import sys  # streams, version info
import itertools  # spinners gonna spin
import subprocess  # loader verification
from bbarchivist import bbconstants  # cap location, version, filename bits
from bbarchivist import compat  # backwards compat
from bbarchivist import dummy  # useless stdout
from bbarchivist import iniconfig  # config parsing

__author__ = "Thurask"
__license__ = "WTFPL v2"
__copyright__ = "Copyright 2015-2016 Thurask"


def grab_cap():
    """
    Figure out where cap is, local, specified or system-supplied.
    """
    try:
        capfile = glob.glob(os.path.join(os.getcwd(), bbconstants.CAP.filename))[0]
    except IndexError:
        try:
            cappath = cappath_config_loader()
            capfile = glob.glob(cappath)[0]
        except IndexError:
            cappath_config_writer(bbconstants.CAP.location)
            return bbconstants.CAP.location  # no ini cap
        else:
            cappath_config_writer(os.path.abspath(capfile))
            return os.path.abspath(capfile)  # ini cap
    else:
        return os.path.abspath(capfile)  # local cap


def grab_cfp():
    """
    Figure out where cfp is, local or system-supplied.
    """
    try:
        cfpfile = glob.glob(os.path.join(os.getcwd(), bbconstants.CFP.filename))[0]
    except IndexError:
        cfpfile = bbconstants.CFP.location  # system cfp
    return os.path.abspath(cfpfile)  # local cfp


def new_enough(minver):
    """
    Check if we're at or above a minimum Python version.

    :param minver: Minimum Python version (3.minver).
    :type minver: int
    """
    return False if minver > sys.version_info[1] else True


def fsizer(file_size):
    """
    Raw byte file size to human-readable string.

    :param file_size: Number to parse.
    :type file_size: float
    """
    if file_size is None:
        file_size = 0
    fsize = float(file_size)
    for sfix in ['B', 'kB', 'MB', 'GB', 'TB', 'PB', 'EB', 'ZB']:
        if fsize < 1024.0:
            size = "{0:3.2f}{1}".format(fsize, sfix)
            break
        else:
            fsize /= 1024.0
    else:
        size = "{0:3.2f}{1}".format(fsize, 'YB')
    return size


def signed_file_args(files):
    """
    Check if there are between 1 and 6 files supplied to argparse.

    :param files: List of signed files, between 1 and 6 strings.
    :type files: list(str)
    """
    filelist = [file for file in files if file]
    if not 1 <= len(filelist) <= 6:
        raise argparse.ArgumentError(argument=None, message="Requires 1-6 signed files")
    return files


def file_exists(file):
    """
    Check if file exists, raise argparse error if it doesn't.

    :param file: Path to a file, including extension.
    :type file: str
    """
    if not os.path.exists(file):
        raise argparse.ArgumentError(argument=None, message="{0} not found.".format(file))
    return file


def positive_integer(input_int):
    """
    Check if number > 0, raise argparse error if it isn't.

    :param input_int: Integer to check.
    :type input_int: str
    """
    if int(input_int) <= 0:
        info = "{0} is not >=0.".format(str(input_int))
        raise argparse.ArgumentError(argument=None, message=info)
    return int(input_int)


def valid_method(method):
    """
    Check if compression method is valid, raise argparse error if it isn't.

    :param method: Compression method to check.
    :type method: str
    """
    methodlist = bbconstants.METHODS
    if not new_enough(3):
        methodlist = [x for x in bbconstants.METHODS if x != "txz"]
    if method not in methodlist:
        info = "Invalid method {0}.".format(method)
        raise argparse.ArgumentError(argument=None, message=info)
    return method


def valid_carrier(mcc_mnc):
    """
    Check if MCC/MNC is valid (1-3 chars), raise argparse error if it isn't.

    :param mcc_mnc: MCC/MNC to check.
    :type mcc_mnc: str
    """
    if not str(mcc_mnc).isdecimal():
        infod = "Non-integer {0}.".format(str(mcc_mnc))
        raise argparse.ArgumentError(argument=None, message=infod)
    if len(str(mcc_mnc)) > 3 or len(str(mcc_mnc)) == 0:
        infol = "{0} is invalid.".format(str(mcc_mnc))
        raise argparse.ArgumentError(argument=None, message=infol)
    else:
        return mcc_mnc


def escreens_pin(pin):
    """
    Check if given PIN is valid, raise argparse error if it isn't.

    :param pin: PIN to check.
    :type pin: str
    """
    if len(pin) == 8:
        try:
            int(pin, 16)  # hexadecimal-ness
        except ValueError:
            raise argparse.ArgumentError(argument=None, message="Invalid PIN.")
        else:
            return pin.lower()
    else:
        raise argparse.ArgumentError(argument=None, message="Invalid PIN.")


def escreens_duration(duration):
    """
    Check if Engineering Screens duration is valid.

    :param duration: Duration to check.
    :type duration: int
    """
    if int(duration) in (1, 3, 6, 15, 30):
        return int(duration)
    else:
        raise argparse.ArgumentError(argument=None, message="Invalid duration.")


def droidlookup_hashtype(method):
    """
    Check if Android autoloader lookup hash type is valid.

    :param method: None for regular OS links, "sha256/512" for SHA256 or 512 hash.
    :type method: str
    """
    if method.lower() in ("sha512", "sha256"):
        return method.lower()
    else:
        raise argparse.ArgumentError(argument=None, message="Invalid type.")


def droidlookup_devicetype(device):
    """
    Check if Android autoloader device type is valid.

    :param device: Android autoloader types to check.
    :type device: str
    """
    devices = ("Priv", "DTEK50")
    #devices = ("Priv", "DTEK50", "DTEK60")
    if device is None:
        return None
    else:
        for dev in devices:
            if dev.lower() == device.lower():
                return dev
        raise argparse.ArgumentError(argument=None, message="Invalid device.")


def s2b(input_check):
    """
    Return Boolean interpretation of string input.

    :param input_check: String to check if it means True or False.
    :type input_check: str
    """
    return str(input_check).lower() in ("yes", "true", "t", "1", "y")


def is_amd64():
    """
    Check if script is running on an AMD64 system.
    """
    return platform.machine().endswith("64")


def is_windows():
    """
    Check if script is running on Windows.
    """
    return platform.system() == "Windows"


def get_seven_zip(talkative=False):
    """
    Return name of 7-Zip executable.
    On POSIX, it MUST be 7za.
    On Windows, it can be installed or supplied with the script.
    :func:`win_seven_zip` is used to determine if it's installed.

    :param talkative: Whether to output to screen. False by default.
    :type talkative: bool
    """
    return win_seven_zip(talkative) if is_windows() else "7za"


def win_seven_zip(talkative=False):
    """
    For Windows, check where 7-Zip is ("where", pretty much).
    Consult registry first for any installed instances of 7-Zip.

    :param talkative: Whether to output to screen. False by default.
    :type talkative: bool
    """
    if talkative:
        print("CHECKING INSTALLED FILES...")
    try:
        import winreg  # windows registry
        hk7z = winreg.OpenKey(winreg.HKEY_CURRENT_USER, "Software\\7-Zip")
        path = winreg.QueryValueEx(hk7z, "Path")
    except OSError as exc:
        if talkative:
            print("SOMETHING WENT WRONG")
            print(str(exc))
            print("TRYING LOCAL FILES...")
        return win_seven_zip_local(talkative)
    else:
        if talkative:
            print("7ZIP USING INSTALLED FILES")
        return '"{0}"'.format(os.path.join(path[0], "7z.exe"))


def win_seven_zip_local(talkative=False):
    """
    If 7-Zip isn't in the registry, fall back onto supplied executables.
    If *those* aren't there, return "error".

    :param talkative: Whether to output to screen. False by default.
    :type talkative: bool
    """
    listdir = os.listdir(os.getcwd())
    filecount = 0
    for i in listdir:
        if i in ["7za.exe", "7za64.exe"]:
            filecount += 1
    if filecount == 2:
        if talkative:
            print("7ZIP USING LOCAL FILES")
        if is_amd64():
            return "7za64.exe"
        else:
            return "7za.exe"
    else:
        if talkative:
            print("NO LOCAL FILES")
        return "error"


def get_core_count():
    """
    Find out how many CPU cores this system has.
    """
    try:
        cores = str(compat.enum_cpus())  # 3.4 and up
    except NotImplementedError:
        cores = "1"  # 3.2-3.3
    else:
        if compat.enum_cpus() is None:
            cores = "1"
    return cores


def prep_seven_zip_path(path, talkative=False):
    """
    Print p7zip path on POSIX, or notify if not there.

    :param path: Path to use.
    :type path: str

    :param talkative: Whether to output to screen. False by default.
    :type talkative: bool
    """
    if path is None:
        if talkative:
            print("NO 7ZIP")
            print("PLEASE INSTALL p7zip")
        return False
    else:
        if talkative:
            print("7ZIP FOUND AT {0}".format(path))
        return True


def prep_seven_zip_posix(talkative=False):
    """
    Check for p7zip on POSIX.

    :param talkative: Whether to output to screen. False by default.
    :type talkative: bool
    """
    try:
        path = compat.where_which("7za")
    except ImportError:
        if talkative:
            print("PLEASE INSTALL SHUTILWHICH WITH PIP")
        return False
    else:
        return prep_seven_zip_path(path, talkative)


def prep_seven_zip(talkative=False):
    """
    Check for presence of 7-Zip.
    On POSIX, check for p7zip.
    On Windows, check for 7-Zip.

    :param talkative: Whether to output to screen. False by default.
    :type talkative: bool
    """
    if is_windows():
        return get_seven_zip(talkative) != "error"
    else:
        return prep_seven_zip_posix(talkative)


def increment(version, inc=3):
    """
    Increment version by given number. For repeated lookups.

    :param version: w.x.y.ZZZZ, becomes w.x.y.(ZZZZ + increment).
    :type version: str

    :param inc: What to increment by. Default is 3.
    :type inc: str
    """
    splitos = version.split(".")
    splitos[3] = int(splitos[3])
    if splitos[3] > 9996:  # prevent overflow
        splitos[3] = 0
    splitos[3] += int(inc)
    splitos[3] = str(splitos[3])
    return ".".join(splitos)


def stripper(name):
    """
    Strip fluff from bar filename.

    :param name: Bar filename, must contain '-nto+armle-v7+signed.bar'.
    :type name: str
    """
    return name.replace("-nto+armle-v7+signed.bar", "")


def generate_urls(baseurl, osversion, radioversion, core=False):
    """
    Generate a list of OS URLs and a list of radio URLs based on input.

    :param baseurl: The URL, from http to the hashed software release.
    :type baseurl: str

    :param osversion: OS version.
    :type osversion: str

    :param radioversion: Radio version.
    :type radioversion: str

    :param core: Whether or not to return core URLs as well.
    :type core: bool
    """
    suffix = "nto+armle-v7+signed.bar"
    osurls = [
        "{0}/winchester.factory_sfi.desktop-{1}-{2}".format(baseurl, osversion, suffix),
        "{0}/qc8960.factory_sfi.desktop-{1}-{2}".format(baseurl, osversion, suffix),
        "{0}/qc8960.factory_sfi.desktop-{1}-{2}".format(baseurl, osversion, suffix),
        "{0}/qc8974.factory_sfi.desktop-{1}-{2}".format(baseurl, osversion, suffix)
    ]
    radiourls = [
        "{0}/m5730-{1}-{2}".format(baseurl, radioversion, suffix),
        "{0}/qc8960-{1}-{2}".format(baseurl, radioversion, suffix),
        "{0}/qc8960.omadm-{1}-{2}".format(baseurl, radioversion, suffix),
        "{0}/qc8960.wtr-{1}-{2}".format(baseurl, radioversion, suffix),
        "{0}/qc8960.wtr5-{1}-{2}".format(baseurl, radioversion, suffix),
        "{0}/qc8930.wtr5-{1}-{2}".format(baseurl, radioversion, suffix),
        "{0}/qc8974.wtr2-{1}-{2}".format(baseurl, radioversion, suffix)
    ]
    coreurls = []
    splitos = [int(i) for i in osversion.split(".")]
    osurls[2] = filter_1031(osurls[2], splitos, 5)
    osurls[3] = filter_1031(osurls[3], splitos, 6)
    if core:
        for url in osurls:
            coreurls.append(url.replace(".desktop", ""))
    return osurls, radiourls, coreurls


def filter_1031(osurl, splitos, device):
    """
    Modify URLs to reflect changes in 10.3.1+.

    :param osurl: OS URL to modify.
    :type osurl: str

    :param splitos: OS version, split and cast to int: [10, 3, 2, 2876]
    :type splitos: list(int)

    :param device: Device to use.
    :type device: int
    """
    if (splitos[1] >= 4) or (splitos[1] == 3 and splitos[2] >= 1):
        if device == 5:
            osurl = osurl.replace("qc8960.factory_sfi", "qc8960.factory_sfi_hybrid_qc8x30")
        elif device == 6:
            osurl = osurl.replace("qc8974.factory_sfi", "qc8960.factory_sfi_hybrid_qc8974")
    return osurl


def generate_lazy_urls(baseurl, osversion, radioversion, device):
    """
    Generate a pair of OS/radio URLs based on input.

    :param baseurl: The URL, from http to the hashed software release.
    :type baseurl: str

    :param osversion: OS version.
    :type osversion: str

    :param radioversion: Radio version.
    :type radioversion: str

    :param device: Device to use.
    :type device: int
    """
    suffix = "nto+armle-v7+signed.bar"
    splitos = [int(i) for i in osversion.split(".")]
    rads = ["m5730", "qc8960", "qc8960.omadm", "qc8960.wtr",
            "qc8960.wtr5", "qc8930.wtr4", "qc8974.wtr2"]
    oses = ["winchester.factory", "qc8960.factory", "qc8960.verizon",
            "qc8974.factory"]
    maps = {0:0, 1:1, 2:2, 3:1, 4:1, 5:1, 6:3}
    osurl = "{0}/{1}_sfi.desktop-{2}-{3}".format(baseurl, oses[maps[device]], osversion, suffix)
    radiourl = "{0}/{1}-{2}-{3}".format(baseurl, rads[device], radioversion, suffix)
    osurl = filter_1031(osurl, splitos, device)
    return osurl, radiourl


def bulk_urls(baseurl, osversion, radioversion, core=False, alturl=None):
    """
    Generate all URLs, plus extra Verizon URLs.

    :param baseurl: The URL, from http to the hashed software release.
    :type baseurl: str

    :param osversion: OS version.
    :type osversion: str

    :param radioversion: Radio version.
    :type radioversion: str

    :param device: Device to use.
    :type device: int

    :param core: Whether or not to return core URLs as well.
    :type core: bool

    :param alturl: The base URL for any alternate radios.
    :type alturl: str
    """
    osurls, radurls, coreurls = generate_urls(baseurl, osversion, radioversion, core)
    vzwos, vzwrad = generate_lazy_urls(baseurl, osversion, radioversion, 2)
    osurls.append(vzwos)
    radurls.append(vzwrad)
    vzwcore = vzwos.replace("sfi.desktop", "sfi")
    if core:
        coreurls.append(vzwcore)
    osurls = list(set(osurls))  # pop duplicates
    radurls = list(set(radurls))
    if core:
        coreurls = list(set(coreurls))
    if alturl is not None:
        radiourls2 = []
        for rad in radurls:
            radiourls2.append(rad.replace(baseurl, alturl))
        radurls = radiourls2
        del radiourls2
    return osurls, coreurls, radurls


def line_begin():
    """
    Go to beginning of line, to overwrite whatever's there.
    """
    sys.stdout.write("\r")
    sys.stdout.flush()


def spinner_clear():
    """
    Get rid of any spinner residue left in stdout.
    """
    sys.stdout.write("\b \b")
    sys.stdout.flush()


class Spinner(object):
    """
    A basic spinner using itertools. No need for progress.
    """

    def __init__(self):
        self.wheel = itertools.cycle(['-', '/', '|', '\\'])
        self.file = dummy.UselessStdout()

    def after(self):
        """
        Iterate over itertools.cycle, write to file.
        """
        try:
            self.file.write(next(self.wheel))
            self.file.flush()
            self.file.write("\b\r")
            self.file.flush()
        except (KeyboardInterrupt, SystemExit):
            self.stop()

    def stop(self):
        """
        Kill output.
        """
        self.file = dummy.UselessStdout()


class SpinManager(object):
    """
    Wraps around the itertools spinner, runs it in another thread.
    """

    def __init__(self):
        spinner = Spinner()
        self.spinner = spinner
        self.thread = threading.Thread(target=self.loop, args=())
        self.thread.daemon = True
        self.scanning = False
        self.spinner.file = dummy.UselessStdout()

    def start(self):
        """
        Begin the spinner.
        """
        self.spinner.file = sys.stderr
        self.scanning = True
        self.thread.start()

    def loop(self):
        """
        Spin if scanning, clean up if not.
        """
        while self.scanning:
            time.sleep(0.5)
            try:
                line_begin()
                self.spinner.after()
            except (KeyboardInterrupt, SystemExit):
                self.scanning = False
                self.stop()

    def stop(self):
        """
        Stop the spinner.
        """
        self.spinner.stop()
        self.scanning = False
        spinner_clear()
        line_begin()
        if not is_windows():
            print("\n")


def return_and_delete(target):
    """
    Read text file, then delete it. Return contents.

    :param target: Text file to read.
    :type target: str
    """
    with open(target, "r") as thefile:
        content = thefile.read()
    os.remove(target)
    return content


def verify_loader_integrity(loaderfile):
    """
    Test for created loader integrity. Windows-only.

    :param loaderfile: Path to loader.
    :type loaderfile: str
    """
    if not is_windows():
        pass
    else:
        excode = None
        try:
            with open(os.devnull, 'rb') as dnull:
                cmd = "{0} fileinfo".format(loaderfile)
                excode = subprocess.call(cmd, stdout=dnull, stderr=subprocess.STDOUT)
        except OSError:
            excode = -1
        return excode == 0  # 0 if OK, non-zero if something broke


def verify_bulk_loaders(ldir):
    """
    Run :func:`verify_loader_integrity` for all files in a dir.

    :param ldir: Directory to use.
    :type ldir: str
    """
    if not is_windows():
        pass
    else:
        files = [os.path.join(ldir, file) for file in os.listdir(ldir) if not os.path.isdir(file)]
        brokens = []
        for file in files:
            fname = os.path.basename(file)
            if fname.endswith(".exe") and fname.startswith(bbconstants.PREFIXES):
                print("TESTING: {0}".format(fname))
                if not verify_loader_integrity(file):
                    brokens.append(fname)
        return brokens


def workers(input_data):
    """
    Count number of CPU workers, smaller of number of threads and length of data.

    :param input_data: Input data, some iterable.
    :type input_data: list
    """
    runners = len(input_data) if len(input_data) < compat.enum_cpus() else compat.enum_cpus()
    return runners


def prep_logfile():
    """
    Prepare log file, labeling it with current date. Select folder based on frozen status.
    """
    logfile = "{0}.txt".format(time.strftime("%Y_%m_%d_%H%M%S"))
    root = os.getcwd() if getattr(sys, 'frozen', False) else os.path.expanduser("~")
    basefolder = os.path.join(root, "lookuplogs")
    os.makedirs(basefolder, exist_ok=True)
    record = os.path.join(basefolder, logfile)
    open(record, "w").close()
    return record


def prepends(file, pre, suf):
    """
    Check if filename starts with/ends with stuff.

    :param file: File to check.
    :type file: str

    :param pre: Prefix(es) to check.
    :type pre: str or list or tuple

    :param suf: Suffix(es) to check.
    :type suf: str or list or tuple
    """
    return file.startswith(pre) and file.endswith(suf)


def lprint(iterable):
    """
    A oneliner for 'for item in x: print item'.

    :param iterable: Iterable to print.
    :type iterable: list/tuple
    """
    for item in iterable:
        print(item)


def cappath_config_loader(homepath=None):
    """
    Read a ConfigParser file to get cap preferences.

    :param homepath: Folder containing ini file. Default is user directory.
    :type homepath: str
    """
    capini = iniconfig.generic_loader('cappath', homepath)
    cappath = capini.get('path', fallback=bbconstants.CAP.location)
    return cappath


def cappath_config_writer(cappath=None, homepath=None):
    """
    Write a ConfigParser file to store cap preferences.

    :param cappath: Method to use.
    :type cappath: str

    :param homepath: Folder containing ini file. Default is user directory.
    :type homepath: str
    """
    cappath = grab_cap() if cappath is None else cappath
    results = {"path": cappath}
    iniconfig.generic_writer("cappath", results, homepath)


1			#!/usr/bin/env python3
2			"""This module is used for miscellaneous utilities."""
3
4			import os # path work
5			import argparse # argument parser for filters
6			import platform # platform info
7			import glob # cap grabbing
8			import threading # get thread for spinner
9			import time # spinner delay
10			import sys # streams, version info
11			import itertools # spinners gonna spin
12			import subprocess # loader verification
13			from bbarchivist import bbconstants # cap location, version, filename bits
14			from bbarchivist import compat # backwards compat
15			from bbarchivist import dummy # useless stdout
16			from bbarchivist import iniconfig # config parsing
17
18			__author__ = "Thurask"
19			__license__ = "WTFPL v2"
20			__copyright__ = "Copyright 2015-2016 Thurask"
21
22
23			def grab_cap():
24			"""
25			Figure out where cap is, local, specified or system-supplied.
26			"""
27			try:
28			capfile = glob.glob(os.path.join(os.getcwd(), bbconstants.CAP.filename))[0]
29			except IndexError:
30			try:
31			cappath = cappath_config_loader()
32			capfile = glob.glob(cappath)[0]
33			except IndexError:
34			cappath_config_writer(bbconstants.CAP.location)
35			return bbconstants.CAP.location # no ini cap
36			else:
37			cappath_config_writer(os.path.abspath(capfile))
38			return os.path.abspath(capfile) # ini cap
39			else:
40			return os.path.abspath(capfile) # local cap
41
42
43			def grab_cfp():
44			"""
45			Figure out where cfp is, local or system-supplied.
46			"""
47			try:
48			cfpfile = glob.glob(os.path.join(os.getcwd(), bbconstants.CFP.filename))[0]
49			except IndexError:
50			cfpfile = bbconstants.CFP.location # system cfp
51			return os.path.abspath(cfpfile) # local cfp
52
53
54			def new_enough(minver):
55			"""
56			Check if we're at or above a minimum Python version.
57
58			:param minver: Minimum Python version (3.minver).
59			:type minver: int
60			"""
61			return False if minver > sys.version_info[1] else True
62
63
64			def fsizer(file_size):
65			"""
66			Raw byte file size to human-readable string.
67
68			:param file_size: Number to parse.
69			:type file_size: float
70			"""
71			if file_size is None:
72			file_size = 0
73			fsize = float(file_size)
74			for sfix in ['B', 'kB', 'MB', 'GB', 'TB', 'PB', 'EB', 'ZB']:
75			if fsize < 1024.0:
76			size = "{0:3.2f}{1}".format(fsize, sfix)
77			break
78			else:
79			fsize /= 1024.0
80			else:
81			size = "{0:3.2f}{1}".format(fsize, 'YB')
82			return size
83
84
85			def signed_file_args(files):
86			"""
87			Check if there are between 1 and 6 files supplied to argparse.
88
89			:param files: List of signed files, between 1 and 6 strings.
90			:type files: list(str)
91			"""
92			filelist = [file for file in files if file]
93			if not 1 <= len(filelist) <= 6:
94			raise argparse.ArgumentError(argument=None, message="Requires 1-6 signed files")
95			return files
96
97
98			def file_exists(file):
99			"""
100			Check if file exists, raise argparse error if it doesn't.
101
102			:param file: Path to a file, including extension.
103			:type file: str
104			"""
105			if not os.path.exists(file):
106			raise argparse.ArgumentError(argument=None, message="{0} not found.".format(file))
107			return file
108
109
110			def positive_integer(input_int):
111			"""
112			Check if number > 0, raise argparse error if it isn't.
113
114			:param input_int: Integer to check.
115			:type input_int: str
116			"""
117			if int(input_int) <= 0:
118			info = "{0} is not >=0.".format(str(input_int))
119			raise argparse.ArgumentError(argument=None, message=info)
120			return int(input_int)
121
122
123			def valid_method(method):
124			"""
125			Check if compression method is valid, raise argparse error if it isn't.
126
127			:param method: Compression method to check.
128			:type method: str
129			"""
130			methodlist = bbconstants.METHODS
131			if not new_enough(3):
132			methodlist = [x for x in bbconstants.METHODS if x != "txz"]
133			if method not in methodlist:
134			info = "Invalid method {0}.".format(method)
135			raise argparse.ArgumentError(argument=None, message=info)
136			return method
137
138
139			def valid_carrier(mcc_mnc):
140			"""
141			Check if MCC/MNC is valid (1-3 chars), raise argparse error if it isn't.
142
143			:param mcc_mnc: MCC/MNC to check.
144			:type mcc_mnc: str
145			"""
146			if not str(mcc_mnc).isdecimal():
147			infod = "Non-integer {0}.".format(str(mcc_mnc))
148			raise argparse.ArgumentError(argument=None, message=infod)
149			if len(str(mcc_mnc)) > 3 or len(str(mcc_mnc)) == 0:
150			infol = "{0} is invalid.".format(str(mcc_mnc))
151			raise argparse.ArgumentError(argument=None, message=infol)
152			else:
153			return mcc_mnc
154
155
156			def escreens_pin(pin):
157			"""
158			Check if given PIN is valid, raise argparse error if it isn't.
159
160			:param pin: PIN to check.
161			:type pin: str
162			"""
163			if len(pin) == 8:
164			try:
165			int(pin, 16) # hexadecimal-ness
166			except ValueError:
167			raise argparse.ArgumentError(argument=None, message="Invalid PIN.")
168			else:
169			return pin.lower()
170			else:
171			raise argparse.ArgumentError(argument=None, message="Invalid PIN.")
172
173
174			def escreens_duration(duration):
175			"""
176			Check if Engineering Screens duration is valid.
177
178			:param duration: Duration to check.
179			:type duration: int
180			"""
181			if int(duration) in (1, 3, 6, 15, 30):
182			return int(duration)
183			else:
184			raise argparse.ArgumentError(argument=None, message="Invalid duration.")
185
186
187			def droidlookup_hashtype(method):
188			"""
189			Check if Android autoloader lookup hash type is valid.
190
191			:param method: None for regular OS links, "sha256/512" for SHA256 or 512 hash.
192			:type method: str
193			"""
194			if method.lower() in ("sha512", "sha256"):
195			return method.lower()
196			else:
197			raise argparse.ArgumentError(argument=None, message="Invalid type.")
198
199
200			def droidlookup_devicetype(device):
201			"""
202			Check if Android autoloader device type is valid.
203
204			:param device: Android autoloader types to check.
205			:type device: str
206			"""
207			devices = ("Priv", "DTEK50")
208			#devices = ("Priv", "DTEK50", "DTEK60")
209			if device is None:
210			return None
211			else:
212			for dev in devices:
213			if dev.lower() == device.lower():
214			return dev
215			raise argparse.ArgumentError(argument=None, message="Invalid device.")
216
217
218			def s2b(input_check):
219			"""
220			Return Boolean interpretation of string input.
221
222			:param input_check: String to check if it means True or False.
223			:type input_check: str
224			"""
225			return str(input_check).lower() in ("yes", "true", "t", "1", "y")
226
227
228			def is_amd64():
229			"""
230			Check if script is running on an AMD64 system.
231			"""
232			return platform.machine().endswith("64")
233
234
235			def is_windows():
236			"""
237			Check if script is running on Windows.
238			"""
239			return platform.system() == "Windows"
240
241
242			def get_seven_zip(talkative=False):
243			"""
244			Return name of 7-Zip executable.
245			On POSIX, it MUST be 7za.
246			On Windows, it can be installed or supplied with the script.
247			:func:`win_seven_zip` is used to determine if it's installed.
248
249			:param talkative: Whether to output to screen. False by default.
250			:type talkative: bool
251			"""
252			return win_seven_zip(talkative) if is_windows() else "7za"
253
254
255			def win_seven_zip(talkative=False):
256			"""
257			For Windows, check where 7-Zip is ("where", pretty much).
258			Consult registry first for any installed instances of 7-Zip.
259
260			:param talkative: Whether to output to screen. False by default.
261			:type talkative: bool
262			"""
263			if talkative:
264			print("CHECKING INSTALLED FILES...")
265			try:
266			import winreg # windows registry
267			hk7z = winreg.OpenKey(winreg.HKEY_CURRENT_USER, "Software\\7-Zip")
268			path = winreg.QueryValueEx(hk7z, "Path")
269			except OSError as exc:
270			if talkative:
271			print("SOMETHING WENT WRONG")
272			print(str(exc))
273			print("TRYING LOCAL FILES...")
274			return win_seven_zip_local(talkative)
275			else:
276			if talkative:
277			print("7ZIP USING INSTALLED FILES")
278			return '"{0}"'.format(os.path.join(path[0], "7z.exe"))
279
280
281			def win_seven_zip_local(talkative=False):
282			"""
283			If 7-Zip isn't in the registry, fall back onto supplied executables.
284			If those aren't there, return "error".
285
286			:param talkative: Whether to output to screen. False by default.
287			:type talkative: bool
288			"""
289			listdir = os.listdir(os.getcwd())
290			filecount = 0
291			for i in listdir:
292			if i in ["7za.exe", "7za64.exe"]:
293			filecount += 1
294			if filecount == 2:
295			if talkative:
296			print("7ZIP USING LOCAL FILES")
297			if is_amd64():
298			return "7za64.exe"
299			else:
300			return "7za.exe"
301			else:
302			if talkative:
303			print("NO LOCAL FILES")
304			return "error"
305
306
307			def get_core_count():
308			"""
309			Find out how many CPU cores this system has.
310			"""
311			try:
312			cores = str(compat.enum_cpus()) # 3.4 and up
313			except NotImplementedError:
314			cores = "1" # 3.2-3.3
315			else:
316			if compat.enum_cpus() is None:
317			cores = "1"
318			return cores
319
320
321			def prep_seven_zip_path(path, talkative=False):
322			"""
323			Print p7zip path on POSIX, or notify if not there.
324
325			:param path: Path to use.
326			:type path: str
327
328			:param talkative: Whether to output to screen. False by default.
329			:type talkative: bool
330			"""
331			if path is None:
332			if talkative:
333			print("NO 7ZIP")
334			print("PLEASE INSTALL p7zip")
335			return False
336			else:
337			if talkative:
338			print("7ZIP FOUND AT {0}".format(path))
339			return True
340
341
342			def prep_seven_zip_posix(talkative=False):
343			"""
344			Check for p7zip on POSIX.
345
346			:param talkative: Whether to output to screen. False by default.
347			:type talkative: bool
348			"""
349			try:
350			path = compat.where_which("7za")
351			except ImportError:
352			if talkative:
353			print("PLEASE INSTALL SHUTILWHICH WITH PIP")
354			return False
355			else:
356			return prep_seven_zip_path(path, talkative)
357
358
359			def prep_seven_zip(talkative=False):
360			"""
361			Check for presence of 7-Zip.
362			On POSIX, check for p7zip.
363			On Windows, check for 7-Zip.
364
365			:param talkative: Whether to output to screen. False by default.
366			:type talkative: bool
367			"""
368			if is_windows():
369			return get_seven_zip(talkative) != "error"
370			else:
371			return prep_seven_zip_posix(talkative)
372
373
374			def increment(version, inc=3):
375			"""
376			Increment version by given number. For repeated lookups.
377
378			:param version: w.x.y.ZZZZ, becomes w.x.y.(ZZZZ + increment).
379			:type version: str
380
381			:param inc: What to increment by. Default is 3.
382			:type inc: str
383			"""
384			splitos = version.split(".")
385			splitos[3] = int(splitos[3])
386			if splitos[3] > 9996: # prevent overflow
387			splitos[3] = 0
388			splitos[3] += int(inc)
389			splitos[3] = str(splitos[3])
390			return ".".join(splitos)
391
392
393			def stripper(name):
394			"""
395			Strip fluff from bar filename.
396
397			:param name: Bar filename, must contain '-nto+armle-v7+signed.bar'.
398			:type name: str
399			"""
400			return name.replace("-nto+armle-v7+signed.bar", "")
401
402
403			def generate_urls(baseurl, osversion, radioversion, core=False):
404			"""
405			Generate a list of OS URLs and a list of radio URLs based on input.
406
407			:param baseurl: The URL, from http to the hashed software release.
408			:type baseurl: str
409
410			:param osversion: OS version.
411			:type osversion: str
412
413			:param radioversion: Radio version.
414			:type radioversion: str
415
416			:param core: Whether or not to return core URLs as well.
417			:type core: bool
418			"""
419			suffix = "nto+armle-v7+signed.bar"
420			osurls = [
421			"{0}/winchester.factory_sfi.desktop-{1}-{2}".format(baseurl, osversion, suffix),
422			"{0}/qc8960.factory_sfi.desktop-{1}-{2}".format(baseurl, osversion, suffix),
423			"{0}/qc8960.factory_sfi.desktop-{1}-{2}".format(baseurl, osversion, suffix),
424			"{0}/qc8974.factory_sfi.desktop-{1}-{2}".format(baseurl, osversion, suffix)
425			]
426			radiourls = [
427			"{0}/m5730-{1}-{2}".format(baseurl, radioversion, suffix),
428			"{0}/qc8960-{1}-{2}".format(baseurl, radioversion, suffix),
429			"{0}/qc8960.omadm-{1}-{2}".format(baseurl, radioversion, suffix),
430			"{0}/qc8960.wtr-{1}-{2}".format(baseurl, radioversion, suffix),
431			"{0}/qc8960.wtr5-{1}-{2}".format(baseurl, radioversion, suffix),
432			"{0}/qc8930.wtr5-{1}-{2}".format(baseurl, radioversion, suffix),
433			"{0}/qc8974.wtr2-{1}-{2}".format(baseurl, radioversion, suffix)
434			]
435			coreurls = []
436			splitos = [int(i) for i in osversion.split(".")]
437			osurls[2] = filter_1031(osurls[2], splitos, 5)
438			osurls[3] = filter_1031(osurls[3], splitos, 6)
439			if core:
440			for url in osurls:
441			coreurls.append(url.replace(".desktop", ""))
442			return osurls, radiourls, coreurls
443
444
445			def filter_1031(osurl, splitos, device):
446			"""
447			Modify URLs to reflect changes in 10.3.1+.
448
449			:param osurl: OS URL to modify.
450			:type osurl: str
451
452			:param splitos: OS version, split and cast to int: [10, 3, 2, 2876]
453			:type splitos: list(int)
454
455			:param device: Device to use.
456			:type device: int
457			"""
458			if (splitos[1] >= 4) or (splitos[1] == 3 and splitos[2] >= 1):
459			if device == 5:
460			osurl = osurl.replace("qc8960.factory_sfi", "qc8960.factory_sfi_hybrid_qc8x30")
461			elif device == 6:
462			osurl = osurl.replace("qc8974.factory_sfi", "qc8960.factory_sfi_hybrid_qc8974")
463			return osurl
464
465
466			def generate_lazy_urls(baseurl, osversion, radioversion, device):
467			"""
468			Generate a pair of OS/radio URLs based on input.
469
470			:param baseurl: The URL, from http to the hashed software release.
471			:type baseurl: str
472
473			:param osversion: OS version.
474			:type osversion: str
475
476			:param radioversion: Radio version.
477			:type radioversion: str
478
479			:param device: Device to use.
480			:type device: int
481			"""
482			suffix = "nto+armle-v7+signed.bar"
483			splitos = [int(i) for i in osversion.split(".")]
484			rads = ["m5730", "qc8960", "qc8960.omadm", "qc8960.wtr",
485			"qc8960.wtr5", "qc8930.wtr4", "qc8974.wtr2"]
486			oses = ["winchester.factory", "qc8960.factory", "qc8960.verizon",
487			"qc8974.factory"]
488			maps = {0:0, 1:1, 2:2, 3:1, 4:1, 5:1, 6:3}
489			osurl = "{0}/{1}_sfi.desktop-{2}-{3}".format(baseurl, oses[maps[device]], osversion, suffix)
490			radiourl = "{0}/{1}-{2}-{3}".format(baseurl, rads[device], radioversion, suffix)
491			osurl = filter_1031(osurl, splitos, device)
492			return osurl, radiourl
493
494
495			def bulk_urls(baseurl, osversion, radioversion, core=False, alturl=None):
496			"""
497			Generate all URLs, plus extra Verizon URLs.
498
499			:param baseurl: The URL, from http to the hashed software release.
500			:type baseurl: str
501
502			:param osversion: OS version.
503			:type osversion: str
504
505			:param radioversion: Radio version.
506			:type radioversion: str
507
508			:param device: Device to use.
509			:type device: int
510
511			:param core: Whether or not to return core URLs as well.
512			:type core: bool
513
514			:param alturl: The base URL for any alternate radios.
515			:type alturl: str
516			"""
517			osurls, radurls, coreurls = generate_urls(baseurl, osversion, radioversion, core)
518			vzwos, vzwrad = generate_lazy_urls(baseurl, osversion, radioversion, 2)
519			osurls.append(vzwos)
520			radurls.append(vzwrad)
521			vzwcore = vzwos.replace("sfi.desktop", "sfi")
522			if core:
523			coreurls.append(vzwcore)
524			osurls = list(set(osurls)) # pop duplicates
525			radurls = list(set(radurls))
526			if core:
527			coreurls = list(set(coreurls))
528			if alturl is not None:
529			radiourls2 = []
530			for rad in radurls:
531			radiourls2.append(rad.replace(baseurl, alturl))
532			radurls = radiourls2
533			del radiourls2
534			return osurls, coreurls, radurls
535
536
537			def line_begin():
538			"""
539			Go to beginning of line, to overwrite whatever's there.
540			"""
541			sys.stdout.write("\r")
542			sys.stdout.flush()
543
544
545			def spinner_clear():
546			"""
547			Get rid of any spinner residue left in stdout.
548			"""
549			sys.stdout.write("\b \b")
550			sys.stdout.flush()
551
552
553			class Spinner(object):
554			"""
555			A basic spinner using itertools. No need for progress.
556			"""
557
558			def __init__(self):
559			self.wheel = itertools.cycle(['-', '/', '\|', '\\'])
560			self.file = dummy.UselessStdout()
561
562			def after(self):
563			"""
564			Iterate over itertools.cycle, write to file.
565			"""
566			try:
567			self.file.write(next(self.wheel))
568			self.file.flush()
569			self.file.write("\b\r")
570			self.file.flush()
571			except (KeyboardInterrupt, SystemExit):
572			self.stop()
573
574			def stop(self):
575			"""
576			Kill output.
577			"""
578			self.file = dummy.UselessStdout()
579
580
581			class SpinManager(object):
582			"""
583			Wraps around the itertools spinner, runs it in another thread.
584			"""
585
586			def __init__(self):
587			spinner = Spinner()
588			self.spinner = spinner
589			self.thread = threading.Thread(target=self.loop, args=())
590			self.thread.daemon = True
591			self.scanning = False
592			self.spinner.file = dummy.UselessStdout()
593
594			def start(self):
595			"""
596			Begin the spinner.
597			"""
598			self.spinner.file = sys.stderr
599			self.scanning = True
600			self.thread.start()
601
602			def loop(self):
603			"""
604			Spin if scanning, clean up if not.
605			"""
606			while self.scanning:
607			time.sleep(0.5)
608			try:
609			line_begin()
610			self.spinner.after()
611			except (KeyboardInterrupt, SystemExit):
612			self.scanning = False
613			self.stop()
614
615			def stop(self):
616			"""
617			Stop the spinner.
618			"""
619			self.spinner.stop()
620			self.scanning = False
621			spinner_clear()
622			line_begin()
623			if not is_windows():
624			print("\n")
625
626
627			def return_and_delete(target):
628			"""
629			Read text file, then delete it. Return contents.
630
631			:param target: Text file to read.
632			:type target: str
633			"""
634			with open(target, "r") as thefile:
635			content = thefile.read()
636			os.remove(target)
637			return content
638
639
640			def verify_loader_integrity(loaderfile):
641			"""
642			Test for created loader integrity. Windows-only.
643
644			:param loaderfile: Path to loader.
645			:type loaderfile: str
646			"""
647			if not is_windows():
648			pass
649			else:
650			excode = None
651			try:
652			with open(os.devnull, 'rb') as dnull:
653			cmd = "{0} fileinfo".format(loaderfile)
654			excode = subprocess.call(cmd, stdout=dnull, stderr=subprocess.STDOUT)
655			except OSError:
656			excode = -1
657			return excode == 0 # 0 if OK, non-zero if something broke
658
659
660			def verify_bulk_loaders(ldir):
661			"""
662			Run :func:`verify_loader_integrity` for all files in a dir.
663
664			:param ldir: Directory to use.
665			:type ldir: str
666			"""
667			if not is_windows():
668			pass
669			else:
670			files = [os.path.join(ldir, file) for file in os.listdir(ldir) if not os.path.isdir(file)]
671			brokens = []
672			for file in files:
673			fname = os.path.basename(file)
674			if fname.endswith(".exe") and fname.startswith(bbconstants.PREFIXES):
675			print("TESTING: {0}".format(fname))
676			if not verify_loader_integrity(file):
677			brokens.append(fname)
678			return brokens
679
680
681			def workers(input_data):
682			"""
683			Count number of CPU workers, smaller of number of threads and length of data.
684
685			:param input_data: Input data, some iterable.
686			:type input_data: list
687			"""
688			runners = len(input_data) if len(input_data) < compat.enum_cpus() else compat.enum_cpus()
689			return runners
690
691
692			def prep_logfile():
693			"""
694			Prepare log file, labeling it with current date. Select folder based on frozen status.
695			"""
696			logfile = "{0}.txt".format(time.strftime("%Y_%m_%d_%H%M%S"))
697			root = os.getcwd() if getattr(sys, 'frozen', False) else os.path.expanduser("~")
698			basefolder = os.path.join(root, "lookuplogs")
699			os.makedirs(basefolder, exist_ok=True)
700			record = os.path.join(basefolder, logfile)
701			open(record, "w").close()
702			return record
703
704
705			def prepends(file, pre, suf):
706			"""
707			Check if filename starts with/ends with stuff.
708
709			:param file: File to check.
710			:type file: str
711
712			:param pre: Prefix(es) to check.
713			:type pre: str or list or tuple
714
715			:param suf: Suffix(es) to check.
716			:type suf: str or list or tuple
717			"""
718			return file.startswith(pre) and file.endswith(suf)
719
720
721			def lprint(iterable):
722			"""
723			A oneliner for 'for item in x: print item'.
724
725			:param iterable: Iterable to print.
726			:type iterable: list/tuple
727			"""
728			for item in iterable:
729			print(item)
730
731
732			def cappath_config_loader(homepath=None):
733			"""
734			Read a ConfigParser file to get cap preferences.
735
736			:param homepath: Folder containing ini file. Default is user directory.
737			:type homepath: str
738			"""
739			capini = iniconfig.generic_loader('cappath', homepath)
740			cappath = capini.get('path', fallback=bbconstants.CAP.location)
741			return cappath
742
743
744			def cappath_config_writer(cappath=None, homepath=None):
745			"""
746			Write a ConfigParser file to store cap preferences.
747
748			:param cappath: Method to use.
749			:type cappath: str
750
751			:param homepath: Folder containing ini file. Default is user directory.
752			:type homepath: str
753			"""
754			cappath = grab_cap() if cappath is None else cappath
755			results = {"path": cappath}
756			iniconfig.generic_writer("cappath", results, homepath)
757

thurask / bbarchivist

Push — master ( d56c20...c7f92e )

cappath_config_writer() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like