signature:
def read ( self , length , assert_ss = True , deassert_ss = True ) :
    """Half - duplex SPI read . If assert _ ss is true , the SS line will be
 asserted low , the specified length of bytes will be clocked in the MISO
 line , and if deassert _ ss is true the SS line will be put back high .
 Bytes which are read will be returned as a bytearray object ."""

implementation:
if self . _miso is None :
    raise RuntimeError ( 'Read attempted with no MISO pin specified.' )
if assert_ss and self . _ss is not None :
    self . _gpio . set_low ( self . _ss )
result = bytearray ( length )
for i in range ( length ) :
    for j in range ( 8 ) : # Flip clock off base .
        self . _gpio . output ( self . _sclk , not self . _clock_base )
        # Handle read on leading edge of clock .
        if self . _read_leading :
            if self . _gpio . is_high ( self . _miso ) : # Set bit to 1 at appropriate location .
                result [ i ] |= self . _read_shift ( self . _mask , j )
            else : # Set bit to 0 at appropriate location .
                result [ i ] &= ~ self . _read_shift ( self . _mask , j )
        # Return clock to base .
        self . _gpio . output ( self . _sclk , self . _clock_base )
        # Handle read on trailing edge of clock .
        if not self . _read_leading :
            if self . _gpio . is_high ( self . _miso ) : # Set bit to 1 at appropriate location .
                result [ i ] |= self . _read_shift ( self . _mask , j )
            else : # Set bit to 0 at appropriate location .
                result [ i ] &= ~ self . _read_shift ( self . _mask , j )
if deassert_ss and self . _ss is not None :
    self . _gpio . set_high ( self . _ss )
return result


signature:
def transfer ( self , data , assert_ss = True , deassert_ss = True ) :
    """Full - duplex SPI read and write . If assert _ ss is true , the SS line
 will be asserted low , the specified bytes will be clocked out the MOSI
 line while bytes will also be read from the MISO line , and if
 deassert _ ss is true the SS line will be put back high . Bytes which are
 read will be returned as a bytearray object ."""

implementation:
if self . _mosi is None :
    raise RuntimeError ( 'Write attempted with no MOSI pin specified.' )
if self . _miso is None :
    raise RuntimeError ( 'Read attempted with no MISO pin specified.' )
if assert_ss and self . _ss is not None :
    self . _gpio . set_low ( self . _ss )
result = bytearray ( len ( data ) )
for i in range ( len ( data ) ) :
    for j in range ( 8 ) : # Write bit to MOSI .
        if self . _write_shift ( data [ i ] , j ) & self . _mask :
            self . _gpio . set_high ( self . _mosi )
        else :
            self . _gpio . set_low ( self . _mosi )
        # Flip clock off base .
        self . _gpio . output ( self . _sclk , not self . _clock_base )
        # Handle read on leading edge of clock .
        if self . _read_leading :
            if self . _gpio . is_high ( self . _miso ) : # Set bit to 1 at appropriate location .
                result [ i ] |= self . _read_shift ( self . _mask , j )
            else : # Set bit to 0 at appropriate location .
                result [ i ] &= ~ self . _read_shift ( self . _mask , j )
        # Return clock to base .
        self . _gpio . output ( self . _sclk , self . _clock_base )
        # Handle read on trailing edge of clock .
        if not self . _read_leading :
            if self . _gpio . is_high ( self . _miso ) : # Set bit to 1 at appropriate location .
                result [ i ] |= self . _read_shift ( self . _mask , j )
            else : # Set bit to 0 at appropriate location .
                result [ i ] &= ~ self . _read_shift ( self . _mask , j )
if deassert_ss and self . _ss is not None :
    self . _gpio . set_high ( self . _ss )
return result


signature:
def setup ( self , pin , value ) :
    """Set the input or output mode for a specified pin . Mode should be
 either GPIO . OUT or GPIO . IN ."""

implementation:
self . _validate_pin ( pin )
# Set bit to 1 for input or 0 for output .
if value == GPIO . IN :
    self . iodir [ int ( pin / 8 ) ] |= 1 << ( int ( pin % 8 ) )
elif value == GPIO . OUT :
    self . iodir [ int ( pin / 8 ) ] &= ~ ( 1 << ( int ( pin % 8 ) ) )
else :
    raise ValueError ( 'Unexpected value.  Must be GPIO.IN or GPIO.OUT.' )
self . write_iodir ( )


signature:
def output_pins ( self , pins ) :
    """Set multiple pins high or low at once . Pins should be a dict of pin
 name to pin value ( HIGH / True for 1 , LOW / False for 0 ) . All provided pins
 will be set to the given values ."""

implementation:
[ self . _validate_pin ( pin ) for pin in pins . keys ( ) ]
# Set each changed pin ' s bit .
for pin , value in iter ( pins . items ( ) ) :
    if value :
        self . gpio [ int ( pin / 8 ) ] |= 1 << ( int ( pin % 8 ) )
    else :
        self . gpio [ int ( pin / 8 ) ] &= ~ ( 1 << ( int ( pin % 8 ) ) )
# Write GPIO state .
self . write_gpio ( )


signature:
def input_pins ( self , pins ) :
    """Read multiple pins specified in the given list and return list of pin values
 GPIO . HIGH / True if the pin is pulled high , or GPIO . LOW / False if pulled low ."""

implementation:
[ self . _validate_pin ( pin ) for pin in pins ]
# Get GPIO state .
self . gpio = self . _device . readList ( self . GPIO , self . gpio_bytes )
# Return True if pin ' s bit is set .
return [ ( self . gpio [ int ( pin / 8 ) ] & 1 << ( int ( pin % 8 ) ) ) > 0 for pin in pins ]


signature:
def pullup ( self , pin , enabled ) :
    """Turn on the pull - up resistor for the specified pin if enabled is True ,
 otherwise turn off the pull - up resistor ."""

implementation:
self . _validate_pin ( pin )
if enabled :
    self . gppu [ int ( pin / 8 ) ] |= 1 << ( int ( pin % 8 ) )
else :
    self . gppu [ int ( pin / 8 ) ] &= ~ ( 1 << ( int ( pin % 8 ) ) )
self . write_gppu ( )


signature:
def write_gpio ( self , gpio = None ) :
    """Write the specified byte value to the GPIO registor . If no value
 specified the current buffered value will be written ."""

implementation:
if gpio is not None :
    self . gpio = gpio
self . _device . writeList ( self . GPIO , self . gpio )


signature:
def write_iodir ( self , iodir = None ) :
    """Write the specified byte value to the IODIR registor . If no value
 specified the current buffered value will be written ."""

implementation:
if iodir is not None :
    self . iodir = iodir
self . _device . writeList ( self . IODIR , self . iodir )


signature:
def write_gppu ( self , gppu = None ) :
    """Write the specified byte value to the GPPU registor . If no value
 specified the current buffered value will be written ."""

implementation:
if gppu is not None :
    self . gppu = gppu
self . _device . writeList ( self . GPPU , self . gppu )


signature:
def disable_FTDI_driver ( ) :
    """Disable the FTDI drivers for the current platform . This is necessary
 because they will conflict with libftdi and accessing the FT232H . Note you
 can enable the FTDI drivers again by calling enable _ FTDI _ driver ."""

implementation:
logger . debug ( 'Disabling FTDI driver.' )
if sys . platform == 'darwin' :
    logger . debug ( 'Detected Mac OSX' )
    # Mac OS commands to disable FTDI driver .
    _check_running_as_root ( )
    subprocess . call ( 'kextunload -b com.apple.driver.AppleUSBFTDI' , shell = True )
    subprocess . call ( 'kextunload /System/Library/Extensions/FTDIUSBSerialDriver.kext' , shell = True )
elif sys . platform . startswith ( 'linux' ) :
    logger . debug ( 'Detected Linux' )
    # Linux commands to disable FTDI driver .
    _check_running_as_root ( )
    subprocess . call ( 'modprobe -r -q ftdi_sio' , shell = True )
    subprocess . call ( 'modprobe -r -q usbserial' , shell = True )


signature:
def enable_FTDI_driver ( ) :
    """Re - enable the FTDI drivers for the current platform ."""

implementation:
logger . debug ( 'Enabling FTDI driver.' )
if sys . platform == 'darwin' :
    logger . debug ( 'Detected Mac OSX' )
    # Mac OS commands to enable FTDI driver .
    _check_running_as_root ( )
    subprocess . check_call ( 'kextload -b com.apple.driver.AppleUSBFTDI' , shell = True )
    subprocess . check_call ( 'kextload /System/Library/Extensions/FTDIUSBSerialDriver.kext' , shell = True )
elif sys . platform . startswith ( 'linux' ) :
    logger . debug ( 'Detected Linux' )
    # Linux commands to enable FTDI driver .
    _check_running_as_root ( )
    subprocess . check_call ( 'modprobe -q ftdi_sio' , shell = True )
    subprocess . check_call ( 'modprobe -q usbserial' , shell = True )


signature:
def enumerate_device_serials ( vid = FT232H_VID , pid = FT232H_PID ) :
    """Return a list of all FT232H device serial numbers connected to the
 machine . You can use these serial numbers to open a specific FT232H device
 by passing it to the FT232H initializer ' s serial parameter ."""

implementation:
try : # Create a libftdi context .
    ctx = None
    ctx = ftdi . new ( )
    # Enumerate FTDI devices .
    device_list = None
    count , device_list = ftdi . usb_find_all ( ctx , vid , pid )
    if count < 0 :
        raise RuntimeError ( 'ftdi_usb_find_all returned error {0}: {1}' . format ( count , ftdi . get_error_string ( self . _ctx ) ) )
    # Walk through list of devices and assemble list of serial numbers .
    devices = [ ]
    while device_list is not None : # Get USB device strings and add serial to list of devices .
        ret , manufacturer , description , serial = ftdi . usb_get_strings ( ctx , device_list . dev , 256 , 256 , 256 )
        if serial is not None :
            devices . append ( serial )
        device_list = device_list . next
    return devices
finally : # Make sure to clean up list and context when done .
    if device_list is not None :
        ftdi . list_free ( device_list )
    if ctx is not None :
        ftdi . free ( ctx )


signature:
def close ( self ) :
    """Close the FTDI device . Will be automatically called when the program ends ."""

implementation:
if self . _ctx is not None :
    ftdi . free ( self . _ctx )
self . _ctx = None


signature:
def _write ( self , string ) :
    """Helper function to call write _ data on the provided FTDI device and
 verify it succeeds ."""

implementation:
# Get modem status . Useful to enable for debugging .
# ret , status = ftdi . poll _ modem _ status ( self . _ ctx )
# if ret = = 0:
# logger . debug ( ' Modem status { 0:02X } ' . format ( status ) )
# else :
# logger . debug ( ' Modem status error { 0 } ' . format ( ret ) )
length = len ( string )
try :
    ret = ftdi . write_data ( self . _ctx , string , length )
except TypeError :
    ret = ftdi . write_data ( self . _ctx , string ) ;
    # compatible with libFtdi 1.3
# Log the string that was written in a python hex string format using a very
# ugly one - liner list comprehension for brevity .
# logger . debug ( ' Wrote { 0 } ' . format ( ' ' . join ( [ ' \ \ x { 0:02X } ' . format ( ord ( x ) ) for x in string ] ) ) )
if ret < 0 :
    raise RuntimeError ( 'ftdi_write_data failed with error {0}: {1}' . format ( ret , ftdi . get_error_string ( self . _ctx ) ) )
if ret != length :
    raise RuntimeError ( 'ftdi_write_data expected to write {0} bytes but actually wrote {1}!' . format ( length , ret ) )


signature:
def _check ( self , command , * args ) :
    """Helper function to call the provided command on the FTDI device and
 verify the response matches the expected value ."""

implementation:
ret = command ( self . _ctx , * args )
logger . debug ( 'Called ftdi_{0} and got response {1}.' . format ( command . __name__ , ret ) )
if ret != 0 :
    raise RuntimeError ( 'ftdi_{0} failed with error {1}: {2}' . format ( command . __name__ , ret , ftdi . get_error_string ( self . _ctx ) ) )


signature:
def _poll_read ( self , expected , timeout_s = 5.0 ) :
    """Helper function to continuously poll reads on the FTDI device until an
 expected number of bytes are returned . Will throw a timeout error if no
 data is received within the specified number of timeout seconds . Returns
 the read data as a string if successful , otherwise raises an execption ."""

implementation:
start = time . time ( )
# Start with an empty response buffer .
response = bytearray ( expected )
index = 0
# Loop calling read until the response buffer is full or a timeout occurs .
while time . time ( ) - start <= timeout_s :
    ret , data = ftdi . read_data ( self . _ctx , expected - index )
    # Fail if there was an error reading data .
    if ret < 0 :
        raise RuntimeError ( 'ftdi_read_data failed with error code {0}.' . format ( ret ) )
    # Add returned data to the buffer .
    response [ index : index + ret ] = data [ : ret ]
    index += ret
    # Buffer is full , return the result data .
    if index >= expected :
        return str ( response )
    time . sleep ( 0.01 )
raise RuntimeError ( 'Timeout while polling ftdi_read_data for {0} bytes!' . format ( expected ) )


signature:
def _mpsse_enable ( self ) :
    """Enable MPSSE mode on the FTDI device ."""

implementation:
# Reset MPSSE by sending mask = 0 and mode = 0
self . _check ( ftdi . set_bitmode , 0 , 0 )
# Enable MPSSE by sending mask = 0 and mode = 2
self . _check ( ftdi . set_bitmode , 0 , 2 )


signature:
def _mpsse_sync ( self , max_retries = 10 ) :
    """Synchronize buffers with MPSSE by sending bad opcode and reading expected
 error response . Should be called once after enabling MPSSE ."""

implementation:
# Send a bad / unknown command ( 0xAB ) , then read buffer until bad command
# response is found .
self . _write ( '\xAB' )
# Keep reading until bad command response ( 0xFA 0xAB ) is returned .
# Fail if too many read attempts are made to prevent sticking in a loop .
tries = 0
sync = False
while not sync :
    data = self . _poll_read ( 2 )
    if data == '\xFA\xAB' :
        sync = True
    tries += 1
    if tries >= max_retries :
        raise RuntimeError ( 'Could not synchronize with FT232H!' )


signature:
def mpsse_set_clock ( self , clock_hz , adaptive = False , three_phase = False ) :
    """Set the clock speed of the MPSSE engine . Can be any value from 450hz
 to 30mhz and will pick that speed or the closest speed below it ."""

implementation:
# Disable clock divisor by 5 to enable faster speeds on FT232H .
self . _write ( '\x8A' )
# Turn on / off adaptive clocking .
if adaptive :
    self . _write ( '\x96' )
else :
    self . _write ( '\x97' )
# Turn on / off three phase clock ( needed for I2C ) .
# Also adjust the frequency for three - phase clocking as specified in section 2.2.4
# of this document :
# http : / / www . ftdichip . com / Support / Documents / AppNotes / AN _ 255 _ USB % 20to % 20I2C % 20Example % 20using % 20the % 20FT232H % 20and % 20FT201X % 20devices . pdf
if three_phase :
    self . _write ( '\x8C' )
else :
    self . _write ( '\x8D' )
# Compute divisor for requested clock .
# Use equation from section 3.8.1 of :
# http : / / www . ftdichip . com / Support / Documents / AppNotes / AN _ 108 _ Command _ Processor _ for _ MPSSE _ and _ MCU _ Host _ Bus _ Emulation _ Modes . pdf
# Note equation is using 60mhz master clock instead of 12mhz .
divisor = int ( math . ceil ( ( 30000000.0 - float ( clock_hz ) ) / float ( clock_hz ) ) ) & 0xFFFF
if three_phase :
    divisor = int ( divisor * ( 2.0 / 3.0 ) )
logger . debug ( 'Setting clockspeed with divisor value {0}' . format ( divisor ) )
# Send command to set divisor from low and high byte values .
self . _write ( str ( bytearray ( ( 0x86 , divisor & 0xFF , ( divisor >> 8 ) & 0xFF ) ) ) )


signature:
def mpsse_read_gpio ( self ) :
    """Read both GPIO bus states and return a 16 bit value with their state .
 D0 - D7 are the lower 8 bits and C0 - C7 are the upper 8 bits ."""

implementation:
# Send command to read low byte and high byte .
self . _write ( '\x81\x83' )
# Wait for 2 byte response .
data = self . _poll_read ( 2 )
# Assemble response into 16 bit value .
low_byte = ord ( data [ 0 ] )
high_byte = ord ( data [ 1 ] )
logger . debug ( 'Read MPSSE GPIO low byte = {0:02X} and high byte = {1:02X}' . format ( low_byte , high_byte ) )
return ( high_byte << 8 ) | low_byte


signature:
def mpsse_gpio ( self ) :
    """Return command to update the MPSSE GPIO state to the current direction
 and level ."""

implementation:
level_low = chr ( self . _level & 0xFF )
level_high = chr ( ( self . _level >> 8 ) & 0xFF )
dir_low = chr ( self . _direction & 0xFF )
dir_high = chr ( ( self . _direction >> 8 ) & 0xFF )
return str ( bytearray ( ( 0x80 , level_low , dir_low , 0x82 , level_high , dir_high ) ) )


signature:
def setup ( self , pin , mode ) :
    """Set the input or output mode for a specified pin . Mode should be
 either OUT or IN ."""

implementation:
self . _setup_pin ( pin , mode )
self . mpsse_write_gpio ( )


signature:
def setup_pins ( self , pins , values = { } , write = True ) :
    """Setup multiple pins as inputs or outputs at once . Pins should be a
 dict of pin name to pin mode ( IN or OUT ) . Optional starting values of
 pins can be provided in the values dict ( with pin name to pin value ) ."""

implementation:
# General implementation that can be improved by subclasses .
for pin , mode in iter ( pins . items ( ) ) :
    self . _setup_pin ( pin , mode )
for pin , value in iter ( values . items ( ) ) :
    self . _output_pin ( pin , value )
if write :
    self . mpsse_write_gpio ( )


signature:
def output ( self , pin , value ) :
    """Set the specified pin the provided high / low value . Value should be
 either HIGH / LOW or a boolean ( true = high ) ."""

implementation:
if pin < 0 or pin > 15 :
    raise ValueError ( 'Pin must be between 0 and 15 (inclusive).' )
self . _output_pin ( pin , value )
self . mpsse_write_gpio ( )


signature:
def output_pins ( self , pins , write = True ) :
    """Set multiple pins high or low at once . Pins should be a dict of pin
 name to pin value ( HIGH / True for 1 , LOW / False for 0 ) . All provided pins
 will be set to the given values ."""

implementation:
for pin , value in iter ( pins . items ( ) ) :
    self . _output_pin ( pin , value )
if write :
    self . mpsse_write_gpio ( )


signature:
def input_pins ( self , pins ) :
    """Read multiple pins specified in the given list and return list of pin values
 GPIO . HIGH / True if the pin is pulled high , or GPIO . LOW / False if pulled low ."""

implementation:
if [ pin for pin in pins if pin < 0 or pin > 15 ] :
    raise ValueError ( 'Pin must be between 0 and 15 (inclusive).' )
_pins = self . mpsse_read_gpio ( )
return [ ( ( _pins >> pin ) & 0x0001 ) == 1 for pin in pins ]


signature:
def set_mode ( self , mode ) :
    """Set SPI mode which controls clock polarity and phase . Should be a
 numeric value 0 , 1 , 2 , or 3 . See wikipedia page for details on meaning :
 http : / / en . wikipedia . org / wiki / Serial _ Peripheral _ Interface _ Bus"""

implementation:
if mode < 0 or mode > 3 :
    raise ValueError ( 'Mode must be a value 0, 1, 2, or 3.' )
if mode == 0 : # Mode 0 captures on rising clock , propagates on falling clock
    self . write_clock_ve = 1
    self . read_clock_ve = 0
    # Clock base is low .
    clock_base = GPIO . LOW
elif mode == 1 : # Mode 1 capture of falling edge , propagate on rising clock
    self . write_clock_ve = 0
    self . read_clock_ve = 1
    # Clock base is low .
    clock_base = GPIO . LOW
elif mode == 2 : # Mode 2 capture on rising clock , propagate on falling clock
    self . write_clock_ve = 1
    self . read_clock_ve = 0
    # Clock base is high .
    clock_base = GPIO . HIGH
elif mode == 3 : # Mode 3 capture on falling edge , propagage on rising clock
    self . write_clock_ve = 0
    self . read_clock_ve = 1
    # Clock base is high .
    clock_base = GPIO . HIGH
# Set clock and DO as output , DI as input . Also start clock at its base value .
self . _ft232h . setup_pins ( { 0 : GPIO . OUT , 1 : GPIO . OUT , 2 : GPIO . IN } , { 0 : clock_base } )


signature:
def set_bit_order ( self , order ) :
    """Set order of bits to be read / written over serial lines . Should be
 either MSBFIRST for most - significant first , or LSBFIRST for
 least - signifcant first ."""

implementation:
if order == MSBFIRST :
    self . lsbfirst = 0
elif order == LSBFIRST :
    self . lsbfirst = 1
else :
    raise ValueError ( 'Order must be MSBFIRST or LSBFIRST.' )


signature:
def write ( self , data ) :
    """Half - duplex SPI write . The specified array of bytes will be clocked
 out the MOSI line ."""

implementation:
# check for hardware limit of FT232H and similar MPSSE chips
if ( len ( data ) > 65536 ) :
    print ( 'the FTDI chip is limited to 65536 bytes (64 KB) of input/output per command!' )
    print ( 'use for loops for larger reads' )
    exit ( 1 )
# Build command to write SPI data .
command = 0x10 | ( self . lsbfirst << 3 ) | self . write_clock_ve
logger . debug ( 'SPI write with command {0:2X}.' . format ( command ) )
# Compute length low and high bytes .
# NOTE : Must actually send length minus one because the MPSSE engine
# considers 0 a length of 1 and FFFF a length of 65536
# splitting into two lists for two commands to prevent buffer errors
data1 = data [ : len ( data ) / 2 ]
data2 = data [ len ( data ) / 2 : ]
len_low1 = ( len ( data1 ) - 1 ) & 0xFF
len_high1 = ( ( len ( data1 ) - 1 ) >> 8 ) & 0xFF
len_low2 = ( len ( data2 ) - 1 ) & 0xFF
len_high2 = ( ( len ( data2 ) - 1 ) >> 8 ) & 0xFF
self . _assert_cs ( )
# Send command and length , then data , split into two commands , handle for length 1
if len ( data1 ) > 0 :
    self . _ft232h . _write ( str ( bytearray ( ( command , len_low1 , len_high1 ) ) ) )
    self . _ft232h . _write ( str ( bytearray ( data1 ) ) )
if len ( data2 ) > 0 :
    self . _ft232h . _write ( str ( bytearray ( ( command , len_low2 , len_high2 ) ) ) )
    self . _ft232h . _write ( str ( bytearray ( data2 ) ) )
self . _deassert_cs ( )


signature:
def read ( self , length ) :
    """Half - duplex SPI read . The specified length of bytes will be clocked
 in the MISO line and returned as a bytearray object ."""

implementation:
# check for hardware limit of FT232H and similar MPSSE chips
if ( 1 > length > 65536 ) :
    print ( 'the FTDI chip is limited to 65536 bytes (64 KB) of input/output per command!' )
    print ( 'use for loops for larger reads' )
    exit ( 1 )
# Build command to read SPI data .
command = 0x20 | ( self . lsbfirst << 3 ) | ( self . read_clock_ve << 2 )
logger . debug ( 'SPI read with command {0:2X}.' . format ( command ) )
# Compute length low and high bytes .
# NOTE : Must actually send length minus one because the MPSSE engine
# considers 0 a length of 1 and FFFF a length of 65536
# force odd numbers to round up instead of down
lengthR = length
if length % 2 == 1 :
    lengthR += 1
lengthR = lengthR / 2
# when odd length requested , get the remainder instead of the same number
lenremain = length - lengthR
len_low = ( lengthR - 1 ) & 0xFF
len_high = ( ( lengthR - 1 ) >> 8 ) & 0xFF
self . _assert_cs ( )
# Send command and length .
# Perform twice to prevent error from hardware defect / limits
self . _ft232h . _write ( str ( bytearray ( ( command , len_low , len_high ) ) ) )
payload1 = self . _ft232h . _poll_read ( lengthR )
self . _ft232h . _write ( str ( bytearray ( ( command , len_low , len_high ) ) ) )
payload2 = self . _ft232h . _poll_read ( lenremain )
self . _deassert_cs ( )
# Read response bytes
return bytearray ( payload1 + payload2 )


signature:
def bulkread ( self , data = [ ] , lengthR = 'None' , readmode = 1 ) :
    """Half - duplex SPI write then read . Send command and payload to slave as bytearray
 then consequently read out response from the slave for length in bytes .
 Designed for use with NOR or NAND flash chips , and possibly SD cards . . . etc . . .
 Read command is cut in half and performed twice in series to prevent single byte errors .
 Hardware limits per command are enforced before doing anything .
 Read length is an optional argument , so that it can function similar to transfer
 but still half - duplex .
 For reading without writing , one can send a blank array or skip that argument ."""

implementation:
# check for hardware limit of FT232H and similar MPSSE chips
if ( 1 > lengthR > 65536 ) | ( len ( data ) > 65536 ) :
    print ( 'the FTDI chip is limited to 65536 bytes (64 KB) of input/output per command!' )
    print ( 'use for loops for larger reads' )
    exit ( 1 )
# default mode is to act like ` transfer ` but half - duplex
if ( lengthR == 'None' ) & ( readmode == 1 ) :
    lengthR = len ( data )
# command parameters definition and math
# MPSSE engine sees length 0 as 1 byte , so - 1 lengths
commandW = 0x10 | ( self . lsbfirst << 3 ) | self . write_clock_ve
lengthW = len ( data ) - 1
len_lowW = ( lengthW ) & 0xFF
len_highW = ( ( lengthW ) >> 8 ) & 0xFF
commandR = 0x20 | ( self . lsbfirst << 3 ) | ( self . read_clock_ve << 2 )
# force odd numbers to round up instead of down
length = lengthR
if lengthR % 2 == 1 :
    length += 1
length = length / 2
# when odd length requested , get the remainder instead of the same number
lenremain = lengthR - length
len_lowR = ( length - 1 ) & 0xFF
len_highR = ( ( length - 1 ) >> 8 ) & 0xFF
# logger debug info
logger . debug ( 'SPI bulkread with write command {0:2X}.' . format ( commandW ) )
logger . debug ( 'and read command {0:2X}.' . format ( commandR ) )
# begin command set
self . _assert_cs ( )
# write command , these have to be separated due to TypeError
self . _ft232h . _write ( str ( bytearray ( ( commandW , len_lowW , len_highW ) ) ) )
self . _ft232h . _write ( str ( bytearray ( data ) ) )
# read command , which is divided into two commands
self . _ft232h . _write ( str ( bytearray ( ( commandR , len_lowR , len_highR ) ) ) )
payload1 = self . _ft232h . _poll_read ( length )
self . _ft232h . _write ( str ( bytearray ( ( commandR , len_lowR , len_highR ) ) ) )
payload2 = self . _ft232h . _poll_read ( lenremain )
self . _deassert_cs ( )
# end command set
# Read response bytes
return bytearray ( payload1 + payload2 )


signature:
def transfer ( self , data ) :
    """Full - duplex SPI read and write . The specified array of bytes will be
 clocked out the MOSI line , while simultaneously bytes will be read from
 the MISO line . Read bytes will be returned as a bytearray object ."""

implementation:
# check for hardware limit of FT232H and similar MPSSE chips
if ( len ( data ) > 65536 ) :
    print ( 'the FTDI chip is limited to 65536 bytes (64 KB) of input/output per command!' )
    print ( 'use for loops for larger reads' )
    exit ( 1 )
# Build command to read and write SPI data .
command = 0x30 | ( self . lsbfirst << 3 ) | ( self . read_clock_ve << 2 ) | self . write_clock_ve
logger . debug ( 'SPI transfer with command {0:2X}.' . format ( command ) )
# Compute length low and high bytes .
# NOTE : Must actually send length minus one because the MPSSE engine
# considers 0 a length of 1 and FFFF a length of 65536
data1 = data [ : len ( data ) / 2 ]
data2 = data [ len ( data ) / 2 : ]
len_low1 = ( len ( data1 ) - 1 ) & 0xFF
len_high1 = ( ( len ( data1 ) - 1 ) >> 8 ) & 0xFF
len_low2 = ( len ( data2 ) - 1 ) & 0xFF
len_high2 = ( ( len ( data2 ) - 1 ) >> 8 ) & 0xFF
payload1 = ''
payload2 = ''
# start command set
self . _assert_cs ( )
# Perform twice to prevent error from hardware defect / limits
# Send command and length , then data , split into two commands , handle for length 1
if len ( data1 ) > 0 :
    self . _ft232h . _write ( str ( bytearray ( ( command , len_low1 , len_high1 ) ) ) )
    self . _ft232h . _write ( str ( bytearray ( data1 ) ) )
    payload1 = self . _ft232h . _poll_read ( len ( data1 ) )
if len ( data2 ) > 0 :
    self . _ft232h . _write ( str ( bytearray ( ( command , len_low2 , len_high2 ) ) ) )
    self . _ft232h . _write ( str ( bytearray ( data2 ) ) )
    payload2 = self . _ft232h . _poll_read ( len ( data2 ) )
# self . _ ft232h . _ write ( ' \ x87 ' )
self . _deassert_cs ( )
# Read response bytes .
return bytearray ( payload1 + payload2 )


signature:
def _idle ( self ) :
    """Put I2C lines into idle state ."""

implementation:
# Put the I2C lines into an idle state with SCL and SDA high .
self . _ft232h . setup_pins ( { 0 : GPIO . OUT , 1 : GPIO . OUT , 2 : GPIO . IN } , { 0 : GPIO . HIGH , 1 : GPIO . HIGH } )


signature:
def _transaction_end ( self ) :
    """End I2C transaction and get response bytes , including ACKs ."""

implementation:
# Ask to return response bytes immediately .
self . _command . append ( '\x87' )
# Send the entire command to the MPSSE .
self . _ft232h . _write ( '' . join ( self . _command ) )
# Read response bytes and return them .
return bytearray ( self . _ft232h . _poll_read ( self . _expected ) )


signature:
def _i2c_start ( self ) :
    """Send I2C start signal . Must be called within a transaction start / end ."""

implementation:
# Set SCL high and SDA low , repeat 4 times to stay in this state for a
# short period of time .
self . _ft232h . output_pins ( { 0 : GPIO . HIGH , 1 : GPIO . LOW } , write = False )
self . _command . append ( self . _ft232h . mpsse_gpio ( ) * _REPEAT_DELAY )
# Now drop SCL to low ( again repeat 4 times for short delay ) .
self . _ft232h . output_pins ( { 0 : GPIO . LOW , 1 : GPIO . LOW } , write = False )
self . _command . append ( self . _ft232h . mpsse_gpio ( ) * _REPEAT_DELAY )


signature:
def _i2c_idle ( self ) :
    """Set I2C signals to idle state with SCL and SDA at a high value . Must
 be called within a transaction start / end ."""

implementation:
self . _ft232h . output_pins ( { 0 : GPIO . HIGH , 1 : GPIO . HIGH } , write = False )
self . _command . append ( self . _ft232h . mpsse_gpio ( ) * _REPEAT_DELAY )


signature:
def _i2c_stop ( self ) :
    """Send I2C stop signal . Must be called within a transaction start / end ."""

implementation:
# Set SCL low and SDA low for a short period .
self . _ft232h . output_pins ( { 0 : GPIO . LOW , 1 : GPIO . LOW } , write = False )
self . _command . append ( self . _ft232h . mpsse_gpio ( ) * _REPEAT_DELAY )
# Set SCL high and SDA low for a short period .
self . _ft232h . output_pins ( { 0 : GPIO . HIGH , 1 : GPIO . LOW } , write = False )
self . _command . append ( self . _ft232h . mpsse_gpio ( ) * _REPEAT_DELAY )
# Finally set SCL high and SDA high for a short period .
self . _ft232h . output_pins ( { 0 : GPIO . HIGH , 1 : GPIO . HIGH } , write = False )
self . _command . append ( self . _ft232h . mpsse_gpio ( ) * _REPEAT_DELAY )


signature:
def _i2c_read_bytes ( self , length = 1 ) :
    """Read the specified number of bytes from the I2C bus . Length is the
 number of bytes to read ( must be 1 or more ) ."""

implementation:
for i in range ( length - 1 ) : # Read a byte and send ACK .
    self . _command . append ( '\x20\x00\x00\x13\x00\x00' )
    # Make sure pins are back in idle state with clock low and data high .
    self . _ft232h . output_pins ( { 0 : GPIO . LOW , 1 : GPIO . HIGH } , write = False )
    self . _command . append ( self . _ft232h . mpsse_gpio ( ) )
# Read last byte and send NAK .
self . _command . append ( '\x20\x00\x00\x13\x00\xFF' )
# Make sure pins are back in idle state with clock low and data high .
self . _ft232h . output_pins ( { 0 : GPIO . LOW , 1 : GPIO . HIGH } , write = False )
self . _command . append ( self . _ft232h . mpsse_gpio ( ) )
# Increase expected number of bytes .
self . _expected += length


signature:
def _i2c_write_bytes ( self , data ) :
    """Write the specified number of bytes to the chip ."""

implementation:
for byte in data : # Write byte .
    self . _command . append ( str ( bytearray ( ( 0x11 , 0x00 , 0x00 , byte ) ) ) )
    # Make sure pins are back in idle state with clock low and data high .
    self . _ft232h . output_pins ( { 0 : GPIO . LOW , 1 : GPIO . HIGH } , write = False )
    self . _command . append ( self . _ft232h . mpsse_gpio ( ) * _REPEAT_DELAY )
    # Read bit for ACK / NAK .
    self . _command . append ( '\x22\x00' )
# Increase expected response bytes .
self . _expected += len ( data )


signature:
def ping ( self ) :
    """Attempt to detect if a device at this address is present on the I2C
 bus . Will send out the device ' s address for writing and verify an ACK
 is received . Returns true if the ACK is received , and false if not ."""

implementation:
self . _idle ( )
self . _transaction_start ( )
self . _i2c_start ( )
self . _i2c_write_bytes ( [ self . _address_byte ( False ) ] )
self . _i2c_stop ( )
response = self . _transaction_end ( )
if len ( response ) != 1 :
    raise RuntimeError ( 'Expected 1 response byte but received {0} byte(s).' . format ( len ( response ) ) )
return ( ( response [ 0 ] & 0x01 ) == 0x00 )


signature:
def writeRaw8 ( self , value ) :
    """Write an 8 - bit value on the bus ( without register ) ."""

implementation:
value = value & 0xFF
self . _idle ( )
self . _transaction_start ( )
self . _i2c_start ( )
self . _i2c_write_bytes ( [ self . _address_byte ( False ) , value ] )
self . _i2c_stop ( )
response = self . _transaction_end ( )
self . _verify_acks ( response )


signature:
def write16 ( self , register , value , little_endian = True ) :
    """Write a 16 - bit value to the specified register ."""

implementation:
value = value & 0xFFFF
value_low = value & 0xFF
value_high = ( value >> 8 ) & 0xFF
if not little_endian :
    value_low , value_high = value_high , value_low
self . _idle ( )
self . _transaction_start ( )
self . _i2c_start ( )
self . _i2c_write_bytes ( [ self . _address_byte ( False ) , register , value_low , value_high ] )
self . _i2c_stop ( )
response = self . _transaction_end ( )
self . _verify_acks ( response )


signature:
def writeList ( self , register , data ) :
    """Write bytes to the specified register ."""

implementation:
self . _idle ( )
self . _transaction_start ( )
self . _i2c_start ( )
self . _i2c_write_bytes ( [ self . _address_byte ( False ) , register ] + data )
self . _i2c_stop ( )
response = self . _transaction_end ( )
self . _verify_acks ( response )


signature:
def readList ( self , register , length ) :
    """Read a length number of bytes from the specified register . Results
 will be returned as a bytearray ."""

implementation:
if length <= 0 :
    raise ValueError ( "Length must be at least 1 byte." )
self . _idle ( )
self . _transaction_start ( )
self . _i2c_start ( )
self . _i2c_write_bytes ( [ self . _address_byte ( True ) , register ] )
self . _i2c_stop ( )
self . _i2c_idle ( )
self . _i2c_start ( )
self . _i2c_read_bytes ( length )
self . _i2c_stop ( )
response = self . _transaction_end ( )
self . _verify_acks ( response [ : - length ] )
return response [ - length : ]


signature:
def readRaw8 ( self ) :
    """Read an 8 - bit value on the bus ( without register ) ."""

implementation:
self . _idle ( )
self . _transaction_start ( )
self . _i2c_start ( )
self . _i2c_write_bytes ( [ self . _address_byte ( False ) ] )
self . _i2c_stop ( )
self . _i2c_idle ( )
self . _i2c_start ( )
self . _i2c_write_bytes ( [ self . _address_byte ( True ) ] )
self . _i2c_read_bytes ( 1 )
self . _i2c_stop ( )
response = self . _transaction_end ( )
self . _verify_acks ( response [ : - 1 ] )
return response [ - 1 ]


signature:
def readS8 ( self , register ) :
    """Read a signed byte from the specified register ."""

implementation:
result = self . readU8 ( register )
if result > 127 :
    result -= 256
return result


signature:
def readS16 ( self , register , little_endian = True ) :
    """Read a signed 16 - bit value from the specified register , with the
 specified endianness ( default little endian , or least significant byte
 first ) ."""

implementation:
result = self . readU16 ( register , little_endian )
if result > 32767 :
    result -= 65536
return result


signature:
def get_default_bus ( ) :
    """Return the default bus number based on the device platform . For a
 Raspberry Pi either bus 0 or 1 ( based on the Pi revision ) will be returned .
 For a Beaglebone Black the first user accessible bus , 1 , will be returned ."""

implementation:
plat = Platform . platform_detect ( )
if plat == Platform . RASPBERRY_PI :
    if Platform . pi_revision ( ) == 1 : # Revision 1 Pi uses I2C bus 0.
        return 0
    else : # Revision 2 Pi uses I2C bus 1.
        return 1
elif plat == Platform . BEAGLEBONE_BLACK : # Beaglebone Black has multiple I2C buses , default to 1 ( P9_19 and P9_20 ) .
    return 1
else :
    raise RuntimeError ( 'Could not determine default I2C bus for platform.' )


signature:
def get_i2c_device ( address , busnum = None , i2c_interface = None , ** kwargs ) :
    """Return an I2C device for the specified address and on the specified bus .
 If busnum isn ' t specified , the default I2C bus for the platform will attempt
 to be detected ."""

implementation:
if busnum is None :
    busnum = get_default_bus ( )
return Device ( address , busnum , i2c_interface , ** kwargs )


signature:
def require_repeated_start ( ) :
    """Enable repeated start conditions for I2C register reads . This is the
 normal behavior for I2C , however on some platforms like the Raspberry Pi
 there are bugs which disable repeated starts unless explicitly enabled with
 this function . See this thread for more details :
 http : / / www . raspberrypi . org / forums / viewtopic . php ? f = 44 & t = 15840"""

implementation:
plat = Platform . platform_detect ( )
if plat == Platform . RASPBERRY_PI and os . path . exists ( '/sys/module/i2c_bcm2708/parameters/combined' ) : # On the Raspberry Pi there is a bug where register reads don ' t send a
# repeated start condition like the kernel smbus I2C driver functions
# define . As a workaround this bit in the BCM2708 driver sysfs tree can
# be changed to enable I2C repeated starts .
    subprocess . check_call ( 'chmod 666 /sys/module/i2c_bcm2708/parameters/combined' , shell = True )
    subprocess . check_call ( 'echo -n 1 > /sys/module/i2c_bcm2708/parameters/combined' , shell = True )


implementation:
value = value & 0xFF
self . _bus . write_byte ( self . _address , value )
self . _logger . debug ( "Wrote 0x%02X" , value )


signature:
def write8 ( self , register , value ) :
    """Write an 8 - bit value to the specified register ."""

implementation:
value = value & 0xFF
self . _bus . write_byte_data ( self . _address , register , value )
self . _logger . debug ( "Wrote 0x%02X to register 0x%02X" , value , register )


signature:
def write16 ( self , register , value ) :
    """Write a 16 - bit value to the specified register ."""

implementation:
value = value & 0xFFFF
self . _bus . write_word_data ( self . _address , register , value )
self . _logger . debug ( "Wrote 0x%04X to register pair 0x%02X, 0x%02X" , value , register , register + 1 )


implementation:
self . _bus . write_i2c_block_data ( self . _address , register , data )
self . _logger . debug ( "Wrote to register 0x%02X: %s" , register , data )


implementation:
results = self . _bus . read_i2c_block_data ( self . _address , register , length )
self . _logger . debug ( "Read the following from register 0x%02X: %s" , register , results )
return results


implementation:
result = self . _bus . read_byte ( self . _address ) & 0xFF
self . _logger . debug ( "Read 0x%02X" , result )
return result


signature:
def readU8 ( self , register ) :
    """Read an unsigned byte from the specified register ."""

implementation:
result = self . _bus . read_byte_data ( self . _address , register ) & 0xFF
self . _logger . debug ( "Read 0x%02X from register 0x%02X" , result , register )
return result


signature:
def readU16 ( self , register , little_endian = True ) :
    """Read an unsigned 16 - bit value from the specified register , with the
 specified endianness ( default little endian , or least significant byte
 first ) ."""

implementation:
result = self . _bus . read_word_data ( self . _address , register ) & 0xFFFF
self . _logger . debug ( "Read 0x%04X from register pair 0x%02X, 0x%02X" , result , register , register + 1 )
# Swap bytes if using big endian because read _ word _ data assumes little
# endian on ARM ( little endian ) systems .
if not little_endian :
    result = ( ( result << 8 ) & 0xFF00 ) + ( result >> 8 )
return result


signature:
def get_platform_gpio ( ** keywords ) :
    """Attempt to return a GPIO instance for the platform which the code is being
 executed on . Currently supports only the Raspberry Pi using the RPi . GPIO
 library and Beaglebone Black using the Adafruit _ BBIO library . Will throw an
 exception if a GPIO instance can ' t be created for the current platform . The
 returned GPIO object is an instance of BaseGPIO ."""

implementation:
plat = Platform . platform_detect ( )
if plat == Platform . RASPBERRY_PI :
    import RPi . GPIO
    return RPiGPIOAdapter ( RPi . GPIO , ** keywords )
elif plat == Platform . BEAGLEBONE_BLACK :
    import Adafruit_BBIO . GPIO
    return AdafruitBBIOAdapter ( Adafruit_BBIO . GPIO , ** keywords )
elif plat == Platform . MINNOWBOARD :
    import mraa
    return AdafruitMinnowAdapter ( mraa , ** keywords )
elif plat == Platform . JETSON_NANO :
    import Jetson . GPIO
    return RPiGPIOAdapter ( Jetson . GPIO , ** keywords )
elif plat == Platform . UNKNOWN :
    raise RuntimeError ( 'Could not determine platform.' )


signature:
def output_pins ( self , pins ) :
    """Set multiple pins high or low at once . Pins should be a dict of pin
 name to pin value ( HIGH / True for 1 , LOW / False for 0 ) . All provided pins
 will be set to the given values ."""

implementation:
# General implementation just loops through pins and writes them out
# manually . This is not optimized , but subclasses can choose to implement
# a more optimal batch output implementation . See the MCP230xx class for
# example of optimized implementation .
for pin , value in iter ( pins . items ( ) ) :
    self . output ( pin , value )


signature:
def setup_pins ( self , pins ) :
    """Setup multiple pins as inputs or outputs at once . Pins should be a
 dict of pin name to pin type ( IN or OUT ) ."""

implementation:
# General implementation that can be optimized by derived classes .
for pin , value in iter ( pins . items ( ) ) :
    self . setup ( pin , value )


signature:
def setup ( self , pin , mode , pull_up_down = PUD_OFF ) :
    """Set the input or output mode for a specified pin . Mode should be
 either OUTPUT or INPUT ."""

implementation: self . rpi_gpio . setup ( pin , self . _dir_mapping [ mode ] , pull_up_down = self . _pud_mapping [ pull_up_down ] )


signature:
def add_event_detect ( self , pin , edge , callback = None , bouncetime = - 1 ) :
    """Enable edge detection events for a particular GPIO channel . Pin
 should be type IN . Edge must be RISING , FALLING or BOTH . Callback is a
 function for the event . Bouncetime is switch bounce timeout in ms for
 callback"""

implementation:
kwargs = { }
if callback :
    kwargs [ 'callback' ] = callback
if bouncetime > 0 :
    kwargs [ 'bouncetime' ] = bouncetime
self . rpi_gpio . add_event_detect ( pin , self . _edge_mapping [ edge ] , ** kwargs )


signature:
def wait_for_edge ( self , pin , edge ) :
    """Wait for an edge . Pin should be type IN . Edge must be RISING ,
 FALLING or BOTH ."""

implementation: self . rpi_gpio . wait_for_edge ( pin , self . _edge_mapping [ edge ] )


signature:
def cleanup ( self , pin = None ) :
    """Clean up GPIO event detection for specific pin , or all pins if none
 is specified ."""

implementation:
if pin is None :
    self . rpi_gpio . cleanup ( )
else :
    self . rpi_gpio . cleanup ( pin )


signature:
def add_event_callback ( self , pin , callback , bouncetime = - 1 ) :
    """Add a callback for an event already defined using add _ event _ detect ( ) .
 Pin should be type IN . Bouncetime is switch bounce timeout in ms for
 callback"""

implementation:
kwargs = { }
if bouncetime > 0 :
    kwargs [ 'bouncetime' ] = bouncetime
self . bbio_gpio . add_event_callback ( pin , callback , ** kwargs )


implementation: self . bbio_gpio . wait_for_edge ( pin , self . _edge_mapping [ edge ] )


implementation:
if pin is None :
    self . bbio_gpio . cleanup ( )
else :
    self . bbio_gpio . cleanup ( pin )


signature:
def setup ( self , pin , mode ) :
    """Set the input or output mode for a specified pin . Mode should be
 either DIR _ IN or DIR _ OUT ."""

implementation: self . mraa_gpio . Gpio . dir ( self . mraa_gpio . Gpio ( pin ) , self . _dir_mapping [ mode ] )


signature:
def output ( self , pin , value ) :
    """Set the specified pin the provided high / low value . Value should be
 either 1 ( ON or HIGH ) , or 0 ( OFF or LOW ) or a boolean ."""

implementation: self . mraa_gpio . Gpio . write ( self . mraa_gpio . Gpio ( pin ) , value )


signature:
def input ( self , pin ) :
    """Read the specified pin and return HIGH / true if the pin is pulled high ,
 or LOW / false if pulled low ."""

implementation: return self . mraa_gpio . Gpio . read ( self . mraa_gpio . Gpio ( pin ) )


implementation:
kwargs = { }
if callback :
    kwargs [ 'callback' ] = callback
if bouncetime > 0 :
    kwargs [ 'bouncetime' ] = bouncetime
self . mraa_gpio . Gpio . isr ( self . mraa_gpio . Gpio ( pin ) , self . _edge_mapping [ edge ] , ** kwargs )


signature:
def remove_event_detect ( self , pin ) :
    """Remove edge detection for a particular GPIO channel . Pin should be
 type IN ."""

implementation: self . mraa_gpio . Gpio . isrExit ( self . mraa_gpio . Gpio ( pin ) )


implementation: self . bbio_gpio . wait_for_edge ( self . mraa_gpio . Gpio ( pin ) , self . _edge_mapping [ edge ] )


signature:
def all_info_files ( self ) :
    'Returns a generator of " Path " s'

implementation:
try :
    for info_file in list_files_in_dir ( self . info_dir ) :
        if not os . path . basename ( info_file ) . endswith ( '.trashinfo' ) :
            self . on_non_trashinfo_found ( )
        else :
            yield info_file
except OSError : # when directory does not exist
    pass


signature:
def describe ( path ) :
    """Return a textual description of the file pointed by this path .
 Options :
 - " symbolic link "
 - " directory "
 - " ' . ' directory "
 - " ' . . ' directory "
 - " regular file "
 - " regular empty file "
 - " non existent "
 - " entry " """

implementation:
if os . path . islink ( path ) :
    return 'symbolic link'
elif os . path . isdir ( path ) :
    if path == '.' :
        return 'directory'
    elif path == '..' :
        return 'directory'
    else :
        if os . path . basename ( path ) == '.' :
            return "'.' directory"
        elif os . path . basename ( path ) == '..' :
            return "'..' directory"
        else :
            return 'directory'
elif os . path . isfile ( path ) :
    if os . path . getsize ( path ) == 0 :
        return 'regular empty file'
    else :
        return 'regular file'
elif not os . path . exists ( path ) :
    return 'non existent'
else :
    return 'entry'


signature:
def trash ( self , file ) :
    """Trash a file in the appropriate trash directory .
 If the file belong to the same volume of the trash home directory it
 will be trashed in the home trash directory .
 Otherwise it will be trashed in one of the relevant volume trash
 directories .
 Each volume can have two trash directories , they are
 - $ volume / . Trash / $ uid
 - $ volume / . Trash - $ uid
 Firstly the software attempt to trash the file in the first directory
 then try to trash in the second trash directory ."""

implementation:
if self . _should_skipped_by_specs ( file ) :
    self . reporter . unable_to_trash_dot_entries ( file )
    return
volume_of_file_to_be_trashed = self . volume_of_parent ( file )
self . reporter . volume_of_file ( volume_of_file_to_be_trashed )
candidates = self . _possible_trash_directories_for ( volume_of_file_to_be_trashed )
self . try_trash_file_using_candidates ( file , volume_of_file_to_be_trashed , candidates )


signature:
def persist_trash_info ( self , basename , content , logger ) :
    """Create a . trashinfo file in the $ trash / info directory .
 returns the created TrashInfoFile ."""

implementation:
self . ensure_dir ( self . info_dir , 0o700 )
# write trash info
index = 0
while True :
    if index == 0 :
        suffix = ""
    elif index < 100 :
        suffix = "_%d" % index
    else :
        import random
        suffix = "_%d" % random . randint ( 0 , 65535 )
    base_id = basename
    trash_id = base_id + suffix
    trash_info_basename = trash_id + ".trashinfo"
    dest = os . path . join ( self . info_dir , trash_info_basename )
    try :
        self . atomic_write ( dest , content )
        logger . debug ( ".trashinfo created as %s." % dest )
        return dest
    except OSError :
        logger . debug ( "Attempt for creating %s failed." % dest )
    index += 1
raise IOError ( )


signature:
def get_process_parser ( self , process_id_or_name ) :
    """Returns the ProcessParser for the given process ID or name . It matches
 by name first ."""

implementation:
if process_id_or_name in self . process_parsers_by_name :
    return self . process_parsers_by_name [ process_id_or_name ]
else :
    return self . process_parsers [ process_id_or_name ]


signature:
def add_bpmn_files ( self , filenames ) :
    """Add all filenames in the given list to the parser ' s set ."""

implementation:
for filename in filenames :
    f = open ( filename , 'r' )
    try :
        self . add_bpmn_xml ( ET . parse ( f ) , filename = filename )
    finally :
        f . close ( )


signature:
def add_bpmn_xml ( self , bpmn , svg = None , filename = None ) :
    """Add the given lxml representation of the BPMN file to the parser ' s set .
 : param svg : Optionally , provide the text data for the SVG of the BPMN
 file
 : param filename : Optionally , provide the source filename ."""

implementation:
xpath = xpath_eval ( bpmn )
processes = xpath ( './/bpmn:process' )
for process in processes :
    process_parser = self . PROCESS_PARSER_CLASS ( self , process , svg , filename = filename , doc_xpath = xpath )
    if process_parser . get_id ( ) in self . process_parsers :
        raise ValidationException ( 'Duplicate process ID' , node = process , filename = filename )
    if process_parser . get_name ( ) in self . process_parsers_by_name :
        raise ValidationException ( 'Duplicate process name' , node = process , filename = filename )
    self . process_parsers [ process_parser . get_id ( ) ] = process_parser
    self . process_parsers_by_name [ process_parser . get_name ( ) ] = process_parser


signature:
def one ( nodes , or_none = False ) :
    """Assert that there is exactly one node in the give list , and return it ."""

implementation:
if not nodes and or_none :
    return None
assert len ( nodes ) == 1 , 'Expected 1 result. Received %d results.' % ( len ( nodes ) )
return nodes [ 0 ]


signature:
def xpath_eval ( node , extra_ns = None ) :
    """Returns an XPathEvaluator , with namespace prefixes ' bpmn ' for
 http : / / www . omg . org / spec / BPMN / 20100524 / MODEL , and additional specified ones"""

implementation:
namespaces = { 'bpmn' : BPMN_MODEL_NS }
if extra_ns :
    namespaces . update ( extra_ns )
return lambda path : node . findall ( path , namespaces )


signature:
def serialize_attrib ( self , op ) :
    """Serializer for : meth : ` SpiffWorkflow . operators . Attrib ` .
 Example : :
 < attribute > foobar < / attribute >"""

implementation:
elem = etree . Element ( 'attribute' )
elem . text = op . name
return elem


signature:
def serialize_pathattrib ( self , op ) :
    """Serializer for : meth : ` SpiffWorkflow . operators . PathAttrib ` .
 Example : :
 < path > foobar < / path >"""

implementation:
elem = etree . Element ( 'path' )
elem . text = op . path
return elem


signature:
def serialize_assign ( self , op ) :
    """Serializer for : meth : ` SpiffWorkflow . operators . Assign ` .
 Example : :
 < assign >
 < name > foobar < / name >
 < value > doodle < / value >
 < / assign >"""

implementation:
elem = etree . Element ( 'assign' )
self . serialize_value ( SubElement ( elem , 'name' ) , op . left_attribute )
if op . right :
    self . serialize_value ( SubElement ( elem , 'value' ) , op . right )
if op . right_attribute :
    self . serialize_value ( SubElement ( elem , 'value-attribute' ) , op . right_attribute )
return elem


signature:
def serialize_value ( self , parent_elem , value ) :
    """Serializes str , Attrib , or PathAttrib objects .
 Example : :
 < attribute > foobar < / attribute >"""

implementation:
if isinstance ( value , ( str , int ) ) or type ( value ) . __name__ == 'str' :
    parent_elem . text = str ( value )
elif value is None :
    parent_elem . text = None
else :
    parent_elem . append ( value . serialize ( self ) )


signature:
def serialize_value_map ( self , map_elem , thedict ) :
    """Serializes a dictionary of key / value pairs , where the values are
 either strings , or Attrib , or PathAttrib objects .
 Example : :
 < variable >
 < name > foo < / name >
 < value > text < / value >
 < / variable >
 < variable >
 < name > foo2 < / name >
 < value > < attribute > foobar < / attribute > < / value >
 < / variable >"""

implementation:
for key , value in sorted ( ( str ( k ) , v ) for ( k , v ) in thedict . items ( ) ) :
    var_elem = SubElement ( map_elem , 'variable' )
    SubElement ( var_elem , 'name' ) . text = str ( key )
    value_elem = SubElement ( var_elem , 'value' )
    self . serialize_value ( value_elem , value )
return map_elem


signature:
def serialize_value_list ( self , list_elem , thelist ) :
    """Serializes a list , where the values are objects of type
 str , Attrib , or PathAttrib .
 Example : :
 < value > text < / value >
 < value > < attribute > foobar < / attribute > < / value >
 < value > < path > foobar < / path > < / value >"""

implementation:
for value in thelist :
    value_elem = SubElement ( list_elem , 'value' )
    self . serialize_value ( value_elem , value )
return list_elem


signature:
def serialize_operator_equal ( self , op ) :
    """Serializer for : meth : ` SpiffWorkflow . operators . Equal ` .
 Example : :
 < equals >
 < value > text < / value >
 < value > < attribute > foobar < / attribute > < / value >
 < value > < path > foobar < / path > < / value >
 < / equals >"""

implementation:
elem = etree . Element ( 'equals' )
return self . serialize_value_list ( elem , op . args )


signature:
def serialize_operator_not_equal ( self , op ) :
    """Serializer for : meth : ` SpiffWorkflow . operators . NotEqual ` .
 Example : :
 < not - equals >
 < value > text < / value >
 < value > < attribute > foobar < / attribute > < / value >
 < value > < path > foobar < / path > < / value >
 < / not - equals >"""

implementation:
elem = etree . Element ( 'not-equals' )
return self . serialize_value_list ( elem , op . args )


signature:
def serialize_operator_greater_than ( self , op ) :
    """Serializer for : meth : ` SpiffWorkflow . operators . NotEqual ` .
 Example : :
 < greater - than >
 < value > text < / value >
 < value > < attribute > foobar < / attribute > < / value >
 < / greater - than >"""

implementation:
elem = etree . Element ( 'greater-than' )
return self . serialize_value_list ( elem , op . args )


signature:
def serialize_operator_less_than ( self , op ) :
    """Serializer for : meth : ` SpiffWorkflow . operators . NotEqual ` .
 Example : :
 < less - than >
 < value > text < / value >
 < value > < attribute > foobar < / attribute > < / value >
 < / less - than >"""

implementation:
elem = etree . Element ( 'less-than' )
return self . serialize_value_list ( elem , op . args )


signature:
def serialize_operator_match ( self , op ) :
    """Serializer for : meth : ` SpiffWorkflow . operators . NotEqual ` .
 Example : :
 < matches >
 < value > text < / value >
 < value > < attribute > foobar < / attribute > < / value >
 < / matches >"""

implementation:
elem = etree . Element ( 'matches' )
return self . serialize_value_list ( elem , op . args )


signature:
def serialize_task_spec ( self , spec , elem ) :
    """Serializes common attributes of : meth : ` SpiffWorkflow . specs . TaskSpec ` ."""

implementation:
if spec . id is not None :
    SubElement ( elem , 'id' ) . text = str ( spec . id )
SubElement ( elem , 'name' ) . text = spec . name
if spec . description :
    SubElement ( elem , 'description' ) . text = spec . description
if spec . manual :
    SubElement ( elem , 'manual' )
if spec . internal :
    SubElement ( elem , 'internal' )
SubElement ( elem , 'lookahead' ) . text = str ( spec . lookahead )
inputs = [ t . name for t in spec . inputs ]
outputs = [ t . name for t in spec . outputs ]
self . serialize_value_list ( SubElement ( elem , 'inputs' ) , inputs )
self . serialize_value_list ( SubElement ( elem , 'outputs' ) , outputs )
self . serialize_value_map ( SubElement ( elem , 'data' ) , spec . data )
self . serialize_value_map ( SubElement ( elem , 'defines' ) , spec . defines )
self . serialize_value_list ( SubElement ( elem , 'pre-assign' ) , spec . pre_assign )
self . serialize_value_list ( SubElement ( elem , 'post-assign' ) , spec . post_assign )
# Note : Events are not serialized ; this is documented in
# the TaskSpec API docs .
return elem


signature:
def serialize_acquire_mutex ( self , spec ) :
    """Serializer for : meth : ` SpiffWorkflow . specs . AcquireMutex ` ."""

implementation:
elem = etree . Element ( 'acquire-mutex' )
self . serialize_task_spec ( spec , elem )
SubElement ( elem , 'mutex' ) . text = spec . mutex
return elem


signature:
def get_event_definition ( self ) :
    """Parse the event definition node , and return an instance of Event"""

implementation:
messageEventDefinition = first ( self . xpath ( './/bpmn:messageEventDefinition' ) )
if messageEventDefinition is not None :
    return self . get_message_event_definition ( messageEventDefinition )
timerEventDefinition = first ( self . xpath ( './/bpmn:timerEventDefinition' ) )
if timerEventDefinition is not None :
    return self . get_timer_event_definition ( timerEventDefinition )
raise NotImplementedError ( 'Unsupported Intermediate Catch Event: %r' , ET . tostring ( self . node ) )


signature:
def get_message_event_definition ( self , messageEventDefinition ) :
    """Parse the messageEventDefinition node and return an instance of
 MessageEventDefinition"""

implementation:
messageRef = first ( self . xpath ( './/bpmn:messageRef' ) )
message = messageRef . get ( 'name' ) if messageRef is not None else self . node . get ( 'name' )
return MessageEventDefinition ( message )


signature:
def get_timer_event_definition ( self , timerEventDefinition ) :
    """Parse the timerEventDefinition node and return an instance of
 TimerEventDefinition
 This currently only supports the timeDate node for specifying an expiry
 time for the timer ."""

implementation:
timeDate = first ( self . xpath ( './/bpmn:timeDate' ) )
return TimerEventDefinition ( self . node . get ( 'name' , timeDate . text ) , self . parser . parse_condition ( timeDate . text , None , None , None , None , self ) )


signature:
def get_all_lanes ( self ) :
    """Returns a set of the distinct lane names used in the process ( including
 called activities )"""

implementation:
done = set ( )
lanes = set ( )
def recursive_find ( task_spec ) :
    if task_spec in done :
        return
    done . add ( task_spec )
    if hasattr ( task_spec , 'lane' ) and task_spec . lane :
        lanes . add ( task_spec . lane )
    if hasattr ( task_spec , 'spec' ) :
        recursive_find ( task_spec . spec . start )
    for t in task_spec . outputs :
        recursive_find ( t )
recursive_find ( self . start )
return lanes


signature:
def get_specs_depth_first ( self ) :
    """Get the specs for all processes ( including called ones ) , in depth first
 order ."""

implementation:
done = set ( )
specs = [ self ]
def recursive_find ( task_spec ) :
    if task_spec in done :
        return
    done . add ( task_spec )
    if hasattr ( task_spec , 'spec' ) :
        specs . append ( task_spec . spec )
        recursive_find ( task_spec . spec . start )
    for t in task_spec . outputs :
        recursive_find ( t )
recursive_find ( self . start )
return specs


signature:
def to_html_string ( self ) :
    """Returns an etree HTML node with a document describing the process . This
 is only supported if the editor provided an SVG representation ."""

implementation:
html = ET . Element ( 'html' )
head = ET . SubElement ( html , 'head' )
title = ET . SubElement ( head , 'title' )
title . text = self . description
body = ET . SubElement ( html , 'body' )
h1 = ET . SubElement ( body , 'h1' )
h1 . text = self . description
span = ET . SubElement ( body , 'span' )
span . text = '___CONTENT___'
html_text = ET . tostring ( html )
svg_content = ''
svg_done = set ( )
for spec in self . get_specs_depth_first ( ) :
    if spec . svg and spec . svg not in svg_done :
        svg_content += '<p>' + spec . svg + "</p>"
        svg_done . add ( spec . svg )
return html_text . replace ( '___CONTENT___' , svg_content )


signature:
def connect ( self , callback , * args , ** kwargs ) :
    """Connects the event with the given callback .
 When the signal is emitted , the callback is invoked .
 . . note : :
 The signal handler is stored with a hard reference , so you
 need to make sure to call : class : ` disconnect ( ) ` if you want the
 handler
 to be garbage collected .
 : type callback : object
 : param callback : The callback function .
 : type args : tuple
 : param args : Optional arguments passed to the callback .
 : type kwargs : dict
 : param kwargs : Optional keyword arguments passed to the callback ."""

implementation:
if self . is_connected ( callback ) :
    raise AttributeError ( 'callback is already connected' )
if self . hard_subscribers is None :
    self . hard_subscribers = [ ]
self . hard_subscribers . append ( ( callback , args , kwargs ) )


signature:
def listen ( self , callback , * args , ** kwargs ) :
    """Like : class : ` connect ( ) ` , but uses a weak reference instead of a
 normal reference .
 The signal is automatically disconnected as soon as the handler
 is garbage collected .
 . . note : :
 Storing signal handlers as weak references means that if
 your handler is a local function , it may be garbage collected . To
 prevent this , use : class : ` connect ( ) ` instead .
 : type callback : object
 : param callback : The callback function .
 : type args : tuple
 : param args : Optional arguments passed to the callback .
 : type kwargs : dict
 : param kwargs : Optional keyword arguments passed to the callback .
 : rtype : : class : ` Exscript . util . weakmethod . WeakMethod `
 : returns : The newly created weak reference to the callback ."""

implementation:
if self . lock is None :
    self . lock = Lock ( )
with self . lock :
    if self . is_connected ( callback ) :
        raise AttributeError ( 'callback is already connected' )
    if self . weak_subscribers is None :
        self . weak_subscribers = [ ]
    ref = weakmethod . ref ( callback , self . _try_disconnect )
    self . weak_subscribers . append ( ( ref , args , kwargs ) )
return ref


signature:
def n_subscribers ( self ) :
    """Returns the number of connected subscribers .
 : rtype : int
 : returns : The number of subscribers ."""

implementation:
hard = self . hard_subscribers and len ( self . hard_subscribers ) or 0
weak = self . weak_subscribers and len ( self . weak_subscribers ) or 0
return hard + weak


signature:
def is_connected ( self , callback ) :
    """Returns True if the event is connected to the given function .
 : type callback : object
 : param callback : The callback function .
 : rtype : bool
 : returns : Whether the signal is connected to the given function ."""

implementation:
index = self . _weakly_connected_index ( callback )
if index is not None :
    return True
if self . hard_subscribers is None :
    return False
return callback in self . _hard_callbacks ( )


signature:
def emit ( self , * args , ** kwargs ) :
    """Emits the signal , passing the given arguments to the callbacks .
 If one of the callbacks returns a value other than None , no further
 callbacks are invoked and the return value of the callback is
 returned to the caller of emit ( ) .
 : type args : tuple
 : param args : Optional arguments passed to the callbacks .
 : type kwargs : dict
 : param kwargs : Optional keyword arguments passed to the callbacks .
 : rtype : object
 : returns : Returns None if all callbacks returned None . Returns
 the return value of the last invoked callback otherwise ."""

implementation:
if self . hard_subscribers is not None :
    for callback , user_args , user_kwargs in self . hard_subscribers :
        kwargs . update ( user_kwargs )
        result = callback ( * args + user_args , ** kwargs )
        if result is not None :
            return result
if self . weak_subscribers is not None :
    for callback , user_args , user_kwargs in self . weak_subscribers :
        kwargs . update ( user_kwargs )
        # Even though WeakMethod notifies us when the underlying
        # function is destroyed , and we remove the item from the
        # the list of subscribers , there is no guarantee that
        # this notification has already happened because the garbage
        # collector may run while this loop is executed .
        # Disabling the garbage collector temporarily also does
        # not work , because other threads may be trying to do
        # the same , causing yet another race condition .
        # So the only solution is to skip such functions .
        function = callback . get_function ( )
        if function is None :
            continue
        result = function ( * args + user_args , ** kwargs )
        if result is not None :
            return result


signature:
def _try_disconnect ( self , ref ) :
    """Called by the weak reference when its target dies .
 In other words , we can assert that self . weak _ subscribers is not
 None at this time ."""

implementation:
with self . lock :
    weak = [ s [ 0 ] for s in self . weak_subscribers ]
    try :
        index = weak . index ( ref )
    except ValueError : # subscriber was already removed by a call to disconnect ( )
        pass
    else :
        self . weak_subscribers . pop ( index )


signature:
def disconnect ( self , callback ) :
    """Disconnects the signal from the given function .
 : type callback : object
 : param callback : The callback function ."""

implementation:
if self . weak_subscribers is not None :
    with self . lock :
        index = self . _weakly_connected_index ( callback )
        if index is not None :
            self . weak_subscribers . pop ( index ) [ 0 ]
if self . hard_subscribers is not None :
    try :
        index = self . _hard_callbacks ( ) . index ( callback )
    except ValueError :
        pass
    else :
        self . hard_subscribers . pop ( index )


signature:
def deserialize_workflow_spec ( self , s_state , filename = None ) :
    """: param s _ state : a byte - string with the contents of the packaged
 workflow archive , or a file - like object .
 : param filename : the name of the package file ."""

implementation:
if isinstance ( s_state , ( str , bytes ) ) :
    s_state = BytesIO ( s_state )
package_zip = zipfile . ZipFile ( s_state , "r" , compression = zipfile . ZIP_DEFLATED )
config = configparser . ConfigParser ( )
ini_fp = TextIOWrapper ( package_zip . open ( Packager . METADATA_FILE ) , encoding = "UTF-8" )
try :
    config . read_file ( ini_fp )
finally :
    ini_fp . close ( )
parser_class = BpmnParser
try :
    parser_class_module = config . get ( 'MetaData' , 'parser_class_module' , fallback = None )
except TypeError : # unfortunately the fallback = does not exist on python 2
    parser_class_module = config . get ( 'MetaData' , 'parser_class_module' , None )
if parser_class_module :
    mod = __import__ ( parser_class_module , fromlist = [ config . get ( 'MetaData' , 'parser_class' ) ] )
    parser_class = getattr ( mod , config . get ( 'MetaData' , 'parser_class' ) )
parser = parser_class ( )
for info in package_zip . infolist ( ) :
    parts = os . path . split ( info . filename )
    if ( len ( parts ) == 2 and not parts [ 0 ] and parts [ 1 ] . lower ( ) . endswith ( '.bpmn' ) ) : # It is in the root of the ZIP and is a BPMN file
        try :
            svg = package_zip . read ( info . filename [ : - 5 ] + '.svg' )
        except KeyError :
            svg = None
        bpmn_fp = package_zip . open ( info )
        try :
            bpmn = ET . parse ( bpmn_fp )
        finally :
            bpmn_fp . close ( )
        parser . add_bpmn_xml ( bpmn , svg = svg , filename = '%s:%s' % ( filename , info . filename ) )
return parser . get_spec ( config . get ( 'MetaData' , 'entry_point_process' ) )


signature:
def parse_node ( self ) :
    """Parse this node , and all children , returning the connected task spec ."""

implementation:
try :
    self . task = self . create_task ( )
    self . task . documentation = self . parser . _parse_documentation ( self . node , xpath = self . xpath , task_parser = self )
    boundary_event_nodes = self . process_xpath ( './/bpmn:boundaryEvent[@attachedToRef="%s"]' % self . get_id ( ) )
    if boundary_event_nodes :
        parent_task = _BoundaryEventParent ( self . spec , '%s.BoundaryEventParent' % self . get_id ( ) , self . task , lane = self . task . lane )
        self . process_parser . parsed_nodes [ self . node . get ( 'id' ) ] = parent_task
        parent_task . connect_outgoing ( self . task , '%s.FromBoundaryEventParent' % self . get_id ( ) , None , None )
        for boundary_event in boundary_event_nodes :
            b = self . process_parser . parse_node ( boundary_event )
            parent_task . connect_outgoing ( b , '%s.FromBoundaryEventParent' % boundary_event . get ( 'id' ) , None , None )
    else :
        self . process_parser . parsed_nodes [ self . node . get ( 'id' ) ] = self . task
    children = [ ]
    outgoing = self . process_xpath ( './/bpmn:sequenceFlow[@sourceRef="%s"]' % self . get_id ( ) )
    if len ( outgoing ) > 1 and not self . handles_multiple_outgoing ( ) :
        raise ValidationException ( 'Multiple outgoing flows are not supported for ' 'tasks of type' , node = self . node , filename = self . process_parser . filename )
    for sequence_flow in outgoing :
        target_ref = sequence_flow . get ( 'targetRef' )
        target_node = one ( self . process_xpath ( './/*[@id="%s"]' % target_ref ) )
        c = self . process_parser . parse_node ( target_node )
        children . append ( ( c , target_node , sequence_flow ) )
    if children :
        default_outgoing = self . node . get ( 'default' )
        if not default_outgoing :
            ( c , target_node , sequence_flow ) = children [ 0 ]
            default_outgoing = sequence_flow . get ( 'id' )
        for ( c , target_node , sequence_flow ) in children :
            self . connect_outgoing ( c , target_node , sequence_flow , sequence_flow . get ( 'id' ) == default_outgoing )
    return parent_task if boundary_event_nodes else self . task
except ValidationException :
    raise
except Exception as ex :
    exc_info = sys . exc_info ( )
    tb = "" . join ( traceback . format_exception ( exc_info [ 0 ] , exc_info [ 1 ] , exc_info [ 2 ] ) )
    LOG . error ( "%r\n%s" , ex , tb )
    raise ValidationException ( "%r" % ( ex ) , node = self . node , filename = self . process_parser . filename )


signature:
def create_task ( self ) :
    """Create an instance of the task appropriately . A subclass can override
 this method to get extra information from the node ."""

implementation: return self . spec_class ( self . spec , self . get_task_spec_name ( ) , lane = self . get_lane ( ) , description = self . node . get ( 'name' , None ) )


signature:
def connect_outgoing ( self , outgoing_task , outgoing_task_node , sequence_flow_node , is_default ) :
    """Connects this task to the indicating outgoing task , with the details in
 the sequence flow . A subclass can override this method to get extra
 information from the node ."""

implementation: self . task . connect_outgoing ( outgoing_task , sequence_flow_node . get ( 'id' ) , sequence_flow_node . get ( 'name' , None ) , self . parser . _parse_documentation ( sequence_flow_node , task_parser = self ) )


signature:
def connect_outgoing ( self , taskspec , sequence_flow_id , sequence_flow_name , documentation ) :
    """Connect this task spec to the indicated child .
 : param sequence _ flow _ id : The ID of the connecting sequenceFlow node .
 : param sequence _ flow _ name : The name of the connecting sequenceFlow
 node ."""

implementation:
self . connect ( taskspec )
s = SequenceFlow ( sequence_flow_id , sequence_flow_name , documentation , taskspec )
self . outgoing_sequence_flows [ taskspec . name ] = s
self . outgoing_sequence_flows_by_id [ sequence_flow_id ] = s


signature:
def connect_outgoing_if ( self , condition , taskspec , sequence_flow_id , sequence_flow_name , documentation ) :
    """Connect this task spec to the indicated child , if the condition
 evaluates to true . This should only be called if the task has a
 connect _ if method ( e . g . ExclusiveGateway ) .
 : param sequence _ flow _ id : The ID of the connecting sequenceFlow node .
 : param sequence _ flow _ name : The name of the connecting sequenceFlow
 node ."""

implementation:
self . connect_if ( _BpmnCondition ( condition ) , taskspec )
s = SequenceFlow ( sequence_flow_id , sequence_flow_name , documentation , taskspec )
self . outgoing_sequence_flows [ taskspec . name ] = s
self . outgoing_sequence_flows_by_id [ sequence_flow_id ] = s


signature:
def get_outgoing_sequence_names ( self ) :
    """Returns a list of the names of outgoing sequences . Some may be None ."""

implementation: return sorted ( [ s . name for s in list ( self . outgoing_sequence_flows_by_id . values ( ) ) ] )


signature:
def connect_if ( self , condition , task_spec ) :
    """Connects a taskspec that is executed if the condition DOES match .
 condition - - a condition ( Condition )
 taskspec - - the conditional task spec"""

implementation:
assert task_spec is not None
self . outputs . append ( task_spec )
self . cond_task_specs . append ( ( condition , task_spec . name ) )
task_spec . _connect_notify ( self )


signature:
def _on_complete_hook ( self , my_task ) :
    """Runs the task . Should not be called directly .
 Returns True if completed , False otherwise ."""

implementation:
# Find all matching conditions .
outputs = [ ]
for condition , output in self . cond_task_specs :
    if self . choice is not None and output not in self . choice :
        continue
    if condition is None :
        outputs . append ( self . _wf_spec . get_task_spec_from_name ( output ) )
        continue
    if not condition . _matches ( my_task ) :
        continue
    outputs . append ( self . _wf_spec . get_task_spec_from_name ( output ) )
my_task . _sync_children ( outputs , Task . FUTURE )
for child in my_task . children :
    child . task_spec . _update ( child )


signature:
def is_completed ( self ) :
    """Returns True if the entire Workflow is completed , False otherwise .
 : rtype : bool
 : return : Whether the workflow is completed ."""

implementation:
mask = Task . NOT_FINISHED_MASK
iter = Task . Iterator ( self . task_tree , mask )
try :
    next ( iter )
except StopIteration : # No waiting tasks found .
    return True
return False


signature:
def cancel ( self , success = False ) :
    """Cancels all open tasks in the workflow .
 : type success : bool
 : param success : Whether the Workflow should be marked as successfully
 completed ."""

implementation:
self . success = success
cancel = [ ]
mask = Task . NOT_FINISHED_MASK
for task in Task . Iterator ( self . task_tree , mask ) :
    cancel . append ( task )
for task in cancel :
    task . cancel ( )


signature:
def get_task ( self , id ) :
    """Returns the task with the given id .
 : type id : integer
 : param id : The id of a task .
 : rtype : Task
 : returns : The task with the given id ."""

implementation:
tasks = [ task for task in self . get_tasks ( ) if task . id == id ]
return tasks [ 0 ] if len ( tasks ) == 1 else None


signature:
def get_tasks_from_spec_name ( self , name ) :
    """Returns all tasks whose spec has the given name .
 : type name : str
 : param name : The name of a task spec .
 : rtype : Task
 : return : The task that relates to the spec with the given name ."""

implementation: return [ task for task in self . get_tasks ( ) if task . task_spec . name == name ]


signature:
def get_tasks ( self , state = Task . ANY_MASK ) :
    """Returns a list of Task objects with the given state .
 : type state : integer
 : param state : A bitmask of states .
 : rtype : list [ Task ]
 : returns : A list of tasks ."""

implementation: return [ t for t in Task . Iterator ( self . task_tree , state ) ]


signature:
def complete_task_from_id ( self , task_id ) :
    """Runs the task with the given id .
 : type task _ id : integer
 : param task _ id : The id of the Task object ."""

implementation:
if task_id is None :
    raise WorkflowException ( self . spec , 'task_id is None' )
for task in self . task_tree :
    if task . id == task_id :
        return task . complete ( )
msg = 'A task with the given task_id (%s) was not found' % task_id
raise WorkflowException ( self . spec , msg )


signature:
def complete_next ( self , pick_up = True , halt_on_manual = True ) :
    """Runs the next task .
 Returns True if completed , False otherwise .
 : type pick _ up : bool
 : param pick _ up : When True , this method attempts to choose the next
 task not by searching beginning at the root , but by
 searching from the position at which the last call
 of complete _ next ( ) left off .
 : type halt _ on _ manual : bool
 : param halt _ on _ manual : When True , this method will not attempt to
 complete any tasks that have manual = True .
 See : meth : ` SpiffWorkflow . specs . TaskSpec . _ _ init _ _ `
 : rtype : bool
 : returns : True if all tasks were completed , False otherwise ."""

implementation:
# Try to pick up where we left off .
blacklist = [ ]
if pick_up and self . last_task is not None :
    try :
        iter = Task . Iterator ( self . last_task , Task . READY )
        task = next ( iter )
    except StopIteration :
        task = None
    self . last_task = None
    if task is not None :
        if not ( halt_on_manual and task . task_spec . manual ) :
            if task . complete ( ) :
                self . last_task = task
                return True
        blacklist . append ( task )
# Walk through all ready tasks .
for task in Task . Iterator ( self . task_tree , Task . READY ) :
    for blacklisted_task in blacklist :
        if task . _is_descendant_of ( blacklisted_task ) :
            continue
    if not ( halt_on_manual and task . task_spec . manual ) :
        if task . complete ( ) :
            self . last_task = task
            return True
    blacklist . append ( task )
# Walk through all waiting tasks .
for task in Task . Iterator ( self . task_tree , Task . WAITING ) :
    task . task_spec . _update ( task )
    if not task . _has_state ( Task . WAITING ) :
        self . last_task = task
        return True
return False


signature:
def complete_all ( self , pick_up = True , halt_on_manual = True ) :
    """Runs all branches until completion . This is a convenience wrapper
 around : meth : ` complete _ next ` , and the pick _ up argument is passed
 along .
 : type pick _ up : bool
 : param pick _ up : Passed on to each call of complete _ next ( ) .
 : type halt _ on _ manual : bool
 : param halt _ on _ manual : When True , this method will not attempt to
 complete any tasks that have manual = True .
 See : meth : ` SpiffWorkflow . specs . TaskSpec . _ _ init _ _ `"""

implementation:
while self . complete_next ( pick_up , halt_on_manual ) :
    pass


signature:
def ref ( function , callback = None ) :
    """Returns a weak reference to the given method or function .
 If the callback argument is not None , it is called as soon
 as the referenced function is garbage deleted .
 : type function : callable
 : param function : The function to reference .
 : type callback : callable
 : param callback : Called when the function dies ."""

implementation:
try :
    function . __func__
except AttributeError :
    return _WeakMethodFree ( function , callback )
return _WeakMethodBound ( function , callback )


signature:
def serialize_workflow ( self , workflow , include_spec = False , ** kwargs ) :
    """: param workflow : the workflow instance to serialize
 : param include _ spec : Always set to False ( The CompactWorkflowSerializer
 only supports workflow serialization )"""

implementation:
if include_spec :
    raise NotImplementedError ( 'Including the spec serialization with the workflow state ' 'is not implemented.' )
return self . _get_workflow_state ( workflow )


signature:
def deserialize_workflow ( self , s_state , workflow_spec = None , read_only = False , ** kwargs ) :
    """: param s _ state : the state of the workflow as returned by
 serialize _ workflow
 : param workflow _ spec : the Workflow Spec of the workflow
 ( CompactWorkflowSerializer only supports workflow serialization )
 : param read _ only : ( Optional ) True if the workflow should be restored in
 READ ONLY mode
 NB : Additional kwargs passed to the deserialize _ workflow method will be
 passed to the new _ workflow method ."""

implementation:
if workflow_spec is None :
    raise NotImplementedError ( 'Including the spec serialization with the workflow state is ' ' not implemented. A \'workflow_spec\' must ' 'be provided.' )
workflow = self . new_workflow ( workflow_spec , read_only = read_only , ** kwargs )
self . _restore_workflow_state ( workflow , s_state )
return workflow


signature:
def new_workflow ( self , workflow_spec , read_only = False , ** kwargs ) :
    """Create a new workflow instance from the given spec and arguments .
 : param workflow _ spec : the workflow spec to use
 : param read _ only : this should be in read only mode
 : param kwargs : Any extra kwargs passed to the deserialize _ workflow
 method will be passed through here"""

implementation: return BpmnWorkflow ( workflow_spec , read_only = read_only , ** kwargs )


signature:
def _start ( self , my_task , force = False ) :
    """Returns False when successfully fired , True otherwise"""

implementation:
if ( not hasattr ( my_task , 'subprocess' ) ) or my_task . subprocess is None :
    my_task . subprocess = subprocess . Popen ( self . args , stderr = subprocess . STDOUT , stdout = subprocess . PIPE )
if my_task . subprocess :
    my_task . subprocess . poll ( )
    if my_task . subprocess . returncode is None : # Still waiting
        return False
    else :
        results = my_task . subprocess . communicate ( )
        my_task . results = results
        return True
return False


signature:
def _setstate ( self , value , force = False ) :
    """Setting force to True allows for changing a state after it
 COMPLETED . This would otherwise be invalid ."""

implementation:
if self . _state == value :
    return
if value < self . _state and not force :
    raise WorkflowException ( self . task_spec , 'state went from %s to %s!' % ( self . get_state_name ( ) , self . state_names [ value ] ) )
if __debug__ :
    old = self . get_state_name ( )
self . _state = value
if __debug__ :
    self . log . append ( "Moving '%s' from %s to %s" % ( self . get_name ( ) , old , self . get_state_name ( ) ) )
self . state_history . append ( value )
LOG . debug ( "Moving '%s' (spec=%s) from %s to %s" % ( self . get_name ( ) , self . task_spec . name , old , self . get_state_name ( ) ) )


signature:
def _set_state ( self , state , force = True ) :
    """Setting force to True allows for changing a state after it
 COMPLETED . This would otherwise be invalid ."""

implementation:
self . _setstate ( state , True )
self . last_state_change = time . time ( )


signature:
def _add_child ( self , task_spec , state = MAYBE ) :
    """Adds a new child and assigns the given TaskSpec to it .
 : type task _ spec : TaskSpec
 : param task _ spec : The task spec that is assigned to the new child .
 : type state : integer
 : param state : The bitmask of states for the new child .
 : rtype : Task
 : returns : The new child task ."""

implementation:
if task_spec is None :
    raise ValueError ( self , '_add_child() requires a TaskSpec' )
if self . _is_predicted ( ) and state & self . PREDICTED_MASK == 0 :
    msg = 'Attempt to add non-predicted child to predicted task'
    raise WorkflowException ( self . task_spec , msg )
task = Task ( self . workflow , task_spec , self , state = state )
task . thread_id = self . thread_id
if state == self . READY :
    task . _ready ( )
return task


signature:
def _assign_new_thread_id ( self , recursive = True ) :
    """Assigns a new thread id to the task .
 : type recursive : bool
 : param recursive : Whether to assign the id to children recursively .
 : rtype : bool
 : returns : The new thread id ."""

implementation:
self . __class__ . thread_id_pool += 1
self . thread_id = self . __class__ . thread_id_pool
if not recursive :
    return self . thread_id
for child in self :
    child . thread_id = self . thread_id
return self . thread_id


signature:
def _sync_children ( self , task_specs , state = MAYBE ) :
    """This method syncs up the task ' s children with the given list of task
 specs . In other words : :
 - Add one child for each given TaskSpec , unless that child already
 exists .
 - Remove all children for which there is no spec in the given list ,
 unless it is a " triggered " task .
 . . note : :
 It is an error if the task has a non - predicted child that is
 not given in the TaskSpecs .
 : type task _ specs : list ( TaskSpec )
 : param task _ specs : The list of task specs that may become children .
 : type state : integer
 : param state : The bitmask of states for the new children ."""

implementation:
LOG . debug ( "Updating children for %s" % self . get_name ( ) )
if task_specs is None :
    raise ValueError ( '"task_specs" argument is None' )
add = task_specs [ : ]
# Create a list of all children that are no longer needed .
remove = [ ]
for child in self . children : # Triggered tasks are never removed .
    if child . triggered :
        continue
    # Check whether the task needs to be removed .
    if child . task_spec in add :
        add . remove ( child . task_spec )
        continue
    # Non - predicted tasks must not be removed , so they HAVE to be in
    # the given task spec list .
    if child . _is_definite ( ) :
        raise WorkflowException ( self . task_spec , 'removal of non-predicted child %s' % repr ( child ) )
    remove . append ( child )
# Remove and add the children accordingly .
for child in remove :
    self . children . remove ( child )
for task_spec in add :
    self . _add_child ( task_spec , state )


signature:
def _is_descendant_of ( self , parent ) :
    """Returns True if parent is in the list of ancestors , returns False
 otherwise .
 : type parent : Task
 : param parent : The parent that is searched in the ancestors .
 : rtype : bool
 : returns : Whether the parent was found ."""

implementation:
if self . parent is None :
    return False
if self . parent == parent :
    return True
return self . parent . _is_descendant_of ( parent )


signature:
def _find_child_of ( self , parent_task_spec ) :
    """Returns the ancestor that has a task with the given task spec
 as a parent .
 If no such ancestor was found , the root task is returned .
 : type parent _ task _ spec : TaskSpec
 : param parent _ task _ spec : The wanted ancestor .
 : rtype : Task
 : returns : The child of the given ancestor ."""

implementation:
if self . parent is None :
    return self
if self . parent . task_spec == parent_task_spec :
    return self
return self . parent . _find_child_of ( parent_task_spec )


signature:
def _find_any ( self , task_spec ) :
    """Returns any descendants that have the given task spec assigned .
 : type task _ spec : TaskSpec
 : param task _ spec : The wanted task spec .
 : rtype : list ( Task )
 : returns : The tasks objects that are attached to the given task spec ."""

implementation:
tasks = [ ]
if self . task_spec == task_spec :
    tasks . append ( self )
for child in self :
    if child . task_spec != task_spec :
        continue
    tasks . append ( child )
return tasks


signature:
def _find_ancestor ( self , task_spec ) :
    """Returns the ancestor that has the given task spec assigned .
 If no such ancestor was found , the root task is returned .
 : type task _ spec : TaskSpec
 : param task _ spec : The wanted task spec .
 : rtype : Task
 : returns : The ancestor ."""

implementation:
if self . parent is None :
    return self
if self . parent . task_spec == task_spec :
    return self . parent
return self . parent . _find_ancestor ( task_spec )


signature:
def _find_ancestor_from_name ( self , name ) :
    """Returns the ancestor that has a task with the given name assigned .
 Returns None if no such ancestor was found .
 : type name : str
 : param name : The name of the wanted task .
 : rtype : Task
 : returns : The ancestor ."""

implementation:
if self . parent is None :
    return None
if self . parent . get_name ( ) == name :
    return self . parent
return self . parent . _find_ancestor_from_name ( name )


signature:
def _ready ( self ) :
    """Marks the task as ready for execution ."""

implementation:
if self . _has_state ( self . COMPLETED ) or self . _has_state ( self . CANCELLED ) :
    return
self . _set_state ( self . READY )
self . task_spec . _on_ready ( self )


signature:
def get_state_name ( self ) :
    """Returns a textual representation of this Task ' s state ."""

implementation:
state_name = [ ]
for state , name in list ( self . state_names . items ( ) ) :
    if self . _has_state ( state ) :
        state_name . append ( name )
return '|' . join ( state_name )


signature:
def _inherit_data ( self ) :
    """Inherits the data from the parent ."""

implementation:
LOG . debug ( "'%s' inheriting data from '%s'" % ( self . get_name ( ) , self . parent . get_name ( ) ) , extra = dict ( data = self . parent . data ) )
self . set_data ( ** self . parent . data )


signature:
def cancel ( self ) :
    """Cancels the item if it was not yet completed , and removes
 any children that are LIKELY ."""

implementation:
if self . _is_finished ( ) :
    for child in self . children :
        child . cancel ( )
    return
self . _set_state ( self . CANCELLED )
self . _drop_children ( )
self . task_spec . _on_cancel ( self )


signature:
def complete ( self ) :
    """Called by the associated task to let us know that its state
 has changed ( e . g . from FUTURE to COMPLETED . )"""

implementation:
self . _set_state ( self . COMPLETED )
return self . task_spec . _on_complete ( self )


signature:
def get_dump ( self , indent = 0 , recursive = True ) :
    """Returns the subtree as a string for debugging .
 : rtype : str
 : returns : The debug information ."""

implementation:
dbg = ( ' ' * indent * 2 )
dbg += '%s/' % self . id
dbg += '%s:' % self . thread_id
dbg += ' Task of %s' % self . get_name ( )
if self . task_spec . description :
    dbg += ' (%s)' % self . get_description ( )
dbg += ' State: %s' % self . get_state_name ( )
dbg += ' Children: %s' % len ( self . children )
if recursive :
    for child in self . children :
        dbg += '\n' + child . get_dump ( indent + 1 )
return dbg


signature:
def _eval_args ( args , my_task ) :
    """Parses args and evaluates any Attrib entries"""

implementation:
results = [ ]
for arg in args :
    if isinstance ( arg , Attrib ) or isinstance ( arg , PathAttrib ) :
        results . append ( valueof ( my_task , arg ) )
    else :
        results . append ( arg )
return results


signature:
def _eval_kwargs ( kwargs , my_task ) :
    """Parses kwargs and evaluates any Attrib entries"""

implementation:
results = { }
for kwarg , value in list ( kwargs . items ( ) ) :
    if isinstance ( value , Attrib ) or isinstance ( value , PathAttrib ) :
        results [ kwarg ] = valueof ( my_task , value )
    else :
        results [ kwarg ] = value
return results


signature:
def Serializable ( o ) :
    """Make sure an object is JSON - serializable
 Use this to return errors and other info that does not need to be
 deserialized or does not contain important app data . Best for returning
 error info and such"""

implementation:
if isinstance ( o , ( str , dict , int ) ) :
    return o
else :
    try :
        json . dumps ( o )
        return o
    except Exception :
        LOG . debug ( "Got a non-serilizeable object: %s" % o )
        return o . __repr__ ( )


signature:
def _send_call ( self , my_task ) :
    """Sends Celery asynchronous call and stores async call information for
 retrieval laster"""

implementation:
args , kwargs = None , None
if self . args :
    args = _eval_args ( self . args , my_task )
if self . kwargs :
    kwargs = _eval_kwargs ( self . kwargs , my_task )
LOG . debug ( "%s (task id %s) calling %s" % ( self . name , my_task . id , self . call ) , extra = dict ( data = dict ( args = args , kwargs = kwargs ) ) )
async_call = default_app . send_task ( self . call , args = args , kwargs = kwargs )
my_task . _set_internal_data ( task_id = async_call . task_id )
my_task . async_call = async_call
LOG . debug ( "'%s' called: %s" % ( self . call , my_task . async_call . task_id ) )


signature:
def _restart ( self , my_task ) :
    """Abort celery task and retry it"""

implementation:
if not my_task . _has_state ( Task . WAITING ) :
    raise WorkflowException ( my_task , "Cannot refire a task that is not" "in WAITING state" )
# Check state of existing call and abort it ( save history )
if my_task . _get_internal_data ( 'task_id' ) is not None :
    if not hasattr ( my_task , 'async_call' ) :
        task_id = my_task . _get_internal_data ( 'task_id' )
        my_task . async_call = default_app . AsyncResult ( task_id )
        my_task . deserialized = True
        my_task . async_call . state
        # manually refresh
    async_call = my_task . async_call
    if async_call . state == 'FAILED' :
        pass
    elif async_call . state in [ 'RETRY' , 'PENDING' , 'STARTED' ] :
        async_call . revoke ( )
        LOG . info ( "Celery task '%s' was in %s state and was revoked" % ( async_call . state , async_call ) )
    elif async_call . state == 'SUCCESS' :
        LOG . warning ( "Celery task '%s' succeeded, but a refire was " "requested" % async_call )
    self . _clear_celery_task_data ( my_task )
# Retrigger
return self . _start ( my_task )


signature:
def _clear_celery_task_data ( self , my_task ) :
    """Clear celery task data"""

implementation:
# Save history
if 'task_id' in my_task . internal_data : # Save history for diagnostics / forensics
    history = my_task . _get_internal_data ( 'task_history' , [ ] )
    history . append ( my_task . _get_internal_data ( 'task_id' ) )
    del my_task . internal_data [ 'task_id' ]
    my_task . _set_internal_data ( task_history = history )
if 'task_state' in my_task . internal_data :
    del my_task . internal_data [ 'task_state' ]
if 'error' in my_task . internal_data :
    del my_task . internal_data [ 'error' ]
if hasattr ( my_task , 'async_call' ) :
    delattr ( my_task , 'async_call' )
if hasattr ( my_task , 'deserialized' ) :
    delattr ( my_task , 'deserialized' )


signature:
def _start ( self , my_task , force = False ) :
    """Returns False when successfully fired , True otherwise"""

implementation:
# Deserialize async call if necessary
if not hasattr ( my_task , 'async_call' ) and my_task . _get_internal_data ( 'task_id' ) is not None :
    task_id = my_task . _get_internal_data ( 'task_id' )
    my_task . async_call = default_app . AsyncResult ( task_id )
    my_task . deserialized = True
    LOG . debug ( "Reanimate AsyncCall %s" % task_id )
# Make the call if not already done
if not hasattr ( my_task , 'async_call' ) :
    self . _send_call ( my_task )
# Get call status ( and manually refresh if deserialized )
if getattr ( my_task , "deserialized" , False ) :
    my_task . async_call . state
    # must manually refresh if deserialized
if my_task . async_call . state == 'FAILURE' :
    LOG . debug ( "Async Call for task '%s' failed: %s" % ( my_task . get_name ( ) , my_task . async_call . info ) )
    info = { }
    info [ 'traceback' ] = my_task . async_call . traceback
    info [ 'info' ] = Serializable ( my_task . async_call . info )
    info [ 'state' ] = my_task . async_call . state
    my_task . _set_internal_data ( task_state = info )
elif my_task . async_call . state == 'RETRY' :
    info = { }
    info [ 'traceback' ] = my_task . async_call . traceback
    info [ 'info' ] = Serializable ( my_task . async_call . info )
    info [ 'state' ] = my_task . async_call . state
    my_task . _set_internal_data ( task_state = info )
elif my_task . async_call . ready ( ) :
    result = my_task . async_call . result
    if isinstance ( result , Exception ) :
        LOG . warn ( "Celery call %s failed: %s" % ( self . call , result ) )
        my_task . _set_internal_data ( error = Serializable ( result ) )
        return False
    LOG . debug ( "Completed celery call %s with result=%s" % ( self . call , result ) )
    # Format result
    if self . result_key :
        data = { self . result_key : result }
    else :
        if isinstance ( result , dict ) :
            data = result
        else :
            data = { 'result' : result }
    # Load formatted result into internal _ data
    if self . merge_results :
        merge_dictionary ( my_task . internal_data , data )
    else :
        my_task . set_data ( ** data )
    return True
else :
    LOG . debug ( "async_call.ready()=%s. TryFire for '%s' " "returning False" % ( my_task . async_call . ready ( ) , my_task . get_name ( ) ) )
    return False


signature:
def ancestors ( self ) :
    """Returns list of ancestor task specs based on inputs"""

implementation:
results = [ ]
def recursive_find_ancestors ( task , stack ) :
    for input in task . inputs :
        if input not in stack :
            stack . append ( input )
            recursive_find_ancestors ( input , stack )
recursive_find_ancestors ( self , results )
return results


signature:
def set_data ( self , ** kwargs ) :
    """Defines the given data field ( s ) using the given name / value pairs ."""

implementation:
for key in kwargs :
    if key in self . defines :
        msg = "Spec data %s can not be modified" % key
        raise WorkflowException ( self , msg )
self . data . update ( kwargs )


signature:
def connect ( self , taskspec ) :
    """Connect the * following * task to this one . In other words , the
 given task is added as an output task .
 : type taskspec : TaskSpec
 : param taskspec : The new output task ."""

implementation:
self . outputs . append ( taskspec )
taskspec . _connect_notify ( self )