I2C_sniffer.py

Go to the documentation of this file.

#!/usr/bin/env python

import time

import pigpio

class sniffer:
   """
   A class to passively monitor activity on an I2C bus.  This should
   work for an I2C bus running at 100kbps or less.  You are unlikely
   to get any usable results for a bus running any faster.
   """

   def __init__(self, pi, SCL, SDA, set_as_inputs=True):
      """
      Instantiate with the Pi and the gpios for the I2C clock
      and data lines.

      If you are monitoring one of the Raspberry Pi buses you
      must set set_as_inputs to False so that they remain in
      I2C mode.

      The pigpio daemon should have been started with a higher
      than default sample rate.

      For an I2C bus rate of 100Kbps sudo pigpiod -s 2 should work.

      A message is printed for each I2C transaction formatted with
      "[" for the START
      "XX" two hex characters for each data byte
      "+" if the data is ACKd, "-" if the data is NACKd
      "]" for the STOP

      E.g. Reading the X, Y, Z values from an ADXL345 gives:

      [A6+32+]
      [A7+01+FF+F2+FF+06+00-]
      """

      self.pi = pi
      self.gSCL = SCL
      self.gSDA = SDA

      self.FALLING = 0
      self.RISING = 1
      self.STEADY = 2

      self.in_data = False
      self.byte = 0
      self.bit = 0
      self.oldSCL = 1
      self.oldSDA = 1

      self.transact = ""

      if set_as_inputs:
         self.pi.set_mode(SCL, pigpio.INPUT)
         self.pi.set_mode(SDA, pigpio.INPUT)

      self.cbA = self.pi.callback(SCL, pigpio.EITHER_EDGE, self._cb)
      self.cbB = self.pi.callback(SDA, pigpio.EITHER_EDGE, self._cb)

   def _parse(self, SCL, SDA):
      """
      Accumulate all the data between START and STOP conditions
      into a string and output when STOP is detected.
      """

      if SCL != self.oldSCL:
         self.oldSCL = SCL
         if SCL:
            xSCL = self.RISING
         else:
            xSCL = self.FALLING
      else:
            xSCL = self.STEADY

      if SDA != self.oldSDA:
         self.oldSDA = SDA
         if SDA:
            xSDA = self.RISING
         else:
            xSDA = self.FALLING
      else:
            xSDA = self.STEADY

      if xSCL == self.RISING:
         if self.in_data:
            if self.bit < 8:
               self.byte = (self.byte << 1) | SDA
               self.bit += 1
            else:
               self.transact += '{:02X}'.format(self.byte)
               if SDA:
                  self.transact += '-'
               else:
                  self.transact += '+'
               self.bit = 0
               self.byte = 0

      elif xSCL == self.STEADY:

         if xSDA == self.RISING:
            if SCL:
               self.in_data = False
               self.byte = 0
               self.bit = 0
               self.transact += ']' # STOP
               print (self.transact)
               self.transact = ""

         if xSDA == self.FALLING:
            if SCL:
               self.in_data = True
               self.byte = 0
               self.bit = 0
               self.transact += '[' # START

   def _cb(self, gpio, level, tick):
      """
      Check which line has altered state (ignoring watchdogs) and
      call the parser with the new state.
      """
      SCL = self.oldSCL
      SDA = self.oldSDA

      if gpio == self.gSCL:
         if level == 0:
            SCL = 0
         elif level == 1:
            SCL = 1

      if gpio == self.gSDA:
         if level == 0:
            SDA = 0
         elif level == 1:
            SDA = 1

      self._parse(SCL, SDA)

   def cancel(self):
      """Cancel the I2C callbacks."""
      self.cbA.cancel()
      self.cbB.cancel()

if __name__ == "__main__":

   import time

   import pigpio

   import I2C_sniffer

   pi = pigpio.pi()

   s = I2C_sniffer.sniffer(pi, 1, 0, False) # leave gpios 1/0 in I2C mode

   time.sleep(60000)

   s.cancel()

   pi.stop()