alpaca_isp/alpaca_isp/__init__.py

# Copyright 2017 Clayton G. Hobbs
# 
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# 
# http://www.apache.org/licenses/LICENSE-2.0
# 
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from enum import Enum
import time

import serial


class LPC:
    """Interface for LPC in-system programming"""

    def __init__(self, serport, baudrate=115200, clock=12000, timeout=0.1):
        self._serport = serport
        self._baudrate = baudrate
        self._clock = clock
        self._timeout = timeout
        self._echo = True

    def open(self):
        """Open the serial port to communicate with the microcontroller"""
        self._uart = serial.Serial(self._serport, baudrate=self._baudrate,
                timeout=self._timeout)

    def _readline(self):
        """Read a line terminated with b'\r\n'"""
        s = b""
        while True:
            c = self._uart.read(1)
            if not c:
                # If we timed out, give up
                raise RecvTimeout(s)
            s += c
            if s.endswith(b"\r\n"):
                return s

    def _writeline(self, line, plain=False):
        """Write a line to the microcontroller and read the echoed response

        If plain is True, the command is taken to be raw binary data.
        """
        self._uart.write(line)
        self._uart.flush()

        # If echo is disabled, don't try to read back what we sent
        if not self.echo:
            return

        # Read the response, raising the exception if there is one
        if plain:
            response = self._uart.read(len(line))
        else:
            response = self._readline()
        # If we got the wrong response, raise an exception
        if response != line:
            raise ISPError("Wrong text echoed: {}".format(response))

    def _send_command_raw(self, cmd):
        """Send a command to the microcontroller, returning bytes"""
        self._writeline(cmd)
        return self._readline()

    def _send_command(self, cmd):
        """Send a command to the microcontroller, returning the result"""
        rr = self._send_command_raw(cmd)
        lr = [int(n) for n in rr.split()]
        lr[0] = ReturnCode(lr[0])
        if lr[0] != ReturnCode.CMD_SUCCESS:
            raise ISPError(lr)
        return lr

    def enter_isp(self, delay=0.01):
        """Enter ISP mode by controlling the DTR (/RST) and RTS (/ISP) lines

        This operation is performed synchronously, with delays.
        """
        self._uart.rts = True
        time.sleep(delay)
        self._uart.dtr = True
        time.sleep(delay)
        self._uart.dtr = False
        time.sleep(delay)
        self._uart.rts = False

    def synchronize(self, verbose=False):
        """Begin communication with the microcontroller"""
        # Synchronize with the MCU
        while True:
            # Send a ?
            self._uart.write(b"?")
            self._uart.flush()
            if verbose:
                print("?")
            # Receive a response
            try:
                s = self._readline()
            except RecvTimeout:
                continue
            # If we got the right response, break
            if s == b"Synchronized\r\n":
                break
        # Tell the MCU we've synchronized
        s = self._send_command_raw(b"Synchronized\r\n")
        # Next, it should say OK, at which point we're done synchronizing
        if s != b"OK\r\n":
            raise ISPError("Wrong response during synchronization")

        # Send clock frequency in kHz
        s = self._send_command_raw("{:d}\r\n".format(self._clock).encode(
            "utf-8"))
        # Next, it should say OK
        if s != b"OK\r\n":
            raise ISPError("Wrong response during synchronization")

    def close(self):
        """Close the serial port"""
        self._uart.close()

    def unlock(self, code="23130"):
        """Unlock the flash write, erase, and go commands"""
        self._send_command("U {}\r\n".format(code).encode("utf-8"))

    @property
    def baudrate(self):
        """The baud rate used for communication"""
        return self._uart.baudrate

    @baudrate.setter
    def baudrate(self, br):
        self._send_command("B {} {}\r\n".format(br,
            self._uart.stopbits).encode("utf-8"))
        # Update the baud rate for our UART
        self._uart.baudrate = br

    @property
    def stopbits(self):
        """The number of stop bits used for communication"""
        return self._uart.stopbits

    @stopbits.setter
    def stopbits(self, sb):
        self._send_command("B {} {}\r\n".format(self._uart.baudrate,
            sb).encode("utf-8"))
        # Update the number of stop bits for our UART
        self._uart.stopbits = sb

    @property
    def echo(self):
        """Whether the microcontroller echoes characters back to the host"""
        return self._echo

    @echo.setter
    def echo(self, setting):
        setting = bool(setting)
        self._send_command("A {}\r\n".format(int(setting)).encode("utf-8"))
        self._echo = setting

    def write_ram(self, start, data, count=None):
        """Write count bytes from data to RAM at the given start address

        Start and count must be multiples of four.  If count is not specified,
        len(data) is used.
        """
        # Get the length of the data we're writing
        if count is None:
            count = len(data)
        # Ask to write data
        self._send_command("W {} {}\r\n".format(start, count).encode("utf-8"))
        # Send the data
        # NOTE: this is right for LPC8xx chips, not others
        self._writeline(data[:count], plain=True)
        return

    def read_memory(self, start, count):
        """Read count bytes starting at the given address

        Start and count must be multiples of four.
        """
        self._send_command("R {} {}\r\n".format(start, count).encode("utf-8"))
        return self._uart.read(count)

    def prepare_write(self, start, end=None):
        """Prepare the the given flash sector(s) for write operations

        If end is not specified, only the start sector is prepared.
        """
        if end is None:
            end = start
        self._send_command("P {} {}\r\n".format(start, end).encode("utf-8"))

    def copy_ram_to_flash(self, flash, ram, count):
        """Copy count bytes from RAM to flash

        The flash address should be a 64 byte boundary.  Count should be a
        power of two in [64, 1024].
        """
        self._send_command("C {} {} {}\r\n".format(flash, ram, count).encode(
            "utf-8"))

    def go(self, address=0, mode=b"T"):
        """Jump to the given address, in the given mode of execution

        Of course, this function generally causes the ISP command handler to
        stop running, so it is typically appropriate to follow this with a call
        to LPC.close.
        """
        self._writeline("G {} {}\r\n".format(address, mode).encode("utf-8"))


class ReturnCode(Enum):
    """LPC ISP return codes

    From UM10800, section 25.6.1.16.
    """
    CMD_SUCCESS = 0
    INVALID_COMMAND = 1
    SRC_ADDR_ERROR = 2
    DST_ADDR_ERROR = 3
    SRC_ADDR_NOT_MAPPED = 4
    DST_ADDR_NOT_MAPPED = 5
    COUNT_ERROR = 6
    INVALID_SECTOR = 7
    SECTOR_NOT_BLANK = 8
    SECTOR_NOT_PREPARED_FOR_WRITE_OPERATION = 9
    COMPARE_ERROR = 10
    BUSY = 11
    PARAM_ERROR = 12
    ADDR_ERROR = 13
    ADDR_NOT_MAPPED = 14
    CMD_LOCKED = 15
    INVALID_CODE = 16
    INVALID_BAUD_RATE = 17
    INVALID_STOP_BIT = 18
    CODE_READ_PROTECTION_ENABLED = 19


class ISPError(IOError):
    """Generic error for ISP"""


class RecvTimeout(ISPError):
    """Timeout while receiving a command"""