servo/stm32gpio.py

# Copyright 2016 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Allows creation of gpio interface via stm32 usb."""

import array
import logging
import struct
import usb

import gpio_interface
import stm32usb


class SgpioError(Exception):
  """Class for exceptions of Sgpio."""
  def __init__(self, msg, value=0):
    """SgpioError constructor.

    Args:
      msg: string, message describing error in detail
      value: integer, value of error when non-zero status returned.  Default=0
    """
    super(SgpioError, self).__init__(msg, value)
    self.msg = msg
    self.value = value


class Sgpio(gpio_interface.GpioInterface):
  """Provide interface to stm32 gpio USB endpoint.

  Instance Variables:
  _logger: Sgpio tagged log output
  _susb: stm32 usb object
  """

  def __init__(self, vendor=0x18d1,
               product=0x500f, interface=1, serialname=None):
    """Sgpio constructor.

    Loads libraries for libusb.  Creates instance objects
    and Gpio to iteract with the library and intializes them.

    Args:
      vendor    : usb vendor id of stm32 device
      product   : usb product id of stm32 device
      interface : interface number ( 1 - 4 ) of stm32 device to use
      serialname: string of device serialname/number TODO: is this a thing?

    Raises:
      SgpioError: An error accessing Sgpio object
    """
    self._logger = logging.getLogger("Sgpio")
    self._logger.debug("")

    self._susb = stm32usb.Susb(vendor=vendor, product=product,
        interface=interface, serialname=serialname, logger=self._logger)

    self._logger.debug("Set up stm32 gpio")

  def __del__(self):
    """Sgpio destructor."""
    self._logger.debug("Close")

  def wr_rd(self, offset, width=1, dir_val=None, wr_val=None, chip=None,
            muxfile=None):
    """Write and/or read GPIO bit.

    Args:
      offset  : bit offset of the gpio to read or write
      width   : integer, number of contiguous bits in gpio to read or write
      dir_val : Not used. defaulted to None.
      wr_val  : value to write to the GPIO. If unset, skips the write.
      chip    : Not used. defaulted to None.
      muxfile : Not used. defaulted to None.

    Returns:
      integer value from reading the gpio value ( masked & aligned )
    """
    self._logger.debug("Sgpio.wr_rd(offset="
        "%s, width=%s, dir_val=%s, wr_val=%s)" % (
        offset, width, dir_val, wr_val))
    # Read preexisting values for debug output.
    ret = self._susb._read_ep.read(4, self._susb.TIMEOUT_MS)
    read_mask = struct.unpack("<I", ret)[0]
    self._logger.debug("Read mask: 0x%08x" % read_mask)

    width_mask = (1 << width) - 1
    set_mask = 0
    clear_mask = 0

    if wr_val != None:
      set_mask = (wr_val & width_mask) << offset;
      clear_mask = (~wr_val & width_mask) << offset;

    byte_str = struct.pack("<II", set_mask, clear_mask)
    ret = self._susb._write_ep.write(byte_str, self._susb.TIMEOUT_MS)
    if (ret != len(byte_str)):
      raise SgpioError("Wrote %d bytes, expected %d" % (ret, len(byte_str)))

    # GPIO cached values update on read,
    ret = self._susb._read_ep.read(4, self._susb.TIMEOUT_MS)
    ret = self._susb._read_ep.read(4, self._susb.TIMEOUT_MS)
    if len(ret) != 4:
      raise SgpioError(
          "Read error: expected 4 bytes, got %d [%s]" % (len(ret), ret))

    read_mask = struct.unpack("<I", ret)[0]
    self._logger.debug("Read mask: 0x%08x" % read_mask)

    readvalue = (read_mask >> offset) & width_mask
    self._logger.debug("Read value: 0x%x" % readvalue)
    return readvalue


def test():
  """Test code.
  """
  loglevel = logging.DEBUG
  logging.basicConfig(level=loglevel,
                      format="%(asctime)s - %(name)s - " +
                      "%(levelname)s - %(message)s")

  logging.debug("Starting")
  sobj = Sgpio()
  for i in range(1, 2):
    rd_val = sobj.wr_rd(2,wr_val=0)
    logging.debug("rd_val = %d after <2> -> 0" % (rd_val))
    if rd_val != 0:
      logging.error("rd_val = %d != 0" % (rd_val))

    rd_val = sobj.wr_rd(2,wr_val=1)
    logging.debug("rd_val = %d after <2> -> 1" % (rd_val))
    if rd_val != 1:
      logging.error("rd_val = %d != 1" % (rd_val))

    rd_val = sobj.wr_rd(2,1,wr_val=0)
    logging.debug("rd_val = %d after <2> -> 0" % (rd_val))
    if rd_val != 0:
      logging.error("rd_val = %d != 0" % (rd_val))

    # release as output
    rd_val = sobj.wr_rd(2)
    logging.debug("rd_val = %d after <2> dir released" % (rd_val))
    logging.info("rd_val = %d should match pu/pd" % (rd_val))
  logging.debug("Finished")

if __name__ == "__main__":
  test()