1
0
mirror of https://github.com/mbirth/tcl_ota_check.git synced 2024-09-19 22:33:25 +01:00

docstring all the things

This commit is contained in:
thurask 2018-02-03 16:25:26 -05:00
parent 080214ec44
commit 1303fa863b
No known key found for this signature in database
GPG Key ID: A6CCCDEA29795048
22 changed files with 101 additions and 11 deletions

View File

@ -3,6 +3,8 @@
# pylint: disable=C0111,C0326,C0103
"""Checks for the latest FULL or OTA updates for specified PRD number."""
import os
import random
import sys
@ -31,6 +33,7 @@ args = dp.parse_args(sys.argv[1:])
def sel_mode(txtmode, autoval, rawval):
"""Handle custom mode."""
if rawval:
enum = tcllib.default_enum("MODE", {"RAW": rawval})
return enum.RAW
@ -42,6 +45,7 @@ def sel_mode(txtmode, autoval, rawval):
def sel_cltp(txtmode, autoval, rawval):
"""Handle custom CLTP."""
if rawval:
enum = tcllib.default_enum("CLTP", {"RAW": rawval})
return enum.RAW

View File

@ -3,6 +3,8 @@
# pylint: disable=C0111,C0326,C0103
"""Check all/given PRDs for FULL updates."""
import sys
from requests.exceptions import RequestException

View File

@ -3,6 +3,8 @@
# pylint: disable=C0111,C0326,C0103
"""Check all/given PRDs for OTA updates."""
import sys
from requests.exceptions import RequestException

View File

@ -3,6 +3,8 @@
# pylint: disable=C0111,C0326,C0103
"""Find new PRDs for a given variant(s)."""
import collections
import sys

View File

@ -3,6 +3,8 @@
# pylint: disable=C0111,C0326,C0103
"""Find new PRDs for a range of variants."""
import sys
from requests.exceptions import RequestException, Timeout

View File

@ -3,6 +3,8 @@
# pylint: disable=C0111,C0326,C0103
"""Find all OTA updates for a given PRD."""
import sys
from requests.exceptions import RequestException, Timeout

View File

@ -3,6 +3,8 @@
# pylint: disable=C0111,C0326,C0103
"""Query existence of missing OTAs."""
import json
import requests

View File

@ -3,6 +3,8 @@
# pylint: disable=C0111,C0326,C0103
"""Return checksum for given firmware."""
import random
import sys

View File

@ -3,6 +3,8 @@
# pylint: disable=C0111,C0326,C0103
"""Download a given firmware file."""
import os
import random
import sys
@ -31,6 +33,7 @@ args = dp.parse_args(sys.argv[1:])
def sel_mode(defaultmode, rawval):
"""Handle custom mode."""
if rawval:
enum = tcllib.default_enum("MODE", {"RAW": rawval})
return enum.RAW
@ -38,6 +41,7 @@ def sel_mode(defaultmode, rawval):
def sel_cltp(txtmode, rawval):
"""Handle custom CLTP."""
if rawval:
enum = tcllib.default_enum("CLTP", {"RAW": rawval})
return enum.RAW

View File

@ -3,6 +3,8 @@
# pylint: disable=C0111,C0326,C0103
"""Library for TCL API work and related functions."""
import enum
import requests
@ -12,20 +14,22 @@ from . import (credentials, devlist, dumpmgr, servervote, tclcheck,
def default_enum(enumname, vardict, qualroot="tcllib.FotaCheck"):
"""Enum with defaults set."""
return enum.IntEnum(enumname, vardict, module=__name__, qualname="{}.{}".format(qualroot, enumname))
class FotaCheck(
tclcheck.TclCheckMixin,
tclrequest.TclRequestMixin,
tclchecksum.TclChecksumMixin,
tclencheader.TclEncHeaderMixin,
servervote.ServerVoteMixin,
credentials.CredentialsMixin,
devlist.DevListMixin,
dumpmgr.DumpMgrMixin,
xmltools.XmlToolsMixin
tclcheck.TclCheckMixin,
tclrequest.TclRequestMixin,
tclchecksum.TclChecksumMixin,
tclencheader.TclEncHeaderMixin,
servervote.ServerVoteMixin,
credentials.CredentialsMixin,
devlist.DevListMixin,
dumpmgr.DumpMgrMixin,
xmltools.XmlToolsMixin
):
"""Main API handler class."""
VDKEY = b"eJwdjwEOwDAIAr8kKFr//7HhmqXp8AIIDrYAgg8byiUXrwRJRXja+d6iNxu0AhUooDCN9rd6rDLxmGIakUVWo3IGCTRWqCAt6X4jGEIUAxgN0eYWnp+LkpHQAg/PsO90ELsy0Npm/n2HbtPndFgGEV31R9OmT4O4nrddjc3Qt6nWscx7e+WRHq5UnOudtjw5skuV09pFhvmqnOEIs4ljPeel1wfLYUF4\n"
CKTP = default_enum("CKTP", ["AUTO", "MANUAL"])
MODE = default_enum("MODE", {"OTA": 2, "FULL": 4})
@ -35,6 +39,7 @@ class FotaCheck(
CKOT = default_enum("CKOT", ["ALL", "AOTA_ONLY", "FOTA_ONLY"])
def __init__(self):
"""Handle mixins and populate variables."""
super().__init__()
self.serid = "543212345000000"
self.curef = "PRD-63117-011"
@ -50,6 +55,7 @@ class FotaCheck(
self.reset_session()
def reset_session(self):
"""Reset everything to default."""
self.g2master = self.get_master_server()
self.sess = requests.Session()
if self.mode == self.MODE.FULL:

View File

@ -3,6 +3,8 @@
# pylint: disable=C0111,C0326,C0103
"""Listing of ANSI colors plus additional Windows support."""
import platform
# Needed to make ANSI escape sequences work in Windows

View File

@ -3,16 +3,22 @@
# pylint: disable=C0111,C0326,C0103
"""Custom argument parser."""
import argparse
import webbrowser
class DefaultParser(argparse.ArgumentParser):
"""argparse parser with some defaults set."""
def __init__(self, appname, desc=None):
super().__init__(prog=appname, description=desc, epilog="https://github.com/mbirth/tcl_ota_check")
"""Set default name, description, epilogue, arguments."""
homeurl = "https://github.com/mbirth/tcl_ota_check"
super().__init__(prog=appname, description=desc, epilog=homeurl)
self.add_argument("--webdb", help="open web database in browser and exit", action="store_true")
def parse_args(self, args=None, namespace=None):
"""Parse special args first, defer to parent class second."""
if set(args) & {"--webdb"}: # if they intersect
webbrowser.open("https://tclota.birth-online.de/", new=2)
raise SystemExit

View File

@ -3,12 +3,16 @@
# pylint: disable=C0111,C0326,C0103
"""Tools to manage request authentication."""
import base64
class CredentialsMixin:
"""A mixin component to provide authentication."""
@staticmethod
def get_creds():
"""Return main authentication."""
creds = {
b"YWNjb3VudA==": b"emhlbmdodWEuZ2Fv",
b"cGFzc3dvcmQ=": b"cWFydUQ0b2s=",
@ -18,6 +22,7 @@ class CredentialsMixin:
@staticmethod
def get_creds2():
"""Return alternate authentication."""
creds = {
b"YWNjb3VudA==": b"VGVsZUV4dFRlc3Q=",
b"cGFzc3dvcmQ=": b"dDA1MjM=",

View File

@ -3,6 +3,8 @@
# pylint: disable=C0111,C0326,C0103
"""Tools to manage saved device databases."""
import json
import os
import time
@ -18,8 +20,10 @@ DEVICELIST_CACHE_SECONDS = 86400
class DevListMixin:
"""A mixin component for device list management."""
@staticmethod
def get_devicelist(force=False, output_diff=True):
"""Return device list from saved database."""
need_download = True
old_prds = None
@ -48,6 +52,7 @@ class DevListMixin:
@staticmethod
def print_prd_diff(old_prds, new_prds):
"""Print changes between old and new databases."""
added_prds = [prd for prd in new_prds if prd not in old_prds]
removed_prds = [prd for prd in old_prds if prd not in new_prds]
for prd in removed_prds:

View File

@ -3,6 +3,8 @@
# pylint: disable=C0111,C0326,C0103
"""Tools to manage dumps of API requests."""
import errno
import glob
import os
@ -11,10 +13,13 @@ from . import ansi
class DumpMgrMixin:
"""A mixin component for XML dump management."""
def __init__(self):
"""Populate dump file name."""
self.last_dump_filename = None
def write_dump(self, data):
"""Write dump to file."""
outfile = os.path.normpath("logs/{}.xml".format(self.get_salt()))
if not os.path.exists(os.path.dirname(outfile)):
try:
@ -27,12 +32,14 @@ class DumpMgrMixin:
self.last_dump_filename = outfile
def delete_last_dump(self):
"""Delete last dump."""
if self.last_dump_filename:
os.unlink(self.last_dump_filename)
self.last_dump_filename = None
@staticmethod
def write_info_if_dumps_found():
"""Notify user to upload dumps if present."""
# To disable this info, uncomment the following line.
# return
files = glob.glob(os.path.normpath("logs/*.xml"))

View File

@ -3,11 +3,15 @@
# pylint: disable=C0111,C0326,C0103
"""Tools to sort API servers to find the least awful one."""
import numpy
class ServerVoteMixin:
"""A mixin component for server sorting."""
def __init__(self):
"""Populate server list and weighting variables."""
self.g2master = None
self.master_servers = [
"g2master-us-east.tclclouds.com",
@ -22,6 +26,7 @@ class ServerVoteMixin:
self.check_time_count = 1
def get_master_server(self):
"""Return weighted choice from server list."""
weight_sum = 0
for i in self.master_servers_weights:
weight_sum += i
@ -31,23 +36,28 @@ class ServerVoteMixin:
return numpy.random.choice(self.master_servers, p=numpy_weights)
def master_server_downvote(self):
"""Decrease weight of a server."""
idx = self.master_servers.index(self.g2master)
if self.master_servers_weights[idx] > 1:
self.master_servers_weights[idx] -= 1
def master_server_upvote(self):
"""Increase weight of a server."""
idx = self.master_servers.index(self.g2master)
if self.master_servers_weights[idx] < 10:
self.master_servers_weights[idx] += 1
def check_time_add(self, duration):
"""Record connection time."""
self.check_time_sum += duration
self.check_time_count += 1
def check_time_avg(self):
"""Return average connection time."""
return self.check_time_sum / self.check_time_count
def master_server_vote_on_time(self, last_duration, avg_duration):
"""Change weight of a server based on average connection time."""
if last_duration < avg_duration - 0.5:
self.master_server_upvote()
elif last_duration > avg_duration + 0.5:

View File

@ -3,6 +3,8 @@
# pylint: disable=C0111,C0326,C0103
"""Tools to interface with TCL's update request API."""
import time
from collections import OrderedDict
@ -11,13 +13,15 @@ from defusedxml import ElementTree
class TclCheckMixin:
"""A mixin component for TCL's update request API."""
def do_check(self, https=True, timeout=10, max_tries=5):
"""Perform update request with given parameters."""
protocol = "https://" if https else "http://"
url = protocol + self.g2master + "/check.php"
params = OrderedDict()
params["id"] = self.serid
params["curef"] = self.curef
params["fv"] = self.fvver
params["fv"] = self.fv
params["mode"] = self.mode.value
params["type"] = self.ftype
params["cltp"] = self.cltp.value
@ -62,6 +66,7 @@ class TclCheckMixin:
@staticmethod
def parse_check(xmlstr):
"""Parse output of ``do_check``."""
root = ElementTree.fromstring(xmlstr)
curef = root.find("CUREF").text
fvver = root.find("VERSION").find("FV").text

View File

@ -3,13 +3,17 @@
# pylint: disable=C0111,C0326,C0103
"""Tools to interface with TCL's checksum API."""
import json
from defusedxml import ElementTree
class TclChecksumMixin:
"""A mixin component for TCL's checksum API."""
def do_checksum(self, encslave, address, uri):
"""Perform checksum request with given parameters."""
url = "http://" + encslave + "/checksum.php"
params = self.get_creds2()
@ -35,6 +39,7 @@ class TclChecksumMixin:
@staticmethod
def parse_checksum(xmlstr):
"""Parse output of ``do_checksum``."""
root = ElementTree.fromstring(xmlstr)
file = root.find("FILE_CHECKSUM_LIST").find("FILE")
file_addr = file.find("ADDRESS").text

View File

@ -3,9 +3,13 @@
# pylint: disable=C0111,C0326,C0103
"""Tools to interface with TCL's encrypted header API."""
class TclEncHeaderMixin:
"""A mixin component for TCL's encrypted header API.."""
def do_encrypt_header(self, encslave, address):
"""Perform encrypted header request with given parameters."""
params = self.get_creds2()
params[b"address"] = bytes(address, "utf-8")
url = "http://" + encslave + "/encrypt_header.php"

View File

@ -3,6 +3,8 @@
# pylint: disable=C0111,C0326,C0103
"""Tools to interface with TCL's download request API."""
import binascii
import hashlib
import random
@ -38,13 +40,16 @@ from defusedxml import ElementTree
class TclRequestMixin:
"""A mixin component for TCL's download request API."""
@staticmethod
def get_salt():
"""Generate cryptographic salt."""
millis = floor(time.time() * 1000)
tail = "{:06d}".format(random.randint(0, 999999))
return "{}{}".format(str(millis), tail)
def get_vk2(self, params_dict, cltp):
"""Generate salted hash of API parameters."""
params_dict["cltp"] = cltp
query = ""
for key, val in params_dict.items():
@ -59,6 +64,7 @@ class TclRequestMixin:
return hexhash
def do_request(self, curef, fvver, tvver, fw_id):
"""Perform download request with given parameters."""
url = "https://" + self.g2master + "/download_request.php"
params = OrderedDict()
params["id"] = self.serid
@ -91,6 +97,7 @@ class TclRequestMixin:
@staticmethod
def parse_request(xmlstr):
"""Parse output of ``do_request``."""
root = ElementTree.fromstring(xmlstr)
file = root.find("FILE_LIST").find("FILE")
fileid = file.find("FILE_ID").text

View File

@ -3,11 +3,15 @@
# pylint: disable=C0111,C0326,C0103
"""XML tools."""
import xml.dom.minidom
class XmlToolsMixin:
"""A mixin component for XML tools."""
@staticmethod
def pretty_xml(xmlstr):
"""Prettify input XML with ``xml.dom.minidom``."""
mdx = xml.dom.minidom.parseString(xmlstr)
return mdx.toprettyxml(indent=" ")

View File

@ -3,6 +3,8 @@
# pylint: disable=C0111,C0326,C0103
"""Upload contents of logs folder to remote database."""
# curl -v -H "Content-Type: text/plain" --data @test.xml http://example.org/tcl_update_db/
import glob