Merge pull request #558 from ColdWindScholar/master

More Beautiful and bug fixes

Example/Library/tcpclient.py

@@ -1,19 +1,20 @@
 import socket
 from binascii import hexlify
 
-class tcpclient():
+
+class tcpclient:
     def __init__(self, port):
         self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
         server_address = ("localhost", port)
         print("connecting to %s port %s" % server_address)
         self.sock.connect(server_address)
 
-    def sendcommands(self,commands):
+    def sendcommands(self, commands):
         try:
             for command in commands:
                 self.sock.sendall(bytes(command, 'utf-8'))
-                data=""
-                while not "<ACK>" in data and not "<NAK>" in data:
+                data = ""
+                while "<ACK>" not in data and "<NAK>" not in data:
                     tmp = self.sock.recv(4096)
                     if tmp == b"":
                         continue
@@ -25,5 +26,3 @@ def sendcommands(self,commands):
         finally:
             print("closing socket")
             self.sock.close()
-
-

Example/tcpclient.py

@@ -1,25 +1,27 @@
 #!/usr/bin/env python3
 from edl.Library.tcpclient import tcpclient
 
-class client():
+
+class client:
     def __init__(self):
-        self.commands=[]
+        self.commands = []
 
     def send(self):
         self.tcp = tcpclient(1340)
         self.tcp.sendcommands(self.commands)
 
-    def read(self,src):
+    def read(self, src):
         self.commands.append(f"peekqword:{hex(src)}")
 
-    def write(self,dest,value):
+    def write(self, dest, value):
         self.commands.append(f"pokeqword:{hex(dest)},{hex(value)}")
 
-    def memcpy(self,dest,src,size):
+    def memcpy(self, dest, src, size):
         self.commands.append(f"memcpy:{hex(dest)},{hex(src)},{hex(size)}")
 
+
 def main():
-    exp=client()
+    exp = client()
     exp.commands = [
         "send:nop",
         "r:boot,boot.img",
@@ -29,5 +31,6 @@ def main():
     ]
     exp.send()
 
-if __name__=="__main__":
+
+if __name__ == "__main__":
     main()

edl

@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
-# (c) B.Kerler 2018-2023 under GPLv3 license
+# (c) B.Kerler 2018-2024 under GPLv3 license
 # If you use my code, make sure you refer to my name
 #
 # !!!!! If you use this code in commercial products, your product is automatically
@@ -127,34 +127,36 @@ Options:
     --resetmode=mode                   Resetmode for reset (poweroff, reset, edl, etc.)
 """
 
+import logging
 import os
+import re
+import subprocess
 import sys
 import time
-import logging
-import subprocess
-import re
+
 from docopt import docopt
+
 from edlclient.Config.usb_ids import default_ids
-from edlclient.Library.utils import LogBase
-from edlclient.Library.Connection.usblib import usb_class
 from edlclient.Library.Connection.seriallib import serial_class
-from edlclient.Library.sahara import sahara
-from edlclient.Library.streaming_client import streaming_client
+from edlclient.Library.Connection.usblib import usb_class
 from edlclient.Library.firehose_client import firehose_client
-from edlclient.Library.streaming import Streaming
+from edlclient.Library.sahara import sahara
 from edlclient.Library.sahara_defs import cmd_t, sahara_mode_t
+from edlclient.Library.streaming import Streaming
+from edlclient.Library.streaming_client import streaming_client
+from edlclient.Library.utils import LogBase
 from edlclient.Library.utils import is_windows
-from binascii import hexlify
 
 args = docopt(__doc__, version='3')
 
-print("Qualcomm Sahara / Firehose Client V3.62 (c) B.Kerler 2018-2023.")
+print("Qualcomm Sahara / Firehose Client V3.62 (c) B.Kerler 2018-2024.")
 
 
 def parse_cmd(rargs):
     cmds = ["server", "printgpt", "gpt", "r", "rl", "rf", "rs", "w", "wl", "wf", "ws", "e", "es", "ep", "footer",
             "peek", "peekhex", "peekdword", "peekqword", "memtbl", "poke", "pokehex", "pokedword", "pokeqword",
-            "memcpy", "secureboot", "pbl", "qfp", "getstorageinfo", "setbootablestoragedrive", "getactiveslot", "setactiveslot",
+            "memcpy", "secureboot", "pbl", "qfp", "getstorageinfo", "setbootablestoragedrive", "getactiveslot",
+            "setactiveslot",
             "send", "xml", "rawxml", "reset", "nop", "modules", "memorydump", "provision", "qfil"]
     for cmd in cmds:
         if rargs[cmd]:
@@ -239,7 +241,6 @@ class main(metaclass=LogBase):
             if re.findall(r'QCUSB', str(proper_driver)):
                 self.warning(f'Please first install libusb_win32 driver from Zadig')
 
-        mode = ""
         loop = 0
         vid = int(args["--vid"], 16)
         pid = int(args["--pid"], 16)
@@ -285,14 +286,13 @@ class main(metaclass=LogBase):
             self.sahara.programmer = loader
 
         self.info("Waiting for the device")
-        resp = None
         self.cdc.timeout = 1500
         conninfo = self.doconnect(loop)
         mode = conninfo["mode"]
-        if not "data" in conninfo:
+        if conninfo.get("data"):
             version = 2
         else:
-            version = conninfo["data"].version
+            version = conninfo.get("data").version
         if mode == "sahara":
             cmd = conninfo["cmd"]
             if cmd == cmd_t.SAHARA_HELLO_REQ:
@@ -303,7 +303,7 @@ class main(metaclass=LogBase):
                             time.sleep(0.5)
                             print("Device is in memory dump mode, dumping memory")
                             if args["--partitions"]:
-                                self.sahara.debug_mode(args["--partitions"].split(","),version=version)
+                                self.sahara.debug_mode(args["--partitions"].split(","), version=version)
                             else:
                                 self.sahara.debug_mode(version=version)
                             self.exit()

edlclient/Config/qualcomm_config.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
-# (c) B.Kerler 2018-2023 under GPLv3 license
+# (c) B.Kerler 2018-2024 under GPLv3 license
 # If you use my code, make sure you refer to my name
 #
 # !!!!! If you use this code in commercial products, your product is automatically

edlclient/Config/usb_ids.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
-# (c) B.Kerler 2018-2023 under GPLv3 license
+# (c) B.Kerler 2018-2024 under GPLv3 license
 # If you use my code, make sure you refer to my name
 #
 # !!!!! If you use this code in commercial products, your product is automatically

edlclient/Library/Connection/devicehandler.py

@@ -1,20 +1,20 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
-# (c) B.Kerler 2018-2023 under GPLv3 license
+# (c) B.Kerler 2018-2024 under GPLv3 license
 # If you use my code, make sure you refer to my name
 #
 # !!!!! If you use this code in commercial products, your product is automatically
 # GPLv3 and has to be open sourced under GPLv3 as well. !!!!!
-import serial
-import serial.tools.list_ports
 import inspect
 import traceback
 from binascii import hexlify
+
 try:
     from edlclient.Library.utils import *
 except:
     from Library.utils import *
 
+
 class DeviceClass(metaclass=LogBase):
 
     def __init__(self, loglevel=logging.INFO, portconfig=None, devclass=-1):
@@ -119,7 +119,7 @@ def verify_data(self, data, pre="RX:"):
             stack_trace = traceback.format_stack(frame)
             td = []
             for trace in stack_trace:
-                if not "verify_data" in trace and not "Port" in trace:
+                if "verify_data" not in trace and "Port" not in trace:
                     td.append(trace)
             self.debug(td[:-1])
 

edlclient/Library/Connection/seriallib.py

@@ -1,29 +1,29 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
-# (c) B.Kerler 2018-2023 under GPLv3 license
+# (c) B.Kerler 2018-2024 under GPLv3 license
 # If you use my code, make sure you refer to my name
 #
 # !!!!! If you use this code in commercial products, your product is automatically
 # GPLv3 and has to be open sourced under GPLv3 as well. !!!!!
-import os.path
-import time
 import sys
+
 if not sys.platform.startswith('win32'):
     import termios
-    
+
 
 def _reset_input_buffer():
     return
 
+
 def _reset_input_buffer_org(self):
     if not sys.platform.startswith('win32'):
         return termios.tcflush(self.fd, termios.TCIFLUSH)
 
+
 import serial
 import serial.tools.list_ports
 import inspect
-import traceback
-from binascii import hexlify
+
 try:
     from edlclient.Library.utils import *
     from edlclient.Library.Connection.devicehandler import DeviceClass
@@ -38,13 +38,13 @@ def __init__(self, loglevel=logging.INFO, portconfig=None, devclass=-1):
         super().__init__(loglevel, portconfig, devclass)
         self.is_serial = True
 
-    def connect(self, EP_IN=-1, EP_OUT=-1, portname:str=""):
+    def connect(self, EP_IN=-1, EP_OUT=-1, portname: str = ""):
         if self.connected:
             self.close()
             self.connected = False
         if portname == "":
-            devices=self.detectdevices()
-            if len(devices)>0:
+            devices = self.detectdevices()
+            if len(devices) > 0:
                 portname = devices[0]
         if portname != "":
             self.device = serial.Serial(baudrate=115200, bytesize=serial.EIGHTBITS,
@@ -88,13 +88,12 @@ def setbreak(self):
         self.debug("Break set")
 
     def setcontrollinestate(self, RTS=None, DTR=None, isFTDI=False):
-        if RTS==1:
+        if RTS == 1:
             self.device.setRTS(RTS)
-        if DTR==1:
+        if DTR == 1:
             self.device.setDTR(DTR)
         self.debug("Linecoding set")
 
-
     def write(self, command, pktsize=None):
         if pktsize is None:
             pktsize = 512
@@ -169,18 +168,18 @@ def usbread(self, resplen=None, timeout=0):
         epr = self.device.read
         extend = res.extend
         if self.xmlread:
-            info=self.device.read(6)
-            bytestoread=resplen-len(info)
+            info = self.device.read(6)
+            bytestoread = resplen - len(info)
             extend(info)
             if b"<?xml " in info:
-                while not b"response " in res or res[-7:]!=b"</data>":
+                while b"response " not in res or res[-7:] != b"</data>":
                     extend(epr(1))
                 return res
         bytestoread = resplen
         while len(res) < bytestoread:
             try:
-                val=epr(bytestoread)
-                if len(val)==0:
+                val = epr(bytestoread)
+                if len(val) == 0:
                     break
                 extend(val)
             except Exception as e:
@@ -218,5 +217,3 @@ def usbreadwrite(self, data, resplen):
         self.device.flush()
         res = self.usbread(resplen)
         return res
-
-

edlclient/Library/Connection/usblib.py

@@ -1,30 +1,29 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
-# (c) B.Kerler 2018-2023 under GPLv3 license
+# (c) B.Kerler 2018-2024 under GPLv3 license
 # If you use my code, make sure you refer to my name
 #
 # !!!!! If you use this code in commercial products, your product is automatically
 # GPLv3 and has to be open sourced under GPLv3 as well. !!!!!
-import io
+import array
+import inspect
 import logging
+from binascii import hexlify
+from ctypes import c_void_p, c_int
+from enum import Enum
 
+import usb.backend.libusb0
 import usb.core  # pyusb
 import usb.util
-import time
-import inspect
-import array
-import usb.backend.libusb0
-from enum import Enum
-from binascii import hexlify
-from ctypes import c_void_p, c_int
+
 try:
     from edlclient.Library.utils import *
 except:
     from Library.utils import *
 if not is_windows():
     import usb.backend.libusb1
-from struct import pack, calcsize
-import traceback
+from struct import pack
+
 try:
     from edlclient.Library.Connection.devicehandler import DeviceClass
 except:
@@ -211,7 +210,7 @@ def setcontrollinestate(self, RTS=None, DTR=None, isFTDI=False):
     def flush(self):
         return
 
-    def connect(self, EP_IN=-1, EP_OUT=-1, portname:str=""):
+    def connect(self, EP_IN=-1, EP_OUT=-1, portname: str = ""):
         if self.connected:
             self.close()
             self.connected = False
@@ -338,7 +337,7 @@ def close(self, reset=False):
 
     def write(self, command, pktsize=None):
         if pktsize is None:
-            #pktsize = self.EP_OUT.wMaxPacketSize
+            # pktsize = self.EP_OUT.wMaxPacketSize
             pktsize = MAX_USB_BULK_BUFFER_SIZE
         if isinstance(command, str):
             command = bytes(command, 'utf-8')
@@ -389,7 +388,7 @@ def usbread(self, resplen=None, timeout=0):
         extend = res.extend
         while len(res) < resplen:
             try:
-                resplen=epr(buffer,timeout)
+                resplen = epr(buffer, timeout)
                 extend(buffer[:resplen])
                 if resplen == self.EP_IN.wMaxPacketSize:
                     break

edlclient/Library/Connection/usbscsi.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
-# (c) B.Kerler 2018-2023 under GPLv3 license
+# (c) B.Kerler 2018-2024 under GPLv3 license
 # If you use my code, make sure you refer to my name
 #
 # !!!!! If you use this code in commercial products, your product is automatically