mirror of
https://github.com/google/pebble.git
synced 2025-03-19 10:31:21 +00:00
326 lines
9.5 KiB
Python
326 lines
9.5 KiB
Python
# Copyright 2024 Google LLC
|
|
#
|
|
# 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.
|
|
|
|
# Code adapted from libmpsse implementation
|
|
# May be combined with pyftdi at a later date
|
|
# - please do not mix Tintin code in
|
|
|
|
from pyftdi.pyftdi import ftdi
|
|
from array import array as Array
|
|
|
|
class MPSSE(ftdi.Ftdi):
|
|
I2C = 5
|
|
CMD_SIZE = 3
|
|
# Write bits, not bytes
|
|
MPSSE_WRITE_NEG = 0x01
|
|
MPSSE_BITMODE = 0x02
|
|
MPSSE_READ_NEG = 0x04
|
|
MPSSE_DO_WRITE = 0x10
|
|
MPSSE_DO_READ = 0x20
|
|
|
|
MPSSE_OK = 0
|
|
MPSSE_FAIL = -1
|
|
|
|
ACK = 0
|
|
NACK = 1
|
|
|
|
I2C_TRANSFER_SIZE = 64
|
|
|
|
# Enum low_bits_status
|
|
STARTED = 0
|
|
STOPPED = 1
|
|
|
|
# Pins
|
|
SK = 1
|
|
DO = 2
|
|
DI = 4
|
|
CS = 8
|
|
GPIO0 = 16
|
|
GPIO1 = 32
|
|
GPIO2 = 64
|
|
GPIO3 = 128
|
|
|
|
DEFAULT_TRIS = (SK | DO | GPIO0 | GPIO1 | GPIO2 | GPIO3)
|
|
DEFAULT_PORT = SK
|
|
|
|
MAX_SETUP_COMMANDS = 10
|
|
DISABLE_ADAPTIVE_CLOCK = 0x97
|
|
ENABLE_3_PHASE_CLOCK = 0x8C
|
|
|
|
# 0x0 for MSB, 0x8 for LSB; I2C is MSB
|
|
ENDIANESS = 0x0
|
|
|
|
def __init__(self):
|
|
# Init the superconstructor
|
|
super(MPSSE, self).__init__()
|
|
|
|
# Init, will open the mpsse and setup the pins
|
|
def Open(self, vid, pid, mode=0, interface=1,
|
|
index=0, frequency=1.0E5):
|
|
self.usb_read_timeout = 5000
|
|
# Ack property
|
|
self.rack = 0
|
|
# Start/stop status
|
|
self.status = self.STOPPED
|
|
|
|
# Open the mpsse
|
|
self.open_mpsse(vendor=vid,
|
|
product=pid,
|
|
interface=interface,
|
|
index=index,
|
|
frequency=frequency)
|
|
|
|
# Finish setup
|
|
self._set_mode()
|
|
|
|
# Start condition
|
|
def Start(self):
|
|
status = self.MPSSE_OK
|
|
|
|
# I2C repeated start condition
|
|
if self.status == self.STARTED:
|
|
status |= self._set_bits_low((self.pidle & ~self.SK))
|
|
# Set pins to idle
|
|
status |= self._set_bits_low(self.pidle)
|
|
|
|
# Set start condition
|
|
status |= self._set_bits_low(self.pstart)
|
|
self.status = self.STARTED
|
|
return status
|
|
|
|
# Stop condition
|
|
def Stop(self):
|
|
retval = self.MPSSE_OK
|
|
retval |= self._set_bits_low((self.pidle & ~self.DO & ~self.SK))
|
|
retval |= self._set_bits_low(self.pstop)
|
|
|
|
if retval == self.MPSSE_OK:
|
|
# Pins to idle
|
|
retval |= self._set_bits_low(self.pidle)
|
|
|
|
self.status = self.STOPPED
|
|
return retval
|
|
|
|
# Write in bytes, input MSB
|
|
def Write(self, data):
|
|
n = 0
|
|
# transfer size of I2C is 1
|
|
txsize = 1
|
|
retval = self.MPSSE_FAIL
|
|
size = len(data)
|
|
|
|
while n < size:
|
|
buf = self._build_block_buffer(self.tx, data[n:n+txsize], txsize)
|
|
|
|
retval = self._ftdi_raw_write(buf)
|
|
n += txsize
|
|
|
|
if retval == self.MPSSE_FAIL:
|
|
break
|
|
|
|
# Read in the ACK bit and store it in self.rack
|
|
buf = Array('B')
|
|
t, buf = self._ftdi_raw_read(buf, 1)
|
|
self.rack = buf[0]
|
|
|
|
if retval == self.MPSSE_OK and n == size:
|
|
retval = self.MPSSE_OK
|
|
else:
|
|
retval = self.MPSSE_FAIL
|
|
return retval
|
|
|
|
# Read in bytes, output MSB
|
|
def Read(self, size):
|
|
buf = self._internal_read(size)
|
|
return buf
|
|
|
|
# Ack returned?
|
|
def GetAck(self):
|
|
return self.rack & 0x01
|
|
|
|
def SetAck(self, ack):
|
|
if ack == self.NACK:
|
|
self.tack = 0xFF
|
|
else:
|
|
self.tack = 0x00
|
|
|
|
def SendAcks(self):
|
|
self.SetAck(self.ACK)
|
|
|
|
def SendNacks(self):
|
|
self.SetAck(self.NACK)
|
|
|
|
# Close the I2C when done
|
|
def Close(self):
|
|
self.set_bitmode(0, self.BITMODE_RESET)
|
|
self.close()
|
|
|
|
# Set the low bit pins high/low
|
|
def _set_bits_low(self, port):
|
|
buf = Array('B')
|
|
|
|
buf.append(self.SET_BITS_LOW)
|
|
buf.append(port)
|
|
buf.append(self.tris)
|
|
|
|
return self._ftdi_raw_write(buf)
|
|
|
|
# Part of the setup
|
|
def _set_mode(self):
|
|
retval = self.MPSSE_OK
|
|
setup_commands = Array('B')
|
|
|
|
self.write_data_set_chunksize(65535)
|
|
|
|
# Set tx and rx
|
|
self.tx = self.MPSSE_DO_WRITE | self.ENDIANESS
|
|
self.rx = self.MPSSE_DO_READ | self.ENDIANESS
|
|
self.txrx = self.MPSSE_DO_WRITE | self.MPSSE_DO_READ | self.ENDIANESS
|
|
# Clock, data out, chip select pins are outputs; all others are inputs.
|
|
self.tris = self.DEFAULT_TRIS | self.CS
|
|
# Clock and chip select pins idle high; all others are low
|
|
self.pidle = self.DEFAULT_PORT | self.CS
|
|
self.pstart = self.DEFAULT_PORT | self.CS
|
|
self.pstop = self.DEFAULT_PORT | self.CS
|
|
# During reads and writes the chip select pin is brought low
|
|
self.pstart &= ~self.CS
|
|
|
|
# Send ACKs by default , set tack to 0x00. or 0xFF
|
|
self.tack = 0x00
|
|
|
|
# Ensure adaptive clock is disabled
|
|
setup_commands.append(self.DISABLE_ADAPTIVE_CLOCK)
|
|
|
|
# I2C configurations on pins:
|
|
|
|
# Send on falling clock edge and read data on falling (or rising) clock edge
|
|
self.tx |= self.MPSSE_WRITE_NEG
|
|
self.rx &= ~self.MPSSE_READ_NEG
|
|
# Both the clock and the data lines are idle high
|
|
self.pidle |= self.DO | self.DI
|
|
# Start bit == data line goes from high to low while clock line is high
|
|
self.pstart &= ~self.DO & ~self.DI
|
|
# Stop bit == data line goes from low to high while clock line is high
|
|
# - set data line low here, so the transition to the idle state triggers the stop condition.
|
|
self.pstop &= ~self.DO & ~self.DI
|
|
# Enable three phase clock, data to be available on both rising and falling clock edges
|
|
setup_commands.append(self.ENABLE_3_PHASE_CLOCK)
|
|
setup_commands_size = len(setup_commands)
|
|
|
|
# Send any setup commands to the chip
|
|
if(retval == self.MPSSE_OK) and (setup_commands_size > 0):
|
|
retval = self._ftdi_raw_write(setup_commands)
|
|
|
|
if retval == self.MPSSE_OK:
|
|
# Set the idle pin states
|
|
self._set_bits_low(self.pidle)
|
|
|
|
# All GPIO pins are outputs, set low
|
|
self.trish = 0xFF
|
|
self.gpioh = 0x00
|
|
|
|
buf = Array('B')
|
|
buf.append(self.SET_BITS_HIGH)
|
|
buf.append(self.gpioh)
|
|
buf.append(self.trish)
|
|
retval = self._ftdi_raw_write(buf)
|
|
return retval
|
|
|
|
# Package to send to chip
|
|
def _build_block_buffer(self, cmd, data, size):
|
|
buf = Array('B')
|
|
k = 0
|
|
for j in range(0, size):
|
|
# Clock pin set low prior to clocking data
|
|
buf.append(self.SET_BITS_LOW)
|
|
buf.append(self.pstart & ~self.SK)
|
|
|
|
if cmd == self.rx:
|
|
buf.append(self.tris & ~self.DO)
|
|
else:
|
|
buf.append(self.tris)
|
|
buf.append(cmd)
|
|
buf.append(0)
|
|
if not (cmd & self.MPSSE_BITMODE):
|
|
buf.append(0)
|
|
|
|
# append data input only if write
|
|
if cmd == self.tx or cmd == self.txrx:
|
|
buf.append(ord(data[k]))
|
|
k += 1
|
|
|
|
# In I2C mode clock one ACK bit
|
|
if cmd == self.rx:
|
|
buf.append(self.SET_BITS_LOW)
|
|
buf.append(self.pstart & ~self.SK)
|
|
buf.append(self.tris)
|
|
|
|
buf.append(self.tx | self.MPSSE_BITMODE)
|
|
buf.append(0)
|
|
buf.append(self.tack)
|
|
elif cmd == self.tx:
|
|
buf.append(self.SET_BITS_LOW)
|
|
buf.append(self.pstart & ~self.SK)
|
|
buf.append(self.tris & ~self.DO)
|
|
|
|
buf.append(self.rx | self.MPSSE_BITMODE)
|
|
buf.append(0)
|
|
buf.append(self.SEND_IMMEDIATE)
|
|
return buf
|
|
|
|
def _internal_read(self, size):
|
|
n = 0
|
|
buf = Array('B')
|
|
while n < size:
|
|
rxsize = size - n
|
|
rxsize - min(self.I2C_TRANSFER_SIZE, rxsize)
|
|
|
|
# buf not used by build_block when reading
|
|
data = self._build_block_buffer(self.rx, buf, rxsize)
|
|
retval = self._ftdi_raw_write(data)
|
|
if retval == self.MPSSE_OK:
|
|
t, buf = self._ftdi_raw_read(buf, rxsize)
|
|
if t == 0:
|
|
raise Exception("Corrupt Read")
|
|
n += t
|
|
else:
|
|
break
|
|
return buf
|
|
|
|
# Write data to the FTDI chip
|
|
def _ftdi_raw_write(self, buf):
|
|
if self.write_data(buf) == len(buf):
|
|
return self.MPSSE_OK
|
|
else:
|
|
return self.MPSSE_FAIL
|
|
|
|
# Read data from the FTDI chip
|
|
def _ftdi_raw_read(self, buf, size):
|
|
n = 0
|
|
prev = -1
|
|
while n < size:
|
|
str_buf = self.read_data(size)
|
|
for s in str_buf:
|
|
buf.append(ord(s))
|
|
r = len(buf)
|
|
if r < 0:
|
|
break
|
|
# detect if hanging
|
|
elif r == 0:
|
|
if prev == r:
|
|
return 0, buf
|
|
else:
|
|
prev = r
|
|
n += r
|
|
return n, buf
|