Добавлена базовая обработка ошибок связи с OpenOCD

flash_drivers/generic_flash.py

@@ -44,6 +44,14 @@ QUAD_PAGE_PROGRAM_COMMAND = 0x32
 JEDEC_ID_COMMAND = 0x9F
 
 
+class FlashError(Exception):
+    def __init__(self, value):
+        self.value = value
+
+    def __str__(self):
+        return ("ERROR: " + repr(self.value))
+
+
 class SREG_Num(Enum):
     SREG1 = 0x00
     SREG2 = 0x30
@@ -146,7 +154,7 @@ def page_program(
 ):
     print(f"Writing Flash page {ByteAddress:#010x}... {progress}", flush=True)
     if byte_count > 256:
-        raise Exception("Byte count more than 256")
+        raise FlashError("Byte count more than 256")
 
     write_enable(openocd)
     spifi.send_command(openocd, PAGE_PROGRAM_COMMAND, spifi.Frameform.OPCODE_3ADDR,
@@ -182,7 +190,7 @@ def quad_page_program(
 ):
     print(f"Writing page {ByteAddress:#010x}... {progress}", flush=True)
     if byte_count > 256:
-        raise Exception("Byte count more than 256")
+        raise FlashError("Byte count more than 256")
 
     write_enable(openocd)
     spifi.send_command(openocd, QUAD_PAGE_PROGRAM_COMMAND, spifi.Frameform.OPCODE_3ADDR,

hex_parser.py

@@ -36,11 +36,11 @@ class Segment:
                 self.memory = section
 
         if self.memory is None:
-            raise Exception(
+            raise ParserError(
-                f"ERROR: segment with offset {self.offset:#0x} doesn't belong to any section")
+                f"segment with offset {self.offset:#0x} doesn't belong to any section")
 
         if (self.offset + self.data.__len__()) > (self.memory.offset + self.memory.length):
-            raise Exception(
+            raise ParserError(
                 f"ERROR: segment with offset {self.offset:#0x} "
                 f"and length {self.data.__len__()} "
                 f"overflows section {self.memory.type.name}"
@@ -75,7 +75,7 @@ class FirmwareFile:
                 bin_content = list(f.read())
                 self.segments.append(Segment(offset=0, data=bin_content, sections=sections))
         else:
-            raise Exception(f"Unsupported file format: {self.file_extension}")
+            raise ParserError(f"Unsupported file format: {self.file_extension}")
 
     def _parse_hex(self, lines: List[str], sections: List[MemorySection]): 
         segments: List[Segment] = []

mik32_check.py

@@ -7,7 +7,7 @@ import time
 from typing import List, Union
 
 from mik32_debug_hal.power_manager import pm_init
-from mik32_upload import BootMode, Pages, form_pages, openocd_exec_path, openocd_scripts_path, openocd_interface_path, openocd_target_path, adapter_default_speed, run_openocd, default_post_action, default_log_path, default_openocd_host, mik32_sections, OpenOCDStartupException, adapter_speed_not_supported, memory_page_size
+from mik32_upload import BootMode, Pages, form_pages, openocd_exec_path, openocd_scripts_path, openocd_interface_path, openocd_target_path, adapter_default_speed, run_openocd, default_post_action, default_log_path, default_openocd_host, mik32_sections, OpenOCDError, adapter_speed_not_supported, memory_page_size
 from mik32_debug_hal.gpio import MIK32_Version, gpio_init, gpio_deinit
 import mik32_debug_hal.eeprom as eeprom
 import mik32_debug_hal.spifi as spifi
@@ -63,7 +63,7 @@ def upload_file(
             logging.debug("OpenOCD started!")
 
         except OSError as e:
-            raise OpenOCDStartupException(e)
+            raise OpenOCDError(e)
     try:
         with OpenOcdTclRpc(host, port) as openocd:
             if (all(openocd_interface.find(i) == -1 for i in adapter_speed_not_supported)):

mik32_debug_hal/dma.py

@@ -7,6 +7,15 @@ from dataclasses import dataclass
 import mik32_debug_hal.registers.memory_map as mem_map
 import mik32_debug_hal.registers.bitfields.dma as dma_fields
 
+
+class DmaError(Exception):
+    def __init__(self, value):
+        self.value = value
+
+    def __str__(self):
+        return ("ERROR: " + repr(self.value))
+
+
 # ReadStatus. Разрешить читать текущий статус канала
 class CurrentValue(Enum):
     ENABLE = 0  # Текущие значения
@@ -201,4 +210,4 @@ class DMA:
             if self.get_control() & mask != 0:
                 return
 
-        raise Exception
+        raise DmaError

mik32_debug_hal/eeprom.py

@@ -117,8 +117,8 @@ def eeprom_check_data_ahb_lite(openocd: OpenOcdTclRpc, words: List[int], offset:
         print("EEPROM check through AHB-Lite...", flush=True)
     mem_array = openocd.read_memory(0x01000000 + offset, 32, len(words))
     if len(words) != len(mem_array):
-        raise Exception(
+        print("ERROR: Wrong number of words in read_memory output!")
-            "Wrong number of words in read_memory output!")
+        return 1
     progress = 0
     if print_progress:
         print("[", end="", flush=True)

mik32_debug_hal/spifi.py

@@ -13,6 +13,14 @@ import mik32_debug_hal.dma as dma
 import flash_drivers.generic_flash as generic_flash
 
 
+class SpifiError(Exception):
+    def __init__(self, value):
+        self.value = value
+
+    def __str__(self):
+        return ("ERROR: " + repr(self.value))
+
+
 def spifi_intrq_clear(openocd: OpenOcdTclRpc):
     openocd.write_word(mem_map.SPIFI_CONFIG_STAT, openocd.read_word(mem_map.SPIFI_CONFIG_STAT) |
                        spifi_fields.SPIFI_CONFIG_STAT_INTRQ_M)
@@ -70,7 +78,7 @@ def spifi_wait_intrq_timeout(openocd: OpenOcdTclRpc, error_message: str):
     while time.perf_counter() < time_end:
         if (openocd.read_word(mem_map.SPIFI_CONFIG_STAT) & spifi_fields.SPIFI_CONFIG_STAT_INTRQ_M) != 0:
             return
-    raise Exception(error_message)
+    raise SpifiError(error_message)
 
 
 class Frameform(Enum):
@@ -167,7 +175,7 @@ def send_command(
 
 def write(openocd: OpenOcdTclRpc, address: int, data: List[int], data_len: int):
     if data_len > 256:
-        raise Exception("Byte count more than 256")
+        raise SpifiError("Byte count more than 256")
 
     generic_flash.page_program(openocd, address, data, data_len)
 

mik32_upload.py

@@ -7,7 +7,7 @@ import time
 from enum import Enum
 from typing import List, Dict, NamedTuple, Union
 from hex_parser import FirmwareFile, MemorySection, MemoryType, Segment
-from tclrpc import OpenOcdTclRpc, TclException
+from tclrpc import OpenOcdTclRpc, TclException, TclPortError
 from mik32_debug_hal.gpio import MIK32_Version, gpio_init, gpio_deinit
 import mik32_debug_hal.eeprom as eeprom
 import mik32_debug_hal.spifi as spifi
@@ -53,19 +53,6 @@ if os.name == 'nt':
 adapter_default_speed = 500
 
 
-def test_connection():
-    output = ""
-    with OpenOcdTclRpc() as openocd:
-        try:
-            output = openocd.run("capture \"reg\"")
-        except OSError:
-            logging.debug("Test connection timed out, try again")
-            output = openocd.run("capture \"reg\"")
-
-    if output == "":
-        raise Exception("ERROR: no regs found, check MCU connection")
-
-
 memory_page_size = {
     MemoryType.EEPROM: 128,
     MemoryType.SPIFI: 256
@@ -126,12 +113,12 @@ def segments_to_pages(segments: List[Segment], page_size: int) -> Dict[int, List
     return pages
 
 
-class OpenOCDStartupException(Exception):
+class OpenOCDError(Exception):
     def __init__(self, msg):
         self.msg = msg
 
     def __repr__(self):
-        return f"OpenOCD Startup Exception: {self.msg}"
+        return f"ERROR: OpenOCD Startup Exception: {self.msg}"
 
 
 def run_openocd(
@@ -215,9 +202,9 @@ def upload_file(
         use_driver=True,
 ) -> int:
     """
-    Write ihex or binary file into MIK32 EEPROM or external flash memory
+    Запись прошивки в формате Intel HEX или бинарном в память MIK32.
-    @filename: full path to the file with hex or bin file format
+    @filename: полный путь до файла прошивки
-    @return: return 0 if successful, 1 if failed
+    @return: возвращает 0 в случае успеха, 1 - если прошивка неудачна
     """
 
     print(f"Using {mik_version.value}")
@@ -226,9 +213,13 @@ def upload_file(
 
     if not os.path.exists(filename):
         print(f"ERROR: File {filename} does not exist")
-        exit(1)
+        return 1
 
-    file = FirmwareFile(filename, mik32_sections)
+    try:
+        file = FirmwareFile(filename, mik32_sections)
+    except ParserError as e:
+        print(e)
+        return 1
 
     segments: List[Segment] = file.get_segments()
     pages: Pages = form_pages(segments, boot_mode)
@@ -244,10 +235,17 @@ def upload_file(
             logging.debug("OpenOCD started!")
 
         except OSError as e:
-            raise OpenOCDStartupException(e)
+            raise OpenOCDError(e)
     try:
         time.sleep(0.1)
         with OpenOcdTclRpc(host, port) as openocd:
+            try:
+                openocd.run("capture \"riscv.cpu curstate\"")
+            except OSError as e:
+                print("ERROR: Tcl port connection failed")
+                print("Check connectivity and OpenOCD log")
+                return 1
+            
             if (all(openocd_interface.find(i) == -1 for i in adapter_speed_not_supported)):
                 openocd.run(f"adapter speed {adapter_speed}")
             openocd.run(f"log_output \"{log_path}\"")
@@ -328,8 +326,13 @@ def upload_file(
             openocd.run(post_action)
     except ConnectionRefusedError:
         print("ERROR: The connection to OpenOCD is not established. Check the settings and connection of the debugger")
-    except TclException as e:
+    except (OpenOCDError, TclPortError, TclException) as e:
-        print(f"ERROR: TclException {e.code} \n {e.msg}")
+        print(e)
+        exit(1)
+    except ConnectionResetError as e:
+        print("ERROR: Tcl connection reset")
+        print("Check OpenOCD log")
+        print(e.strerror)
     finally:
         if proc is not None:
             proc.kill()

parsers.py

@@ -5,6 +5,14 @@ from enum import Enum
 from typing import List
 
 
+class ParserError(Exception):
+    def __init__(self, value):
+        self.value = value
+
+    def __str__(self):
+        return ("ERROR: " + repr(self.value))
+
+
 class RecordType(Enum):
     UNKNOWN = -1
     DATA = 0
@@ -24,14 +32,14 @@ class Record:
 
 def parse_line(line: str, line_n: int, file_extension: str) -> Record:
     if file_extension != ".hex":
-        raise Exception("Unsupported file format: %s" % (file_extension))
+        raise ParserError("Unsupported file format: %s" % (file_extension))
 
     return parse_hex_line(line, line_n)
 
 
 def parse_hex_line(line: str, line_n: int) -> Record:
     if line[0] != ':':
-        raise Exception("Error: unexpected record mark in line %d: %s, expect \':\', get \'%c\'" % (
+        raise ParserError("Error: unexpected record mark in line %d: %s, expect \':\', get \'%c\'" % (
             line_n, line, line[0]))
 
     datalen = int(line[1:3], base=16)               # Data field length
@@ -45,9 +53,10 @@ def parse_hex_line(line: str, line_n: int) -> Record:
         splitted_by_bytes.append(data_bytes_line[i*2:i*2+2])
 
     data_bytes = list(map(lambda x: int(x, base=16), splitted_by_bytes))
-    checksum = (datalen + int(line[3:5], base=16) + int(line[5:7], base=16) + rectype + sum(data_bytes)) % 256
+    checksum = (datalen + int(line[3:5], base=16) +
+                int(line[5:7], base=16) + rectype + sum(data_bytes)) % 256
     if (checksum + crc) % 256 != 0:
-        raise Exception("Checksum mismatch in line %d %s" % (line_n, line))
+        raise ParserError("Checksum mismatch in line %d %s" % (line_n, line))
 
     record = Record(RecordType.UNKNOWN, 0, [])
 
@@ -67,7 +76,8 @@ def parse_hex_line(line: str, line_n: int) -> Record:
         # record.data = list(map(lambda x: int(x, base=16), splitted_by_bytes))
     elif rectype == 4:  # Extended Linear Address Record
         record.type = RecordType.EXTADDR
-        record.address = data_bytes[1] * pow(256, 2) + data_bytes[0] * pow(256, 3)
+        record.address = data_bytes[1] * \
+            pow(256, 2) + data_bytes[0] * pow(256, 3)
     elif rectype == 5:  # Start Linear Address Record
         record.type = RecordType.LINEARSTARTADDR
         address = 0

tclrpc.py

@@ -11,7 +11,14 @@ class TclException(Exception):
         self.msg = msg
 
     def __repr__(self):
-        return 'TclException(%d, %r)' % (self.code, self.msg)
+        return '\nTclException(%d, %r)' % (self.code, self.msg)
+
+class TclPortError(Exception):
+    def __init__(self, msg):
+        self.msg = msg
+
+    def __repr__(self):
+        return 'TclPortError %r' % (self.msg)
 
 _RE_SIMPLE_TCL_WORD = re.compile(r"^[a-zA-Z_0-9:+./@=,'-]+$")
 def tcl_quote_word(word):
@@ -85,7 +92,7 @@ class OpenOcdTclRpc:
             index = data.find(self.SEPARATOR_BYTES)
             if index >= 0:
                 if index != len(data) - 1:
-                    raise Exception('Unhandled extra bytes after %r'.format(self.SEPARATOR_BYTES))
+                    raise TclPortError('Unhandled extra bytes after %r'.format(self.SEPARATOR_BYTES))
                 return data[:-1]
         
     def wait_for_port(self, timeout: float = 5.0):