1
0
mirror of https://github.com/mbirth/tcl_ota_check.git synced 2024-11-10 06:16:46 +00:00

Turned DevListMixin into namespaced methods.

This commit is contained in:
Markus Birth 2018-02-05 16:32:26 +01:00
parent 0262c67531
commit 7332846805
Signed by: mbirth
GPG Key ID: A9928D7A098C3A9A
6 changed files with 55 additions and 56 deletions

View File

@ -12,6 +12,7 @@ from requests.exceptions import RequestException
import tcllib import tcllib
import tcllib.argparser import tcllib.argparser
from tcllib import ansi from tcllib import ansi
from tcllib import devlist
fc = tcllib.FotaCheck() fc = tcllib.FotaCheck()
@ -34,7 +35,7 @@ fc.cltp = fc.CLTP.DESKTOP
prdcheck = "" if args.tocheck is None else args.tocheck prdcheck = "" if args.tocheck is None else args.tocheck
print("Loading list of devices.") print("Loading list of devices.")
prds = tcllib.FotaCheck.get_devicelist() prds = devlist.get_devicelist()
print("List of latest FULL firmware by PRD:") print("List of latest FULL firmware by PRD:")

View File

@ -12,6 +12,7 @@ from requests.exceptions import RequestException
import tcllib import tcllib
import tcllib.argparser import tcllib.argparser
from tcllib import ansi from tcllib import ansi
from tcllib import devlist
fc = tcllib.FotaCheck() fc = tcllib.FotaCheck()
@ -37,7 +38,7 @@ else:
prdcheck = "" if args.tocheck is None else args.tocheck prdcheck = "" if args.tocheck is None else args.tocheck
print("Loading list of devices.") print("Loading list of devices.")
prds = tcllib.FotaCheck.get_devicelist() prds = devlist.get_devicelist()
print("List of latest OTA firmware{} by PRD:".format(force_ver_text)) print("List of latest OTA firmware{} by PRD:".format(force_ver_text))

View File

@ -13,6 +13,7 @@ from requests.exceptions import RequestException, Timeout
import tcllib import tcllib
import tcllib.argparser import tcllib.argparser
from tcllib import ansi from tcllib import ansi
from tcllib import devlist
fc = tcllib.FotaCheck() fc = tcllib.FotaCheck()
@ -41,7 +42,7 @@ if ceiling < floor:
raise SystemExit raise SystemExit
print("Loading list of devices...", end="", flush=True) print("Loading list of devices...", end="", flush=True)
prd_db = tcllib.FotaCheck.get_devicelist() prd_db = devlist.get_devicelist()
print(" OK") print(" OK")
print("Valid PRDs not already in database:") print("Valid PRDs not already in database:")

View File

@ -12,6 +12,7 @@ from requests.exceptions import RequestException, Timeout
import tcllib import tcllib
import tcllib.argparser import tcllib.argparser
from tcllib import ansi from tcllib import ansi
from tcllib import devlist
# Variants to scan for # Variants to scan for
@ -42,7 +43,7 @@ if ceiling < floor:
raise SystemExit raise SystemExit
print("Loading list of known devices...", end="", flush=True) print("Loading list of known devices...", end="", flush=True)
prd_db = tcllib.FotaCheck.get_devicelist() prd_db = devlist.get_devicelist()
print(" OK") print(" OK")
print("Valid PRDs not already in database:") print("Valid PRDs not already in database:")

View File

@ -25,7 +25,6 @@ class FotaCheck(
tclencheader.TclEncHeaderMixin, tclencheader.TclEncHeaderMixin,
servervote.ServerVoteMixin, servervote.ServerVoteMixin,
credentials.CredentialsMixin, credentials.CredentialsMixin,
devlist.DevListMixin,
dumpmgr.DumpMgrMixin dumpmgr.DumpMgrMixin
): ):
"""Main API handler class.""" """Main API handler class."""

View File

@ -19,59 +19,55 @@ DEVICELIST_FILE = "prds.json"
DEVICELIST_CACHE_SECONDS = 86400 DEVICELIST_CACHE_SECONDS = 86400
class DevListMixin: def get_devicelist(force=False, output_diff=True):
"""A mixin component for device list management.""" """Return device list from saved database."""
@staticmethod need_download = True
def get_devicelist(force=False, output_diff=True):
"""Return device list from saved database."""
need_download = True
old_prds = None
try:
filestat = os.stat(DEVICELIST_FILE)
filemtime = filestat.st_mtime
if filemtime > time.time() - DEVICELIST_CACHE_SECONDS:
need_download = False
with open(DEVICELIST_FILE, "rt") as dlfile:
old_prds = json.load(dlfile)
except FileNotFoundError:
pass
if need_download or force:
prds_json = requests.get(DEVICELIST_URL).text
with open(DEVICELIST_FILE, "wt") as dlfile:
dlfile.write(prds_json)
old_prds = None
try:
filestat = os.stat(DEVICELIST_FILE)
filemtime = filestat.st_mtime
if filemtime > time.time() - DEVICELIST_CACHE_SECONDS:
need_download = False
with open(DEVICELIST_FILE, "rt") as dlfile: with open(DEVICELIST_FILE, "rt") as dlfile:
prds = json.load(dlfile) old_prds = json.load(dlfile)
except FileNotFoundError:
pass
if old_prds and output_diff: if need_download or force:
DevListMixin.print_prd_diff(old_prds, prds) prds_json = requests.get(DEVICELIST_URL).text
with open(DEVICELIST_FILE, "wt") as dlfile:
dlfile.write(prds_json)
return prds with open(DEVICELIST_FILE, "rt") as dlfile:
prds = json.load(dlfile)
@staticmethod if old_prds and output_diff:
def print_prd_diff(old_prds, new_prds): print_prd_diff(old_prds, prds)
"""Print changes between old and new databases."""
added_prds = [prd for prd in new_prds if prd not in old_prds] return prds
removed_prds = [prd for prd in old_prds if prd not in new_prds]
for prd in removed_prds: def print_prd_diff(old_prds, new_prds):
print("> Removed device {} (was at {} / OTA: {}).".format(ansi.RED + prd + ansi.RESET, old_prds[prd]["last_full"], old_prds[prd]["last_ota"])) """Print changes between old and new databases."""
for prd in added_prds: added_prds = [prd for prd in new_prds if prd not in old_prds]
print("> New device {} ({} / OTA: {}).".format(ansi.GREEN + prd + ansi.RESET, new_prds[prd]["last_full"], new_prds[prd]["last_ota"])) removed_prds = [prd for prd in old_prds if prd not in new_prds]
for prd, pdata in new_prds.items(): for prd in removed_prds:
if prd in added_prds: print("> Removed device {} (was at {} / OTA: {}).".format(ansi.RED + prd + ansi.RESET, old_prds[prd]["last_full"], old_prds[prd]["last_ota"]))
continue for prd in added_prds:
odata = old_prds[prd] print("> New device {} ({} / OTA: {}).".format(ansi.GREEN + prd + ansi.RESET, new_prds[prd]["last_full"], new_prds[prd]["last_ota"]))
if pdata["last_full"] != odata["last_full"] and pdata["last_ota"] != odata["last_ota"]: for prd, pdata in new_prds.items():
print("> {}: {}{} (OTA: {}{})".format( if prd in added_prds:
prd, continue
ansi.CYAN_DARK + str(odata["last_full"]) + ansi.RESET, odata = old_prds[prd]
ansi.CYAN + str(pdata["last_full"]) + ansi.RESET, if pdata["last_full"] != odata["last_full"] and pdata["last_ota"] != odata["last_ota"]:
ansi.YELLOW_DARK + str(odata["last_ota"]) + ansi.RESET, print("> {}: {}{} (OTA: {}{})".format(
ansi.YELLOW + str(pdata["last_ota"]) + ansi.RESET prd,
)) ansi.CYAN_DARK + str(odata["last_full"]) + ansi.RESET,
elif pdata["last_full"] != odata["last_full"]: ansi.CYAN + str(pdata["last_full"]) + ansi.RESET,
print("> {}: {}{} (FULL)".format(prd, ansi.CYAN_DARK + str(odata["last_full"]) + ansi.RESET, ansi.CYAN + str(pdata["last_full"]) + ansi.RESET)) ansi.YELLOW_DARK + str(odata["last_ota"]) + ansi.RESET,
elif pdata["last_ota"] != odata["last_ota"]: ansi.YELLOW + str(pdata["last_ota"]) + ansi.RESET
print("> {}: {}{} (OTA)".format(prd, ansi.YELLOW_DARK + str(odata["last_ota"]) + ansi.RESET, ansi.YELLOW + str(pdata["last_ota"]) + ansi.RESET)) ))
elif pdata["last_full"] != odata["last_full"]:
print("> {}: {}{} (FULL)".format(prd, ansi.CYAN_DARK + str(odata["last_full"]) + ansi.RESET, ansi.CYAN + str(pdata["last_full"]) + ansi.RESET))
elif pdata["last_ota"] != odata["last_ota"]:
print("> {}: {}{} (OTA)".format(prd, ansi.YELLOW_DARK + str(odata["last_ota"]) + ansi.RESET, ansi.YELLOW + str(pdata["last_ota"]) + ansi.RESET))