GlasgowEmbedded · Jul 28, 2020 · Jul 28, 2020
Showing with 326 additions and 250 deletions.

+19 −15 software/glasgow/applet/__init__.py

+292 −1 software/glasgow/applet/program/avr/__init__.py

+15 −234 software/glasgow/applet/program/avr/spi.py
diff --git a/software/glasgow/applet/__init__.py b/software/glasgow/applet/__init__.py
@@ -1,5 +1,6 @@
 import re
 import argparse
+from abc import ABCMeta, abstractmethod
 
 from ..gateware.clockgen import *
 
@@ -11,26 +12,27 @@ class GlasgowAppletError(Exception):
     """An exception raised when an applet encounters an error."""
 
 
-class _GlasgowAppletMeta(type):
-    def __new__(metacls, clsname, bases, namespace, **kwargs):
+class GlasgowAppletMeta(ABCMeta):
+    all_applets = {}
+
+    def __new__(metacls, clsname, bases, namespace, name=None, **kwargs):
+        if name is not None:
+            if name in metacls.all_applets:
+                raise NameError(f"Applet {name:r} already exists")
+            namespace["name"] = name
+
         # Any class that overrides interact() no longer has its superclass' custom REPL, so be
         # helpful and reset that attribute.
         if "has_custom_repl" not in namespace and "interact" in namespace:
             namespace["has_custom_repl"] = False
 
-        return type.__new__(metacls, clsname, bases, namespace, **kwargs)
+        cls = ABCMeta.__new__(metacls, clsname, bases, namespace, **kwargs)
+        if name is not None:
+            metacls.all_applets[name] = cls
+        return cls
 
 
-class GlasgowApplet(metaclass=_GlasgowAppletMeta):
-    all_applets = {}
-
-    def __init_subclass__(cls, name):
-        if name in cls.all_applets:
-            raise ValueError("Applet {!r} already exists".format(name))
-
-        cls.all_applets[name] = cls
-        cls.name = name
-
+class GlasgowApplet(metaclass=GlasgowAppletMeta):
     preview = False
     help = "applet help missing"
     description = "applet description missing"
@@ -50,8 +52,9 @@ def derive_clock(self, *args, clock_name=None, **kwargs):
             else:
                 raise GlasgowAppletError("clock {}: {}".format(clock_name, e))
 
+    @abstractmethod
     def build(self, target):
-        raise NotImplementedError
+        pass
 
     @classmethod
     def add_run_arguments(cls, parser, access):
@@ -60,8 +63,9 @@ def add_run_arguments(cls, parser, access):
     async def run_lower(self, cls, device, args, **kwargs):
         return await super(cls, self).run(device, args, **kwargs)
 
+    @abstractmethod
     async def run(self, device, args):
-        raise NotImplementedError
+        pass
 
     @classmethod
     def add_interact_arguments(cls, parser):

diff --git a/software/glasgow/applet/program/avr/__init__.py b/software/glasgow/applet/program/avr/__init__.py
@@ -31,8 +31,299 @@
     Described in e.g. AT32UC3L064 datasheet.
 """
 
+import logging
+import argparse
+from abc import ABCMeta, abstractmethod
+from fx2.format import autodetect, input_data, output_data
+
 from ... import *
+from ....database.microchip.avr import *
+
 
+__all__ = ["ProgramAVRError", "ProgramAVRInterface", "ProgramAVRApplet"]
 
-class AVRError(GlasgowAppletError):
+
+class ProgramAVRError(GlasgowAppletError):
     pass
+
+
+class ProgramAVRInterface(metaclass=ABCMeta):
+    @abstractmethod
+    async def programming_enable(self):
+        pass
+
+    @abstractmethod
+    async def programming_disable(self):
+        pass
+
+    @abstractmethod
+    async def read_signature(self):
+        pass
+
+    @abstractmethod
+    async def read_fuse(self, address):
+        pass
+
+    @abstractmethod
+    async def read_fuse_range(self, addresses):
+        return bytearray([await self.read_fuse(address) for address in addresses])
+
+    @abstractmethod
+    async def write_fuse(self, address, data):
+        pass
+
+    @abstractmethod
+    async def read_lock_bits(self):
+        pass
+
+    @abstractmethod
+    async def write_lock_bits(self, data):
+        pass
+
+    @abstractmethod
+    async def read_calibration(self, address):
+        pass
+
+    async def read_calibration_range(self, addresses):
+        return bytearray([await self.read_calibration(address) for address in addresses])
+
+    @abstractmethod
+    async def read_program_memory(self, address):
+        pass
+
+    async def read_program_memory_range(self, addresses):
+        return bytearray([await self.read_program_memory(address) for address in addresses])
+
+    @abstractmethod
+    async def load_program_memory_page(self, address, data):
+        pass
+
+    @abstractmethod
+    async def write_program_memory_page(self, address):
+        pass
+
+    async def write_program_memory_range(self, address, chunk, page_size):
+        dirty_page = False
+        page_mask  = page_size - 1
+
+        for offset, byte in enumerate(chunk):
+            byte_address = address + offset
+            if dirty_page and byte_address % page_size == 0:
+                await self.write_program_memory_page((byte_address - 1) & ~page_mask)
+
+            await self.load_program_memory_page(byte_address & page_mask, byte)
+            dirty_page = True
+
+        if dirty_page:
+            await self.write_program_memory_page(byte_address & ~page_mask)
+
+    @abstractmethod
+    async def read_eeprom(self, address):
+        pass
+
+    async def read_eeprom_range(self, addresses):
+        return bytearray([await self.read_eeprom(address) for address in addresses])
+
+    @abstractmethod
+    async def load_eeprom_page(self, address, data):
+        pass
+
+    @abstractmethod
+    async def write_eeprom_page(self, address):
+        pass
+
+    async def write_eeprom_range(self, address, chunk, page_size):
+        dirty_page = False
+        page_mask  = page_size - 1
+
+        for offset, byte in enumerate(chunk):
+            byte_address = address + offset
+            if dirty_page and byte_address % page_size == 0:
+                await self.write_eeprom_page((byte_address - 1) & ~page_mask)
+
+            await self.load_eeprom_page(byte_address & page_mask, byte)
+            dirty_page = True
+
+        if dirty_page:
+            await self.write_eeprom_page(byte_address & ~page_mask)
+
+    @abstractmethod
+    async def chip_erase(self):
+        pass
+
+
+class ProgramAVRApplet(GlasgowApplet):
+    logger = logging.getLogger(__name__)
+    help = "program Microchip (Atmel) AVR microcontrollers"
+
+    @classmethod
+    def add_interact_arguments(cls, parser):
+        def bits(arg): return int(arg, 2)
+
+        p_operation = parser.add_subparsers(dest="operation", metavar="OPERATION")
+
+        p_identify = p_operation.add_parser(
+            "identify", help="identify connected device")
+
+        p_read = p_operation.add_parser(
+            "read", help="read device memories")
+        p_read.add_argument(
+            "-f", "--fuses", default=False, action="store_true",
+            help="display fuse bytes")
+        p_read.add_argument(
+            "-l", "--lock-bits", default=False, action="store_true",
+            help="display lock bits")
+        p_read.add_argument(
+            "-c", "--calibration", default=False, action="store_true",
+            help="display calibration bytes")
+        p_read.add_argument(
+            "-p", "--program", metavar="FILE", type=argparse.FileType("wb"),
+            help="write program memory contents to FILE")
+        p_read.add_argument(
+            "-e", "--eeprom", metavar="FILE", type=argparse.FileType("wb"),
+            help="write EEPROM contents to FILE")
+
+        p_write_fuses = p_operation.add_parser(
+            "write-fuses", help="write and verify device fuses")
+        p_write_fuses.add_argument(
+            "-L", "--low", metavar="BITS", type=bits,
+            help="set low fuse to binary BITS")
+        p_write_fuses.add_argument(
+            "-H", "--high", metavar="BITS", type=bits,
+            help="set high fuse to binary BITS")
+        p_write_fuses.add_argument(
+            "-E", "--extra", metavar="BITS", type=bits,
+            help="set extra fuse to binary BITS")
+
+        p_write_lock = p_operation.add_parser(
+            "write-lock", help="write and verify device lock bits")
+        p_write_lock.add_argument(
+            "bits", metavar="BITS", type=bits,
+            help="write lock bits BITS")
+
+        p_write_program = p_operation.add_parser(
+            "write-program", help="write and verify device program memory")
+        p_write_program.add_argument(
+            "file", metavar="FILE", type=argparse.FileType("rb"),
+            help="read program memory contents from FILE")
+
+        p_write_eeprom = p_operation.add_parser(
+            "write-eeprom", help="write and verify device EEPROM")
+        p_write_eeprom.add_argument(
+            "file", metavar="FILE", type=argparse.FileType("rb"),
+            help="read EEPROM contents from FILE")
+
+    @staticmethod
+    def _check_format(file, kind):
+        try:
+            autodetect(file)
+        except ValueError:
+            raise ProgramAVRError("cannot determine %s file format" % kind)
+
+    async def interact(self, device, args, avr_iface):
+        await avr_iface.programming_enable()
+
+        signature = await avr_iface.read_signature()
+        device = devices_by_signature[signature]
+        self.logger.info("device signature: %s (%s)",
+            "{:02x} {:02x} {:02x}".format(*signature),
+            "unknown" if device is None else device.name)
+
+        if args.operation not in (None, "identify") and device is None:
+            raise ProgramAVRError("cannot operate on unknown device")
+
+        if args.operation == "read":
+            if args.fuses:
+                fuses = await avr_iface.read_fuse_range(range(device.fuses_size))
+                if device.fuses_size > 2:
+                    self.logger.info("fuses: low %s high %s extra %s",
+                                     "{:08b}".format(fuses[0]),
+                                     "{:08b}".format(fuses[1]),
+                                     "{:08b}".format(fuses[2]))
+                elif device.fuses_size > 1:
+                    self.logger.info("fuses: low %s high %s",
+                                     "{:08b}".format(fuses[0]),
+                                     "{:08b}".format(fuses[1]))
+                else:
+                    self.logger.info("fuse: %s", "{:08b}".format(fuses[0]))
+
+            if args.lock_bits:
+                lock_bits = await avr_iface.read_lock_bits()
+                self.logger.info("lock bits: %s", "{:08b}".format(lock_bits))
+
+            if args.calibration:
+                calibration = \
+                    await avr_iface.read_calibration_range(range(device.calibration_size))
+                self.logger.info("calibration bytes: %s",
+                                 " ".join(["%02x" % b for b in calibration]))
+
+            if args.program:
+                self._check_format(args.program, "program memory")
+                self.logger.info("reading program memory (%d bytes)", device.program_size)
+                output_data(args.program,
+                    await avr_iface.read_program_memory_range(range(device.program_size)))
+
+            if args.eeprom:
+                self._check_format(args.eeprom, "EEPROM")
+                self.logger.info("reading EEPROM (%d bytes)", device.eeprom_size)
+                output_data(args.eeprom,
+                    await avr_iface.read_eeprom_range(range(device.eeprom_size)))
+
+        if args.operation == "write-fuses":
+            if args.high and device.fuses_size < 2:
+                raise ProgramAVRError("device does not have high fuse")
+
+            if args.low:
+                self.logger.info("writing low fuse")
+                await avr_iface.write_fuse(0, args.low)
+                written = await avr_iface.read_fuse(0)
+                if written != args.low:
+                    raise ProgramAVRError("verification of low fuse failed: %s" %
+                                          "{:08b} != {:08b}".format(written, args.low))
+
+            if args.high:
+                self.logger.info("writing high fuse")
+                await avr_iface.write_fuse(1, args.high)
+                written = await avr_iface.read_fuse(1)
+                if written != args.high:
+                    raise ProgramAVRError("verification of high fuse failed: %s" %
+                                          "{:08b} != {:08b}".format(written, args.high))
+
+        if args.operation == "write-lock":
+            self.logger.info("writing lock bits")
+            await avr_iface.write_lock_bits(args.bits)
+            written = await avr_iface.read_lock_bits()
+            if written != args.bits:
+                raise ProgramAVRError("verification of lock bits failed: %s" %
+                                      "{:08b} != {:08b}".format(written, args.bits))
+
+        if args.operation == "write-program":
+            self.logger.info("erasing chip")
+            await avr_iface.chip_erase()
+
+            self._check_format(args.file, "program memory")
+            data = input_data(args.file)
+            self.logger.info("writing program memory (%d bytes)",
+                             sum([len(chunk) for address, chunk in data]))
+            for address, chunk in data:
+                chunk = bytes(chunk)
+                await avr_iface.write_program_memory_range(address, chunk, device.program_page)
+                written = await avr_iface.read_program_memory_range(range(address, len(chunk)))
+                if written != chunk:
+                    raise ProgramAVRError("verification failed at address %#06x: %s != %s" %
+                                          (address, written.hex(), chunk.hex()))
+
+        if args.operation == "write-eeprom":
+            self._check_format(args.file, "EEPROM")
+            data = input_data(args.file)
+            self.logger.info("writing EEPROM (%d bytes)",
+                             sum([len(chunk) for address, chunk in data]))
+            for address, chunk in data:
+                chunk = bytes(chunk)
+                await avr_iface.write_eeprom_range(address, chunk, device.eeprom_page)
+                written = await avr_iface.read_eeprom_range(range(address, len(chunk)))
+                if written != chunk:
+                    raise ProgramAVRError("verification failed at address %#06x: %s != %s" %
+                                          (address, written.hex(), chunk.hex()))
+
+        await avr_iface.programming_disable()