winbond.py

# SPDX-FileCopyrightText: Copyright (c) 2022 brainelectronics and contributors
# SPDX-License-Identifier: MIT
"""
Implementation to interact with Winbond W25Q Flash with software reset.

Credits & kudos to crizeo
Taken from https://forum.micropython.org/viewtopic.php?f=16&t=3899
Converted from https://github.com/brainelectronics/micropython-winbond
"""

from micropython import const
from microcontroller import delay_us
from adafruit_bus_device.spi_device import SPIDevice

_SECTOR_SIZE = const(4096)
_BLOCK_SIZE = const(512)
_PAGE_SIZE = const(256)

class W25QFlash(object):
    """W25QFlash implementation"""

    def __init__(self,
                 spi: SPI,
                 cs: Pin,
                 baud: int = 40000000,
                 software_reset: bool = True) -> None:
        """
        Constructs a new instance.

        :param      spi:             The SPI object
        :type       spi:             SPI
        :param      cs:              The CS pin object
        :type       cs:              Pin
        :param      baud:            The SCK clock rate
        :type       baud:            int
        :param      software_reset:  Flag to use software reset
        :type       software_reset:  bool
        """
        self.device = SPIDevice(spi, cs, baudrate=baud, phase=1, polarity=1)
        self._busy = False

        if software_reset:
            self.reset()

        # buffer for writing single blocks
        self._cache = bytearray(_SECTOR_SIZE)

        # calc number of bytes (and makes sure the chip is detected and
        # supported)
        self.identify()

        # address length (default: 3 bytes, 32MB+: 4)
        self._ADR_LEN = 3 if (len(bin(self._CAPACITY - 1)) - 2) <= 24 else 4

        # setup address mode:
        if self._ADR_LEN == 4:
            if not self._read_status_reg(nr=16):  # not in 4-byte mode
                print("entering 4-byte address mode")
                self._await()
                with self.device as spi:
                    spi.write(b'\xB7')  # 'Enter 4-Byte Address Mode'

    def reset(self) -> None:
        """
        Reset the Winbond flash if the device has no hardware reset pin.

        See datasheet section 7.2.43 Enable Reset (66h) and Reset (99h)
        Because of the small package and the limitation on the number of pins,
        the W25Q64FV provide a software Reset instruction instead of a
        dedicated RESET pin. Once the Reset instruction is accepted, any
        on-going internal operations will be terminated and the device will
        return to its default power-on state and lose all the current volatile
        settings, such as Volatile Status Register bits, Write Enable Latch
        (WEL) status, Program/Erase Suspend status, Read parameter setting
        (P7-P0), Continuous Read Mode bit setting (M7-M0) and Wrap Bit setting
        (W6-W4).
        "Enable Reset (66h)" and "Reset (99h)" instructions can be issued in
        either SPI mode or QPI mode. To avoid accidental reset, both
        instructions must be issued in sequence. Any other commands other than
        "Reset (99h)" after the "Enable Reset (66h)" command will disable the
        "Reset Enable" state. A new sequence of "Enable Reset (66h)" and
        "Reset (99h)" is needed to reset the device. Once the Reset command is
        accepted by the device, the device will take approximately tRST=30us
        to reset. During this period, no command will be accepted.
        Data corruption may happen if there is an on-going or suspended
        internal Erase or Program operation when Reset command sequence is
        accepted by the device. It is recommended to check the BUSY bit and
        the SUS bit in Status Register before issuing the Reset command
        sequence.
        """
        if self._busy:
            self._await()
        self._busy = True
        with self.device as spi:
            spi.write(b'\x66')  # 'Enable Reset' command
        with self.device as spi:
            spi.write(b'\x99')  # 'Reset' command
        delay_us(30)
        self._busy = False
        # print('Reset performed')

    def identify(self) -> None:
        """
        Identify the Winbond chip.

        Determine the manufacturer and device ID and raises an error if the
        device is not detected or not supported.
        The capacity variable is set to the number of blocks (calculated based
        on the detected chip).
        """
        self._await()
        with self.device as spi:
            spi.write(b'\x9F')  # 'Read JEDEC ID' command
            # manufacturer id, memory type id, capacity id
            buffer = bytearray(3)
            spi.readinto(buffer)
            mf, mem_type, cap = buffer

        self._CAPACITY = int(2**cap)

        if not (mf and mem_type and cap):  # something is 0x00
            raise OSError("device not responding, check wiring. ({}, {}, {})".
                          format(hex(mf), hex(mem_type), hex(cap)))
        if mf != 0xEF or mem_type not in [0x40, 0x60]:
            # Winbond manufacturer, Q25 series memory (tested 0x40 only)
            raise OSError("manufacturer ({}) or memory type ({}) unsupported".
                          format(hex(mf), hex(mem_type)))

        print("manufacturer: {}".format(hex(mf)))               # 0xef
        print("mem_type: {}".format(mem_type))
        print("device: {}".format(hex(mem_type << 8 | cap)))    # 0x4016
        print("capacity: {} bytes".format(self._CAPACITY))      # 4194304 bytes
        # return self._CAPACITY  # calculate number of bytes

    def get_size(self) -> int:
        """
        Get the flash chip size.

        :returns:   The flash size in byte.
        :rtype:     int
        """
        return self._CAPACITY

    def format(self) -> None:
        """
        Format the Winbond flash chip by resetting all memory to 0xFF.

        Important: Run "os.VfsFat.mkfs(flash)" to make the flash an accessible
        file system. As always, you will then need to run
        "os.mount(flash, '/MyFlashDir')" then to mount the flash
        """
        self._wren()
        self._await()
        with self.device as spi:
            spi.write(b'\xC7')  # 'Chip Erase' command
        self._await()  # wait for the chip to finish formatting

    def _read_status_reg(self, nr) -> int:
        """
        Read a status register.

        :param      nr:   Register number to read
        :type       nr:   int

        :returns:   The value (0 or 1) in status register (S0, S1, S2, ...)
        :rtype:     int
        """
        reg, bit = divmod(nr, 8)
        with self.device as spi:
            # 'Read Status Register-...' (1, 2, 3) command
            spi.write((b'\x05', b'\x35', b'\x15')[reg])
            buffer = bytearray(1)
            spi.readinto(buffer, write_value=0xFF)
            stat = 2**bit & buffer[0]

        return stat

    def _await(self) -> None:
        """
        Wait for device not to be busy
        """
        self._busy = True
        buffer = bytearray(1)
        with self.device as spi:
            spi.write(b'\x05')  # 'Read Status Register-1' command

            # last bit (1) is BUSY bit in stat. reg. byte (0 = not busy, 1 = busy)
            spi.readinto(buffer, write_value=0xFF)
            while 0x1 & buffer[0]:
                pass

        self._busy = False

    def _sector_erase(self, addr) -> None:
        """
        Resets all memory within the specified sector (4kB) to 0xFF

        :param      addr:  The address
        :type       addr:  int
        """
        self._wren()
        self._await()
        with self.device as spi:
            spi.write(b'\x20')  # 'Sector Erase' command
            spi.write(addr.to_bytes(self._ADR_LEN, 'big'))

    def _read(self, buf: bytearray, addr: int) -> None:
        """
        Read the length of the buffer bytes from the chip.

        The buffer length has to be a multiple of _SECTOR_SIZE (or less).

        :param      buf:   The buffer
        :type       buf:   list
        :param      addr:  The start address
        :type       addr:  int
        """
        assert addr + len(buf) <= self._CAPACITY, \
            "memory not addressable at %s with range %d (max.: %s)" % \
            (hex(addr), len(buf), hex(self._CAPACITY - 1))
        # print("read {} bytes starting at {}".format(len(buf), hex(addr)))

        self._await()
        with self.device as spi:
            # 'Fast Read' (0x03 = default), 0x0C for 4-byte mode command
            spi.write(b'\x0C' if self._ADR_LEN == 4 else b'\x0B')
            spi.write(addr.to_bytes(self._ADR_LEN, 'big'))
            spi.write(b'\xFF')  # dummy byte
            spi.readinto(buf, 0xFF)

    def _wren(self) -> None:
        """
        Set the Write Enable Latch (WEL) bit in the status register
        """
        self._await()
        with self.device as spi:
            spi.write(b'\x06')  # 'Write Enable' command

    def _write(self, buf: bytearray, addr: int) -> None:
        """
        Write the data of the given buffer to the address location

        Writes the data from <buf> to the device starting at <addr>, which
        has to be erased (0xFF) before. Last byte of <addr> has to be zero,
        which means <addr> has to be a multiple of _PAGE_SIZE (= start
        of page), because wrapping to the next page (if page size exceeded)
        is implemented for full pages only. Length of <buf> has to be a
        multiple of _PAGE_SIZE, because only full pages are supported at
        the moment (<addr> will be auto-incremented).

        :param      buf:   The data buffer to write
        :type       buf:   list
        :param      addr:  The starting address
        :type       addr:  int
        """
        assert len(buf) % _PAGE_SIZE == 0, \
            "invalid buffer length: {}".format(len(buf))
        assert not addr & 0xf, \
            "address ({}) not at page start".format(addr)
        assert addr + len(buf) <= self._CAPACITY, \
            ("memory not addressable at {} with range {} (max.: {})".
                format(hex(addr), len(buf), hex(self._CAPACITY - 1)))
        # print("write buf[{}] to {} ({})".format(len(buf), hex(addr), addr))

        for i in range(0, len(buf), _PAGE_SIZE):
            self._wren()
            self._await()
            with self.device as spi:
                spi.write(b'\x02')  # 'Page Program' command
                spi.write(addr.to_bytes(self._ADR_LEN, 'big'))
                spi.write(buf[i:i + _PAGE_SIZE])
                addr += _PAGE_SIZE

    def _writeblock(self, blocknum: int, buf: list) -> None:
        """
        Write a data block.

        To write a block, the sector (e.g. 4kB = 8 blocks) has to be erased
        first. Therefore, a sector will be read and saved in cache first,
        then the given block will be replaced and the whole sector written
        back when

        :param      blocknum:  The block number
        :type       blocknum:  int
        :param      buf:       The data buffer
        :type       buf:       list
        """
        assert len(buf) == self.BLOCK_SIZE, \
            "invalid block length: {}".format(len(buf))
        # print("writeblock({}, buf[{}])".format(blocknum, len(buf)))

        sector_nr = blocknum // 8
        sector_addr = sector_nr * _SECTOR_SIZE
        # index of first byte of page in sector (multiple of _PAGE_SIZE)
        index = (blocknum << 9) & 0xfff

        self._read(buf=self._cache, addr=sector_addr)
        self._cache[index:index + self.BLOCK_SIZE] = buf  # apply changes
        self._sector_erase(addr=sector_addr)
        # addr is multiple of _SECTOR_SIZE, so last byte is zero
        self._write(buf=self._cache, addr=sector_addr)

    def readblocks(self, blocknum: int, buf: bytearray) -> None:
        """
        Read a data block. The length has to be a multiple of self.BLOCK_SIZE

        :param      blocknum:  The starting block number
        :type       blocknum:  int
        :param      buf:       The data buffer
        :type       buf:       list
        """
        assert len(buf) % _BLOCK_SIZE == 0, \
            'invalid buffer length: {}'.format(len(buf))

        buf_len = len(buf)
        if buf_len == self.BLOCK_SIZE:
            self._read(buf=buf, addr=blocknum << 9)
        else:
            offset = 0
            buf_mv = memoryview(buf)
            while offset < buf_len:
                self._read(buf=buf_mv[offset:offset + self.BLOCK_SIZE],
                           addr=blocknum << 9)
                offset += self.BLOCK_SIZE
                blocknum += 1

    def writeblocks(self, blocknum: int, buf: list) -> None:
        """
        Write a data block.The length has to be a multiple of self.BLOCK_SIZE

        :param      blocknum:  The block number
        :type       blocknum:  int
        :param      buf:       The data buffer
        :type       buf:       list
        """
        assert len(buf) % _BLOCK_SIZE == 0, \
            'invalid buffer length: {}'.format(len(buf))

        buf_len = len(buf)
        if buf_len == self.BLOCK_SIZE:
            self._writeblock(blocknum=blocknum, buf=buf)
        else:
            offset = 0
            buf_mv = memoryview(buf)
            while offset < buf_len:
                self._writeblock(blocknum=blocknum,
                                 buf=buf_mv[offset:offset + self.BLOCK_SIZE])
                offset += self.BLOCK_SIZE
                blocknum += 1

    def count(self) -> int:
        """
        Return the number of blocks available on the device

        :returns:   Number of blocks
        :rtype:     int
        """
        return int(self._CAPACITY / self.BLOCK_SIZE)