signature:
def _write_APSR ( self , apsr ) :
    """Auxiliary function - Writes flags from a full APSR ( only 4 msb used )"""

implementation:
V = Operators . EXTRACT ( apsr , 28 , 1 )
C = Operators . EXTRACT ( apsr , 29 , 1 )
Z = Operators . EXTRACT ( apsr , 30 , 1 )
N = Operators . EXTRACT ( apsr , 31 , 1 )
self . write ( 'APSR_V' , V )
self . write ( 'APSR_C' , C )
self . write ( 'APSR_Z' , Z )
self . write ( 'APSR_N' , N )


signature:
def _swap_mode ( self ) :
    """Toggle between ARM and Thumb mode"""

implementation:
assert self . mode in ( cs . CS_MODE_ARM , cs . CS_MODE_THUMB )
if self . mode == cs . CS_MODE_ARM :
    self . mode = cs . CS_MODE_THUMB
else :
    self . mode = cs . CS_MODE_ARM


signature:
def set_flags ( self , ** flags ) :
    """Note : For any unmodified flags , update _ last _ flags with the most recent
 committed value . Otherwise , for example , this could happen :
 overflow = 0
 instr1 computes overflow = 1 , updates _ last _ flags , doesn ' t commit
 instr2 updates all flags in _ last _ flags except overflow ( overflow remains 1 in _ last _ flags )
 instr2 commits all in _ last _ flags
 now overflow = 1 even though it should still be 0"""

implementation:
unupdated_flags = self . _last_flags . keys ( ) - flags . keys ( )
for flag in unupdated_flags :
    flag_name = f'APSR_{flag}'
    self . _last_flags [ flag ] = self . regfile . read ( flag_name )
self . _last_flags . update ( flags )


signature:
def _shift ( cpu , value , _type , amount , carry ) :
    """See Shift ( ) and Shift _ C ( ) in the ARM manual"""

implementation:
assert ( cs . arm . ARM_SFT_INVALID < _type <= cs . arm . ARM_SFT_RRX_REG )
# XXX : Capstone should set the value of an RRX shift to 1 , which is
# asserted in the manual , but it sets it to 0 , so we have to check
if _type in ( cs . arm . ARM_SFT_RRX , cs . arm . ARM_SFT_RRX_REG ) and amount != 1 :
    amount = 1
elif _type in range ( cs . arm . ARM_SFT_ASR_REG , cs . arm . ARM_SFT_RRX_REG + 1 ) :
    if cpu . mode == cs . CS_MODE_THUMB :
        src = amount . read ( )
    else :
        src_reg = cpu . instruction . reg_name ( amount ) . upper ( )
        src = cpu . regfile . read ( src_reg )
    amount = Operators . EXTRACT ( src , 0 , 8 )
if amount == 0 :
    return value , carry
width = cpu . address_bit_size
if _type in ( cs . arm . ARM_SFT_ASR , cs . arm . ARM_SFT_ASR_REG ) :
    return ASR_C ( value , amount , width )
elif _type in ( cs . arm . ARM_SFT_LSL , cs . arm . ARM_SFT_LSL_REG ) :
    return LSL_C ( value , amount , width )
elif _type in ( cs . arm . ARM_SFT_LSR , cs . arm . ARM_SFT_LSR_REG ) :
    return LSR_C ( value , amount , width )
elif _type in ( cs . arm . ARM_SFT_ROR , cs . arm . ARM_SFT_ROR_REG ) :
    return ROR_C ( value , amount , width )
elif _type in ( cs . arm . ARM_SFT_RRX , cs . arm . ARM_SFT_RRX_REG ) :
    return RRX_C ( value , carry , width )
raise NotImplementedError ( "Bad shift value" )


signature:
def MOV ( cpu , dest , src ) :
    """Implement the MOV { S } instruction .
 Note : If src operand is PC , temporarily release our logical PC
 view and conform to the spec , which dictates PC = curr instr + 8
 : param Armv7Operand dest : The destination operand ; register .
 : param Armv7Operand src : The source operand ; register or immediate ."""

implementation:
if cpu . mode == cs . CS_MODE_ARM :
    result , carry_out = src . read ( with_carry = True )
    dest . write ( result )
    cpu . set_flags ( C = carry_out , N = HighBit ( result ) , Z = ( result == 0 ) )
else : # thumb mode cannot do wonky things to the operand , so no carry calculation
    result = src . read ( )
    dest . write ( result )
    cpu . set_flags ( N = HighBit ( result ) , Z = ( result == 0 ) )


signature:
def MOVT ( cpu , dest , src ) :
    """MOVT writes imm16 to Rd [ 31:16 ] . The write does not affect Rd [ 15:0 ] .
 : param Armv7Operand dest : The destination operand ; register
 : param Armv7Operand src : The source operand ; 16 - bit immediate"""

implementation:
assert src . type == 'immediate'
imm = src . read ( )
low_halfword = dest . read ( ) & Mask ( 16 )
dest . write ( ( imm << 16 ) | low_halfword )


signature:
def MRC ( cpu , coprocessor , opcode1 , dest , coprocessor_reg_n , coprocessor_reg_m , opcode2 ) :
    """MRC moves to ARM register from coprocessor .
 : param Armv7Operand coprocessor : The name of the coprocessor ; immediate
 : param Armv7Operand opcode1 : coprocessor specific opcode ; 3 - bit immediate
 : param Armv7Operand dest : the destination operand : register
 : param Armv7Operand coprocessor _ reg _ n : the coprocessor register ; immediate
 : param Armv7Operand coprocessor _ reg _ m : the coprocessor register ; immediate
 : param Armv7Operand opcode2 : coprocessor specific opcode ; 3 - bit immediate"""

implementation:
assert coprocessor . type == 'coprocessor'
assert opcode1 . type == 'immediate'
assert opcode2 . type == 'immediate'
assert dest . type == 'register'
imm_coprocessor = coprocessor . read ( )
imm_opcode1 = opcode1 . read ( )
imm_opcode2 = opcode2 . read ( )
coprocessor_n_name = coprocessor_reg_n . read ( )
coprocessor_m_name = coprocessor_reg_m . read ( )
if 15 == imm_coprocessor : # MMU
    if 0 == imm_opcode1 :
        if 13 == coprocessor_n_name :
            if 3 == imm_opcode2 :
                dest . write ( cpu . regfile . read ( 'P15_C13' ) )
                return
raise NotImplementedError ( "MRC: unimplemented combination of coprocessor, opcode, and coprocessor register" )


signature:
def LDRD ( cpu , dest1 , dest2 , src , offset = None ) :
    """Loads double width data from memory ."""

implementation:
assert dest1 . type == 'register'
assert dest2 . type == 'register'
assert src . type == 'memory'
mem1 = cpu . read_int ( src . address ( ) , 32 )
mem2 = cpu . read_int ( src . address ( ) + 4 , 32 )
writeback = cpu . _compute_writeback ( src , offset )
dest1 . write ( mem1 )
dest2 . write ( mem2 )
cpu . _cs_hack_ldr_str_writeback ( src , offset , writeback )


signature:
def STRD ( cpu , src1 , src2 , dest , offset = None ) :
    """Writes the contents of two registers to memory ."""

implementation:
assert src1 . type == 'register'
assert src2 . type == 'register'
assert dest . type == 'memory'
val1 = src1 . read ( )
val2 = src2 . read ( )
writeback = cpu . _compute_writeback ( dest , offset )
cpu . write_int ( dest . address ( ) , val1 , 32 )
cpu . write_int ( dest . address ( ) + 4 , val2 , 32 )
cpu . _cs_hack_ldr_str_writeback ( dest , offset , writeback )


signature:
def LDREX ( cpu , dest , src , offset = None ) :
    """LDREX loads data from memory .
 * If the physical address has the shared TLB attribute , LDREX
 tags the physical address as exclusive access for the current
 processor , and clears any exclusive access tag for this
 processor for any other physical address .
 * Otherwise , it tags the fact that the executing processor has
 an outstanding tagged physical address .
 : param Armv7Operand dest : the destination register ; register
 : param Armv7Operand src : the source operand : register"""

implementation:
# TODO : add lock mechanism to underlying memory - - GR , 2017-06-06
cpu . _LDR ( dest , src , 32 , False , offset )


signature:
def STREX ( cpu , status , * args ) :
    """STREX performs a conditional store to memory .
 : param Armv7Operand status : the destination register for the returned status ; register"""

implementation:
# TODO : implement conditional return with appropriate status - - GR , 2017-06-06
status . write ( 0 )
return cpu . _STR ( cpu . address_bit_size , * args )


signature:
def _UXT ( cpu , dest , src , src_width ) :
    """Helper for UXT * family of instructions .
 : param ARMv7Operand dest : the destination register ; register
 : param ARMv7Operand dest : the source register ; register
 : param int src _ width : bits to consider of the src operand"""

implementation:
val = GetNBits ( src . read ( ) , src_width )
word = Operators . ZEXTEND ( val , cpu . address_bit_size )
dest . write ( word )


signature:
def ADR ( cpu , dest , src ) :
    """Address to Register adds an immediate value to the PC value , and writes the result to the destination register .
 : param ARMv7Operand dest : Specifies the destination register .
 : param ARMv7Operand src :
 Specifies the label of an instruction or literal data item whose address is to be loaded into
 < Rd > . The assembler calculates the required value of the offset from the Align ( PC , 4)
 value of the ADR instruction to this label ."""

implementation:
aligned_pc = ( cpu . instruction . address + 4 ) & 0xfffffffc
dest . write ( aligned_pc + src . read ( ) )


signature:
def ADDW ( cpu , dest , src , add ) :
    """This instruction adds an immediate value to a register value , and writes the result to the destination register .
 It doesn ' t update the condition flags .
 : param ARMv7Operand dest : Specifies the destination register . If omitted , this register is the same as src .
 : param ARMv7Operand src :
 Specifies the register that contains the first operand . If the SP is specified for dest , see ADD ( SP plus
 immediate ) . If the PC is specified for dest , see ADR .
 : param ARMv7Operand add :
 Specifies the immediate value to be added to the value obtained from src . The range of allowed values is
 0-4095."""

implementation:
aligned_pc = ( cpu . instruction . address + 4 ) & 0xfffffffc
if src . type == 'register' and src . reg in ( 'PC' , 'R15' ) :
    src = aligned_pc
else :
    src = src . read ( )
dest . write ( src + add . read ( ) )


signature:
def CBZ ( cpu , op , dest ) :
    """Compare and Branch on Zero compares the value in a register with zero , and conditionally branches forward
 a constant value . It does not affect the condition flags .
 : param ARMv7Operand op : Specifies the register that contains the first operand .
 : param ARMv7Operand dest :
 Specifies the label of the instruction that is to be branched to . The assembler calculates the
 required value of the offset from the PC value of the CBZ instruction to this label , then
 selects an encoding that will set imm32 to that offset . Allowed offsets are even numbers in
 the range 0 to 126."""

implementation: cpu . PC = Operators . ITEBV ( cpu . address_bit_size , op . read ( ) , cpu . PC , dest . read ( ) )


signature:
def TBH ( cpu , dest ) :
    """Table Branch Halfword causes a PC - relative forward branch using a table of single halfword offsets . A base
 register provides a pointer to the table , and a second register supplies an index into the table . The branch
 length is twice the value of the halfword returned from the table .
 : param ARMv7Operand dest : see below ; register"""

implementation:
# Capstone merges the two registers values into one operand , so we need to extract them back
# Specifies the base register . This contains the address of the table of branch lengths . This
# register is allowed to be the PC . If it is , the table immediately follows this instruction .
base_addr = dest . get_mem_base_addr ( )
if dest . mem . base in ( 'PC' , 'R15' ) :
    base_addr = cpu . PC
# Specifies the index register . This contains an integer pointing to a halfword within the table .
# The offset within the table is twice the value of the index .
offset = cpu . read_int ( base_addr + dest . get_mem_offset ( ) , 16 )
offset = Operators . ZEXTEND ( offset , cpu . address_bit_size )
cpu . PC += ( offset << 1 )


signature:
def _LDM ( cpu , insn_id , base , regs ) :
    """LDM ( Load Multiple ) loads a non - empty subset , or possibly all , of the general - purpose registers from
 sequential memory locations . It is useful for block loads , stack operations and procedure exit sequences .
 : param int insn _ id : should be one of ARM _ INS _ LDM , ARM _ INS _ LDMIB , ARM _ INS _ LDMDA , ARM _ INS _ LDMDB
 : param Armv7Operand base : Specifies the base register .
 : param list [ Armv7Operand ] regs :
 Is a list of registers . It specifies the set of registers to be loaded by the LDM instruction .
 The registers are loaded in sequence , the lowest - numbered register from the lowest memory
 address ( start _ address ) , through to the highest - numbered register from the highest memory
 address ( end _ address ) . If the PC is specified in the register list ( opcode bit [ 15 ] is set ) ,
 the instruction causes a branch to the address ( data ) loaded into the PC .
 It ' s technically UNKNOWN if you writeback to a register you loaded into , but we let it slide ."""

implementation:
if cpu . instruction . usermode :
    raise NotImplementedError ( "Use of the S bit is not supported" )
increment = insn_id in ( cs . arm . ARM_INS_LDM , cs . arm . ARM_INS_LDMIB )
after = insn_id in ( cs . arm . ARM_INS_LDM , cs . arm . ARM_INS_LDMDA )
address = base . read ( )
for reg in regs :
    if not after :
        address += ( 1 if increment else - 1 ) * ( reg . size // 8 )
    reg . write ( cpu . read_int ( address , reg . size ) )
    if reg . reg in ( 'PC' , 'R15' ) : # The general - purpose registers loaded can include the PC . If they do , the word loaded for the PC is
    # treated as an address and a branch occurs to that address . In ARMv5 and above , bit [ 0 ] of the loaded
    # value determines whether execution continues after this branch in ARM state or in Thumb state , as
    # though a BX instruction had been executed .
        cpu . _set_mode_by_val ( cpu . PC )
        cpu . PC = cpu . PC & ~ 1
    if after :
        address += ( 1 if increment else - 1 ) * ( reg . size // 8 )
if cpu . instruction . writeback :
    base . writeback ( address )


signature:
def _STM ( cpu , insn_id , base , regs ) :
    """STM ( Store Multiple ) stores a non - empty subset ( or possibly all ) of the general - purpose registers to
 sequential memory locations .
 : param int insn _ id : should be one of ARM _ INS _ STM , ARM _ INS _ STMIB , ARM _ INS _ STMDA , ARM _ INS _ STMDB
 : param Armv7Operand base : Specifies the base register .
 : param list [ Armv7Operand ] regs :
 Is a list of registers . It specifies the set of registers to be stored by the STM instruction .
 The registers are stored in sequence , the lowest - numbered register to the lowest
 memory address ( start _ address ) , through to the highest - numbered register to the
 highest memory address ( end _ address ) ."""

implementation:
if cpu . instruction . usermode :
    raise NotImplementedError ( "Use of the S bit is not supported" )
increment = insn_id in ( cs . arm . ARM_INS_STM , cs . arm . ARM_INS_STMIB )
after = insn_id in ( cs . arm . ARM_INS_STM , cs . arm . ARM_INS_STMDA )
address = base . read ( )
for reg in regs :
    if not after :
        address += ( 1 if increment else - 1 ) * ( reg . size // 8 )
    cpu . write_int ( address , reg . read ( ) , reg . size )
    if after :
        address += ( 1 if increment else - 1 ) * ( reg . size // 8 )
if cpu . instruction . writeback :
    base . writeback ( address )


signature:
def _SR ( cpu , insn_id , dest , op , * rest ) :
    """Notes on Capstone behavior :
 - In ARM mode , _ SR reg has ` rest ` , but _ SR imm does not , its baked into ` op ` .
 - In ARM mode , ` lsr r1 , r2 ` will have a ` rest [ 0 ] `
 - In Thumb mode , ` lsr r1 , r2 ` will have an empty ` rest `
 - In ARM mode , something like ` lsr r1 , 3 ` will not have ` rest ` and op will be
 the immediate ."""

implementation:
assert insn_id in ( cs . arm . ARM_INS_ASR , cs . arm . ARM_INS_LSL , cs . arm . ARM_INS_LSR )
if insn_id == cs . arm . ARM_INS_ASR :
    if rest and rest [ 0 ] . type == 'immediate' :
        srtype = cs . arm . ARM_SFT_ASR
    else :
        srtype = cs . arm . ARM_SFT_ASR_REG
elif insn_id == cs . arm . ARM_INS_LSL :
    if rest and rest [ 0 ] . type == 'immediate' :
        srtype = cs . arm . ARM_SFT_LSL
    else :
        srtype = cs . arm . ARM_SFT_LSL_REG
elif insn_id == cs . arm . ARM_INS_LSR :
    if rest and rest [ 0 ] . type == 'immediate' :
        srtype = cs . arm . ARM_SFT_LSR
    else :
        srtype = cs . arm . ARM_SFT_LSR_REG
carry = cpu . regfile . read ( 'APSR_C' )
if rest and rest [ 0 ] . type == 'register' : # FIXME we should make Operand . op private ( and not accessible )
    result , carry = cpu . _shift ( op . read ( ) , srtype , rest [ 0 ] . op . reg , carry )
elif rest and rest [ 0 ] . type == 'immediate' :
    amount = rest [ 0 ] . read ( )
    result , carry = cpu . _shift ( op . read ( ) , srtype , amount , carry )
elif cpu . mode == cs . CS_MODE_THUMB :
    result , carry = cpu . _shift ( dest . read ( ) , srtype , op , carry )
else :
    result , carry = op . read ( with_carry = True )
dest . write ( result )
cpu . set_flags ( N = HighBit ( result ) , Z = ( result == 0 ) , C = carry )


signature:
def _fix_index ( self , index ) :
    """: param slice index :"""

implementation:
stop , start = index . stop , index . start
if start is None :
    start = 0
if stop is None :
    stop = len ( self )
return start , stop


signature:
def _dict_diff ( d1 , d2 ) :
    """Produce a dict that includes all the keys in d2 that represent different values in d1 , as well as values that
 aren ' t in d1.
 : param dict d1 : First dict
 : param dict d2 : Dict to compare with
 : rtype : dict"""

implementation:
d = { }
for key in set ( d1 ) . intersection ( set ( d2 ) ) :
    if d2 [ key ] != d1 [ key ] :
        d [ key ] = d2 [ key ]
for key in set ( d2 ) . difference ( set ( d1 ) ) :
    d [ key ] = d2 [ key ]
return d


signature:
def locked_context ( self , key = None , value_type = list ) :
    """A context manager that provides safe parallel access to the global Manticore context .
 This should be used to access the global Manticore context
 when parallel analysis is activated . Code within the ` with ` block is executed
 atomically , so access of shared variables should occur within ."""

implementation:
plugin_context_name = str ( type ( self ) )
with self . manticore . locked_context ( plugin_context_name , dict ) as context :
    assert value_type in ( list , dict , set )
    ctx = context . get ( key , value_type ( ) )
    yield ctx
    context [ key ] = ctx


signature:
def context ( self ) :
    """Convenient access to shared context"""

implementation:
plugin_context_name = str ( type ( self ) )
if plugin_context_name not in self . manticore . context :
    self . manticore . context [ plugin_context_name ] = { }
return self . manticore . context [ plugin_context_name ]


signature:
def add_finding ( self , state , address , pc , finding , at_init , constraint = True ) :
    """Logs a finding at specified contract and assembler line .
 : param state : current state
 : param address : contract address of the finding
 : param pc : program counter of the finding
 : param at _ init : true if executing the constructor
 : param finding : textual description of the finding
 : param constraint : finding is considered reproducible only when constraint is True"""

implementation:
if issymbolic ( pc ) :
    pc = simplify ( pc )
if isinstance ( pc , Constant ) :
    pc = pc . value
if not isinstance ( pc , int ) :
    raise ValueError ( "PC must be a number" )
self . get_findings ( state ) . add ( ( address , pc , finding , at_init , constraint ) )
with self . locked_global_findings ( ) as gf :
    gf . add ( ( address , pc , finding , at_init ) )
# Fixme for ever broken logger
logger . warning ( finding )


signature:
def add_finding_here ( self , state , finding , constraint = True ) :
    """Logs a finding in current contract and assembler line .
 : param state : current state
 : param finding : textual description of the finding
 : param constraint : finding is considered reproducible only when constraint is True"""

implementation:
address = state . platform . current_vm . address
pc = state . platform . current_vm . pc
if isinstance ( pc , Constant ) :
    pc = pc . value
if not isinstance ( pc , int ) :
    raise ValueError ( "PC must be a number" )
at_init = state . platform . current_transaction . sort == 'CREATE'
self . add_finding ( state , address , pc , finding , at_init , constraint )


signature:
def _save_current_location ( self , state , finding , condition = True ) :
    """Save current location in the internal locations list and returns a textual id for it .
 This is used to save locations that could later be promoted to a finding if other conditions hold
 See _ get _ location ( )
 : param state : current state
 : param finding : textual description of the finding
 : param condition : general purpose constraint"""

implementation:
address = state . platform . current_vm . address
pc = state . platform . current_vm . pc
at_init = state . platform . current_transaction . sort == 'CREATE'
location = ( address , pc , finding , at_init , condition )
hash_id = hashlib . sha1 ( str ( location ) . encode ( ) ) . hexdigest ( )
state . context . setdefault ( '{:s}.locations' . format ( self . name ) , { } ) [ hash_id ] = location
return hash_id


signature:
def _get_location ( self , state , hash_id ) :
    """Get previously saved location
 A location is composed of : address , pc , finding , at _ init , condition"""

implementation: return state . context . setdefault ( '{:s}.locations' . format ( self . name ) , { } ) [ hash_id ]


signature:
def _signed_sub_overflow ( state , a , b ) :
    """Sign extend the value to 512 bits and check the result can be represented
 in 256 . Following there is a 32 bit excerpt of this condition :
 a - b - 800000 - 3fffff - 000001 + 00000 + 000001 + 3fffff + 7fffff
 + 800000 False False False False True True True
 + c000001 False False False False False False True
 + fffff False False False False False False False
 + 00000 True False False False False False False
 + 000001 True False False False False False False
 + 3fffff True False False False False False False
 + 7fffff True True True False False False False"""

implementation:
sub = Operators . SEXTEND ( a , 256 , 512 ) - Operators . SEXTEND ( b , 256 , 512 )
cond = Operators . OR ( sub < - ( 1 << 255 ) , sub >= ( 1 << 255 ) )
return cond


signature:
def _signed_add_overflow ( state , a , b ) :
    """Sign extend the value to 512 bits and check the result can be represented
 in 256 . Following there is a 32 bit excerpt of this condition :
 a + b - 800000 - 3fffff - 000001 + 00000 + 000001 + 3fffff + 7fffff
 + 800000 True True True False False False False
 + c000001 True False False False False False False
 + fffff True False False False False False False
 + 00000 False False False False False False False
 + 000001 False False False False False False True
 + 3fffff False False False False False False True
 + 7fffff False False False False True True True"""

implementation:
add = Operators . SEXTEND ( a , 256 , 512 ) + Operators . SEXTEND ( b , 256 , 512 )
cond = Operators . OR ( add < - ( 1 << 255 ) , add >= ( 1 << 255 ) )
return cond


signature:
def _unsigned_sub_overflow ( state , a , b ) :
    """Sign extend the value to 512 bits and check the result can be represented
 in 256 . Following there is a 32 bit excerpt of this condition :
 a - b fffff bfffff 8000001 00000 000001 3fffff 7fffff
 fffff True True True False True True True
 bfffff True True True False False True True
 8000001 True True True False False True True
 00000 False False False False False True False
 000001 True False False False False True False
 fffff True True True True True True True
 7fffff True True True False False True False"""

implementation:
cond = Operators . UGT ( b , a )
return cond


signature:
def _unsigned_add_overflow ( state , a , b ) :
    """Sign extend the value to 512 bits and check the result can be represented
 in 256 . Following there is a 32 bit excerpt of this condition :
 a + b fffff bfffff 8000001 00000 000001 3fffff 7fffff
 fffff True True True False True True True
 bfffff True True True False False True True
 8000001 True True True False False True True
 00000 False False False False False True False
 000001 True False False False False True False
 fffff True True True True True True True
 7fffff True True True False False True False"""

implementation:
add = Operators . ZEXTEND ( a , 512 ) + Operators . ZEXTEND ( b , 512 )
cond = Operators . UGE ( add , 1 << 256 )
return cond


implementation:
mul = Operators . SEXTEND ( a , 256 , 512 ) * Operators . SEXTEND ( b , 256 , 512 )
cond = Operators . OR ( mul < - ( 1 << 255 ) , mul >= ( 1 << 255 ) )
return cond


implementation:
mul = Operators . SEXTEND ( a , 256 , 512 ) * Operators . SEXTEND ( b , 256 , 512 )
cond = Operators . UGE ( mul , 1 << 256 )
return cond


signature:
def _in_user_func ( state ) :
    """: param state : current state
 : return : whether the current execution is in a user - defined function or not .
 NOTE / TODO / FIXME : As this may produce false postives , this is not in the base ` Detector ` class .
 It should be fixed at some point and moved there . See below .
 The first 4 bytes of tx data is keccak256 hash of the function signature that is called by given tx .
 All transactions start within Solidity dispatcher function : it takes passed hash and dispatches
 the execution to given function based on it .
 So : if we are in the dispatcher , * and contract have some functions * one of the first four tx data bytes
 will effectively have more than one solutions .
 BUT if contract have only a fallback function , the equation below may return more solutions when we are
 in a dispatcher function . < - - - because of that , we warn that the detector is not that stable
 for contracts with only a fallback function ."""

implementation:
# If we are already in user function ( we cached it ) let ' s just return True
in_function = state . context . get ( 'in_function' , False )
prev_tx_count = state . context . get ( 'prev_tx_count' , 0 )
curr_tx_count = len ( state . platform . transactions )
new_human_tx = prev_tx_count != curr_tx_count
if in_function and not new_human_tx :
    return True
# This is expensive call , so we cache it
in_function = len ( state . solve_n ( state . platform . current_transaction . data [ : 4 ] , 2 ) ) == 1
state . context [ 'in_function' ] = in_function
state . context [ 'prev_tx_count' ] = curr_tx_count
return in_function


signature:
def disassemble_instruction ( self , code , pc ) :
    """Get next instruction using the Capstone disassembler
 : param str code : binary blob to be disassembled
 : param long pc : program counter"""

implementation: return next ( self . disasm . disasm ( code , pc ) )


signature:
def istainted ( arg , taint = None ) :
    """Helper to determine whether an object if tainted .
 : param arg : a value or Expression
 : param taint : a regular expression matching a taint value ( eg . ' IMPORTANT . * ' ) . If None , this function checks for any taint value ."""

implementation:
if not issymbolic ( arg ) :
    return False
if taint is None :
    return len ( arg . taint ) != 0
for arg_taint in arg . taint :
    m = re . match ( taint , arg_taint , re . DOTALL | re . IGNORECASE )
    if m :
        return True
return False


signature:
def get_taints ( arg , taint = None ) :
    """Helper to list an object taints .
 : param arg : a value or Expression
 : param taint : a regular expression matching a taint value ( eg . ' IMPORTANT . * ' ) . If None , this function checks for any taint value ."""

implementation:
if not issymbolic ( arg ) :
    return
for arg_taint in arg . taint :
    if taint is not None :
        m = re . match ( taint , arg_taint , re . DOTALL | re . IGNORECASE )
        if m :
            yield arg_taint
    else :
        yield arg_taint
return


signature:
def taint_with ( arg , taint , value_bits = 256 , index_bits = 256 ) :
    """Helper to taint a value .
 : param arg : a value or Expression
 : param taint : a regular expression matching a taint value ( eg . ' IMPORTANT . * ' ) . If None , this function checks for any taint value ."""

implementation:
from . . core . smtlib import BitVecConstant
# prevent circular imports
tainted_fset = frozenset ( ( taint , ) )
if not issymbolic ( arg ) :
    if isinstance ( arg , int ) :
        arg = BitVecConstant ( value_bits , arg )
        arg . _taint = tainted_fset
    else :
        raise ValueError ( "type not supported" )
else :
    arg = copy . copy ( arg )
    arg . _taint |= tainted_fset
return arg


signature:
def interval_intersection ( min1 , max1 , min2 , max2 ) :
    """Given two intervals , ( min1 , max1 ) and ( min2 , max2 ) return their intersecting interval ,
 or None if they do not overlap ."""

implementation:
left , right = max ( min1 , min2 ) , min ( max1 , max2 )
if left < right :
    return left , right
return None


signature:
def add ( self , constraint , check = False ) :
    """Add a constraint to the set
 : param constraint : The constraint to add to the set .
 : param check : Currently unused .
 : return :"""

implementation:
if isinstance ( constraint , bool ) :
    constraint = BoolConstant ( constraint )
assert isinstance ( constraint , Bool )
constraint = simplify ( constraint )
# If self . _ child is not None this constraint set has been forked and a
# a derived constraintset may be using this . So we can ' t add any more
# constraints to this one . After the child constraintSet is deleted
# we regain the ability to add constraints .
if self . _child is not None :
    raise Exception ( 'ConstraintSet is frozen' )
if isinstance ( constraint , BoolConstant ) :
    if not constraint . value :
        logger . info ( "Adding an impossible constant constraint" )
        self . _constraints = [ constraint ]
    else :
        return
self . _constraints . append ( constraint )
if check :
    from . . . core . smtlib import solver
    if not solver . check ( self ) :
        raise ValueError ( "Added an impossible constraint" )


signature:
def _declare ( self , var ) :
    """Declare the variable ` var `"""

implementation:
if var . name in self . _declarations :
    raise ValueError ( 'Variable already declared' )
self . _declarations [ var . name ] = var
return var


signature:
def declarations ( self ) :
    """Returns the variable expressions of this constraint set"""

implementation:
declarations = GetDeclarations ( )
for a in self . constraints :
    try :
        declarations . visit ( a )
    except RuntimeError : # TODO : ( defunct ) move recursion management out of PickleSerializer
        if sys . getrecursionlimit ( ) >= PickleSerializer . MAX_RECURSION :
            raise Exception ( f'declarations recursion limit surpassed {PickleSerializer.MAX_RECURSION}, aborting' )
        new_limit = sys . getrecursionlimit ( ) + PickleSerializer . DEFAULT_RECURSION
        if new_limit <= PickleSerializer . DEFAULT_RECURSION :
            sys . setrecursionlimit ( new_limit )
            return self . declarations
return declarations . result


signature:
def constraints ( self ) :
    """: rtype tuple
 : return : All constraints represented by this and parent sets ."""

implementation:
if self . _parent is not None :
    return tuple ( self . _constraints ) + self . _parent . constraints
return tuple ( self . _constraints )


signature:
def is_declared ( self , expression_var ) :
    """True if expression _ var is declared in this constraint set"""

implementation:
if not isinstance ( expression_var , Variable ) :
    raise ValueError ( f'Expression must be a Variable (not a {type(expression_var)})' )
return any ( expression_var is x for x in self . get_declared_variables ( ) )


signature:
def migrate ( self , expression , name_migration_map = None ) :
    """Migrate an expression created for a different constraint set to self .
 Returns an expression that can be used with this constraintSet
 All the foreign variables used in the expression are replaced by
 variables of this constraint set . If the variable was replaced before
 the replacement is taken from the provided migration map .
 The migration mapping is updated with new replacements .
 : param expression : the potentially foreign expression
 : param name _ migration _ map : mapping of already migrated variables . maps from string name of foreign variable to its currently existing migrated string name . this is updated during this migration .
 : return : a migrated expression where all the variables are local . name _ migration _ map is updated"""

implementation:
if name_migration_map is None :
    name_migration_map = { }
# name _ migration _ map - > object _ migration _ map
# Based on the name mapping in name _ migration _ map build an object to
# object mapping to be used in the replacing of variables
# inv : object _ migration _ map ' s keys should ALWAYS be external / foreign
# expressions , and its values should ALWAYS be internal / local expressions
object_migration_map = { }
# List of foreign vars used in expression
foreign_vars = itertools . filterfalse ( self . is_declared , get_variables ( expression ) )
for foreign_var in foreign_vars : # If a variable with the same name was previously migrated
    if foreign_var . name in name_migration_map :
        migrated_name = name_migration_map [ foreign_var . name ]
        native_var = self . get_variable ( migrated_name )
        assert native_var is not None , "name_migration_map contains a variable that does not exist in this ConstraintSet"
        object_migration_map [ foreign_var ] = native_var
    else : # foreign _ var was not found in the local declared variables nor
    # any variable with the same name was previously migrated
    # let ' s make a new unique internal name for it
        migrated_name = foreign_var . name
        if migrated_name in self . _declarations :
            migrated_name = self . _make_unique_name ( f'{foreign_var.name}_migrated' )
        # Create and declare a new variable of given type
        if isinstance ( foreign_var , Bool ) :
            new_var = self . new_bool ( name = migrated_name )
        elif isinstance ( foreign_var , BitVec ) :
            new_var = self . new_bitvec ( foreign_var . size , name = migrated_name )
        elif isinstance ( foreign_var , Array ) : # Note that we are discarding the ArrayProxy encapsulation
            new_var = self . new_array ( index_max = foreign_var . index_max , index_bits = foreign_var . index_bits , value_bits = foreign_var . value_bits , name = migrated_name ) . array
        else :
            raise NotImplemented ( f"Unknown expression type {type(var)} encountered during expression migration" )
        # Update the var to var mapping
        object_migration_map [ foreign_var ] = new_var
        # Update the name to name mapping
        name_migration_map [ foreign_var . name ] = new_var . name
# Actually replace each appearance of migrated variables by the new ones
migrated_expression = replace ( expression , object_migration_map )
return migrated_expression


signature:
def new_bool ( self , name = None , taint = frozenset ( ) , avoid_collisions = False ) :
    """Declares a free symbolic boolean in the constraint store
 : param name : try to assign name to internal variable representation ,
 if not unique , a numeric nonce will be appended
 : param avoid _ collisions : potentially avoid _ collisions the variable to avoid name collisions if True
 : return : a fresh BoolVariable"""

implementation:
if name is None :
    name = 'B'
    avoid_collisions = True
if avoid_collisions :
    name = self . _make_unique_name ( name )
if not avoid_collisions and name in self . _declarations :
    raise ValueError ( f'Name {name} already used' )
var = BoolVariable ( name , taint = taint )
return self . _declare ( var )


signature:
def new_bitvec ( self , size , name = None , taint = frozenset ( ) , avoid_collisions = False ) :
    """Declares a free symbolic bitvector in the constraint store
 : param size : size in bits for the bitvector
 : param name : try to assign name to internal variable representation ,
 if not unique , a numeric nonce will be appended
 : param avoid _ collisions : potentially avoid _ collisions the variable to avoid name collisions if True
 : return : a fresh BitVecVariable"""

implementation:
if not ( size == 1 or size % 8 == 0 ) :
    raise Exception ( f'Invalid bitvec size {size}' )
if name is None :
    name = 'BV'
    avoid_collisions = True
if avoid_collisions :
    name = self . _make_unique_name ( name )
if not avoid_collisions and name in self . _declarations :
    raise ValueError ( f'Name {name} already used' )
var = BitVecVariable ( size , name , taint = taint )
return self . _declare ( var )


signature:
def new_array ( self , index_bits = 32 , name = None , index_max = None , value_bits = 8 , taint = frozenset ( ) , avoid_collisions = False , default = None ) :
    """Declares a free symbolic array of value _ bits long bitvectors in the constraint store .
 : param index _ bits : size in bits for the array indexes one of [ 32 , 64]
 : param value _ bits : size in bits for the array values
 : param name : try to assign name to internal variable representation ,
 if not unique , a numeric nonce will be appended
 : param index _ max : upper limit for indexes on this array ( # FIXME )
 : param avoid _ collisions : potentially avoid _ collisions the variable to avoid name collisions if True
 : param default : default for not initialized values
 : return : a fresh ArrayProxy"""

implementation:
if name is None :
    name = 'A'
    avoid_collisions = True
if avoid_collisions :
    name = self . _make_unique_name ( name )
if not avoid_collisions and name in self . _declarations :
    raise ValueError ( f'Name {name} already used' )
var = self . _declare ( ArrayVariable ( index_bits , index_max , value_bits , name , taint = taint ) )
return ArrayProxy ( var , default = default )


signature:
def train_encoder ( X , y , fold_count , encoder ) :
    """Defines folds and performs the data preprocessing ( categorical encoding , NaN imputation , normalization )
 Returns a list with { X _ train , y _ train , X _ test , y _ test } , average fit _ encoder _ time and average score _ encoder _ time
 Note : We normalize all features ( not only numerical features ) because otherwise SVM would
 get stuck for hours on ordinal encoded cylinder . bands . arff dataset due to presence of
 unproportionally high values .
 Note : The fold count is variable because there are datasets , which have less than 10 samples in the minority class .
 Note : We do not use pipelines because of :
 https : / / github . com / scikit - learn / scikit - learn / issues / 11832"""

implementation:
kf = StratifiedKFold ( n_splits = fold_count , shuffle = True , random_state = 2001 )
encoder = deepcopy ( encoder )
# Because of https : / / github . com / scikit - learn - contrib / categorical - encoding / issues / 106
imputer = SimpleImputer ( strategy = 'mean' )
scaler = StandardScaler ( )
folds = [ ]
fit_encoder_time = 0
score_encoder_time = 0
for train_index , test_index in kf . split ( X , y ) : # Split data
    X_train , X_test = X . iloc [ train_index , : ] . reset_index ( drop = True ) , X . iloc [ test_index , : ] . reset_index ( drop = True )
    y_train , y_test = y [ train_index ] . reset_index ( drop = True ) , y [ test_index ] . reset_index ( drop = True )
    # Training
    start_time = time . time ( )
    X_train = encoder . fit_transform ( X_train , y_train )
    fit_encoder_time += time . time ( ) - start_time
    X_train = imputer . fit_transform ( X_train )
    X_train = scaler . fit_transform ( X_train )
    # Testing
    start_time = time . time ( )
    X_test = encoder . transform ( X_test )
    score_encoder_time += time . time ( ) - start_time
    X_test = imputer . transform ( X_test )
    X_test = scaler . transform ( X_test )
    folds . append ( [ X_train , y_train , X_test , y_test ] )
return folds , fit_encoder_time / fold_count , score_encoder_time / fold_count


signature:
def train_model ( folds , model ) :
    """Evaluation with :
 Matthews correlation coefficient : represents thresholding measures
 AUC : represents ranking measures
 Brier score : represents calibration measures"""

implementation:
scores = [ ]
fit_model_time = 0
# Sum of all the time spend on fitting the training data , later on normalized
score_model_time = 0
# Sum of all the time spend on scoring the testing data , later on normalized
for X_train , y_train , X_test , y_test in folds : # Training
    start_time = time . time ( )
    with ignore_warnings ( category = ConvergenceWarning ) : # Yes , neural networks do not always converge
        model . fit ( X_train , y_train )
    fit_model_time += time . time ( ) - start_time
    prediction_train_proba = model . predict_proba ( X_train ) [ : , 1 ]
    prediction_train = ( prediction_train_proba >= 0.5 ) . astype ( 'uint8' )
    # Testing
    start_time = time . time ( )
    prediction_test_proba = model . predict_proba ( X_test ) [ : , 1 ]
    score_model_time += time . time ( ) - start_time
    prediction_test = ( prediction_test_proba >= 0.5 ) . astype ( 'uint8' )
    # When all the predictions are of a single class , we get a RuntimeWarning in matthews _ corr
    with warnings . catch_warnings ( ) :
        warnings . simplefilter ( "ignore" )
        scores . append ( [ sklearn . metrics . matthews_corrcoef ( y_test , prediction_test ) , sklearn . metrics . matthews_corrcoef ( y_train , prediction_train ) , sklearn . metrics . roc_auc_score ( y_test , prediction_test_proba ) , sklearn . metrics . roc_auc_score ( y_train , prediction_train_proba ) , sklearn . metrics . brier_score_loss ( y_test , prediction_test_proba ) , sklearn . metrics . brier_score_loss ( y_train , prediction_train_proba ) ] )
return np . mean ( scores , axis = 0 ) , fit_model_time / len ( folds ) , score_model_time / len ( folds )


signature:
def fit ( self , X , y = None , ** kwargs ) :
    """Fit encoder according to X and y .
 Parameters
 X : array - like , shape = [ n _ samples , n _ features ]
 Training vectors , where n _ samples is the number of samples
 and n _ features is the number of features .
 y : array - like , shape = [ n _ samples ]
 Target values .
 Returns
 self : encoder
 Returns self ."""

implementation:
self . base_n_encoder . fit ( X , y , ** kwargs )
return self


implementation:
# if the input dataset isn ' t already a dataframe , convert it to one ( using default column names )
# first check the type
X = util . convert_input ( X )
self . _dim = X . shape [ 1 ]
# if columns aren ' t passed , just use every string column
if self . cols is None :
    self . cols = util . get_obj_cols ( X )
else :
    self . cols = util . convert_cols_to_list ( self . cols )
if self . handle_missing == 'error' :
    if X [ self . cols ] . isnull ( ) . any ( ) . bool ( ) :
        raise ValueError ( 'Columns to be encoded can not contain null' )
# train an ordinal pre - encoder
self . ordinal_encoder = OrdinalEncoder ( verbose = self . verbose , cols = self . cols , handle_unknown = 'value' , handle_missing = 'value' )
self . ordinal_encoder = self . ordinal_encoder . fit ( X )
ordinal_mapping = self . ordinal_encoder . category_mapping
mappings_out = [ ]
for switch in ordinal_mapping :
    values = switch . get ( 'mapping' )
    col = switch . get ( 'col' )
    column_mapping = self . fit_sum_coding ( col , values , self . handle_missing , self . handle_unknown )
    mappings_out . append ( { 'col' : switch . get ( 'col' ) , 'mapping' : column_mapping , } )
self . mapping = mappings_out
X_temp = self . transform ( X , override_return_df = True )
self . feature_names = X_temp . columns . tolist ( )
# drop all output columns with 0 variance .
if self . drop_invariant :
    self . drop_cols = [ ]
    generated_cols = util . get_generated_cols ( X , X_temp , self . cols )
    self . drop_cols = [ x for x in generated_cols if X_temp [ x ] . var ( ) <= 10e-5 ]
    try :
        [ self . feature_names . remove ( x ) for x in self . drop_cols ]
    except KeyError as e :
        if self . verbose > 0 :
            print ( "Could not remove column from feature names." "Not found in generated cols.\n{}" . format ( e ) )
return self


signature:
def inverse_transform ( self , X_in ) :
    """Perform the inverse transformation to encoded data . Will attempt best case reconstruction , which means
 it will return nan for handle _ missing and handle _ unknown settings that break the bijection . We issue
 warnings when some of those cases occur .
 Parameters
 X _ in : array - like , shape = [ n _ samples , n _ features ]
 Returns
 p : array , the same size of X _ in"""

implementation:
X = X_in . copy ( deep = True )
# first check the type
X = util . convert_input ( X )
if self . _dim is None :
    raise ValueError ( 'Must train encoder before it can be used to inverse_transform data' )
# then make sure that it is the right size
if X . shape [ 1 ] != self . _dim :
    if self . drop_invariant :
        raise ValueError ( "Unexpected input dimension %d, the attribute drop_invariant should " "set as False when transform data" % ( X . shape [ 1 ] , ) )
    else :
        raise ValueError ( 'Unexpected input dimension %d, expected %d' % ( X . shape [ 1 ] , self . _dim , ) )
if not self . cols :
    return X if self . return_df else X . values
if self . handle_unknown == 'value' :
    for col in self . cols :
        if any ( X [ col ] == - 1 ) :
            warnings . warn ( "inverse_transform is not supported because transform impute " "the unknown category -1 when encode %s" % ( col , ) )
if self . handle_unknown == 'return_nan' and self . handle_missing == 'return_nan' :
    for col in self . cols :
        if X [ col ] . isnull ( ) . any ( ) :
            warnings . warn ( "inverse_transform is not supported because transform impute " "the unknown category nan when encode %s" % ( col , ) )
for switch in self . mapping :
    column_mapping = switch . get ( 'mapping' )
    inverse = pd . Series ( data = column_mapping . index , index = column_mapping . get_values ( ) )
    X [ switch . get ( 'col' ) ] = X [ switch . get ( 'col' ) ] . map ( inverse ) . astype ( switch . get ( 'data_type' ) )
return X if self . return_df else X . values


signature:
def ordinal_encoding ( X_in , mapping = None , cols = None , handle_unknown = 'value' , handle_missing = 'value' ) :
    """Ordinal encoding uses a single column of integers to represent the classes . An optional mapping dict can be passed
 in , in this case we use the knowledge that there is some true order to the classes themselves . Otherwise , the classes
 are assumed to have no true order and integers are selected at random ."""

implementation:
return_nan_series = pd . Series ( data = [ np . nan ] , index = [ - 2 ] )
X = X_in . copy ( deep = True )
if cols is None :
    cols = X . columns . values
if mapping is not None :
    mapping_out = mapping
    for switch in mapping :
        column = switch . get ( 'col' )
        X [ column ] = X [ column ] . map ( switch [ 'mapping' ] )
        try :
            X [ column ] = X [ column ] . astype ( int )
        except ValueError as e :
            X [ column ] = X [ column ] . astype ( float )
        if handle_unknown == 'value' :
            X [ column ] . fillna ( - 1 , inplace = True )
        elif handle_unknown == 'error' :
            missing = X [ column ] . isnull ( )
            if any ( missing ) :
                raise ValueError ( 'Unexpected categories found in column %s' % column )
        if handle_missing == 'return_nan' :
            X [ column ] = X [ column ] . map ( return_nan_series ) . where ( X [ column ] == - 2 , X [ column ] )
else :
    mapping_out = [ ]
    for col in cols :
        nan_identity = np . nan
        if util . is_category ( X [ col ] . dtype ) :
            categories = X [ col ] . cat . categories
        else :
            categories = X [ col ] . unique ( )
        index = pd . Series ( categories ) . fillna ( nan_identity ) . unique ( )
        data = pd . Series ( index = index , data = range ( 1 , len ( index ) + 1 ) )
        if handle_missing == 'value' and ~ data . index . isnull ( ) . any ( ) :
            data . loc [ nan_identity ] = - 2
        elif handle_missing == 'return_nan' :
            data . loc [ nan_identity ] = - 2
        mapping_out . append ( { 'col' : col , 'mapping' : data , 'data_type' : X [ col ] . dtype } , )
return X , mapping_out


signature:
def transform ( self , X , override_return_df = False ) :
    """Perform the transformation to new categorical data .
 Parameters
 X : array - like , shape = [ n _ samples , n _ features ]
 Returns
 p : array , shape = [ n _ samples , n _ numeric + N ]
 Transformed values with encoding applied ."""

implementation:
if self . handle_missing == 'error' :
    if X [ self . cols ] . isnull ( ) . any ( ) . bool ( ) :
        raise ValueError ( 'Columns to be encoded can not contain null' )
if self . _dim is None :
    raise ValueError ( 'Must train encoder before it can be used to transform data.' )
# first check the type
X = util . convert_input ( X )
# then make sure that it is the right size
if X . shape [ 1 ] != self . _dim :
    raise ValueError ( 'Unexpected input dimension %d, expected %d' % ( X . shape [ 1 ] , self . _dim , ) )
if not self . cols :
    return X if self . return_df else X . values
X = self . ordinal_encoder . transform ( X )
if self . handle_unknown == 'error' :
    if X [ self . cols ] . isin ( [ - 1 ] ) . any ( ) . any ( ) :
        raise ValueError ( 'Columns to be encoded can not contain new values' )
X = self . get_dummies ( X )
if self . drop_invariant :
    for col in self . drop_cols :
        X . drop ( col , 1 , inplace = True )
if self . return_df or override_return_df :
    return X
else :
    return X . values


signature:
def reverse_dummies ( self , X , mapping ) :
    """Convert dummy variable into numerical variables
 Parameters
 X : DataFrame
 mapping : list - like
 Contains mappings of column to be transformed to it ' s new columns and value represented
 Returns
 numerical : DataFrame"""

implementation:
out_cols = X . columns . values . tolist ( )
mapped_columns = [ ]
for switch in mapping :
    col = switch . get ( 'col' )
    mod = switch . get ( 'mapping' )
    insert_at = out_cols . index ( mod . columns [ 0 ] )
    X . insert ( insert_at , col , 0 )
    positive_indexes = mod . index [ mod . index > 0 ]
    for i in range ( positive_indexes . shape [ 0 ] ) :
        existing_col = mod . columns [ i ]
        val = positive_indexes [ i ]
        X . loc [ X [ existing_col ] == 1 , col ] = val
        mapped_columns . append ( existing_col )
    X . drop ( mod . columns , axis = 1 , inplace = True )
    out_cols = X . columns . values . tolist ( )
return X


signature:
def get_cars_data ( ) :
    """Load the cars dataset , split it into X and y , and then call the label encoder to get an integer y column .
 : return :"""

implementation:
df = pd . read_csv ( 'source_data/cars/car.data.txt' )
X = df . reindex ( columns = [ x for x in df . columns . values if x != 'class' ] )
y = df . reindex ( columns = [ 'class' ] )
y = preprocessing . LabelEncoder ( ) . fit_transform ( y . values . reshape ( - 1 , ) )
mapping = [ { 'col' : 'buying' , 'mapping' : [ ( 'vhigh' , 0 ) , ( 'high' , 1 ) , ( 'med' , 2 ) , ( 'low' , 3 ) ] } , { 'col' : 'maint' , 'mapping' : [ ( 'vhigh' , 0 ) , ( 'high' , 1 ) , ( 'med' , 2 ) , ( 'low' , 3 ) ] } , { 'col' : 'doors' , 'mapping' : [ ( '2' , 0 ) , ( '3' , 1 ) , ( '4' , 2 ) , ( '5more' , 3 ) ] } , { 'col' : 'persons' , 'mapping' : [ ( '2' , 0 ) , ( '4' , 1 ) , ( 'more' , 2 ) ] } , { 'col' : 'lug_boot' , 'mapping' : [ ( 'small' , 0 ) , ( 'med' , 1 ) , ( 'big' , 2 ) ] } , { 'col' : 'safety' , 'mapping' : [ ( 'high' , 0 ) , ( 'med' , 1 ) , ( 'low' , 2 ) ] } , ]
return X , y , mapping


signature:
def get_splice_data ( ) :
    """Load the mushroom dataset , split it into X and y , and then call the label encoder to get an integer y column .
 : return :"""

implementation:
df = pd . read_csv ( 'source_data/splice/splice.csv' )
X = df . reindex ( columns = [ x for x in df . columns . values if x != 'class' ] )
X [ 'dna' ] = X [ 'dna' ] . map ( lambda x : list ( str ( x ) . strip ( ) ) )
for idx in range ( 60 ) :
    X [ 'dna_%d' % ( idx , ) ] = X [ 'dna' ] . map ( lambda x : x [ idx ] )
del X [ 'dna' ]
y = df . reindex ( columns = [ 'class' ] )
y = preprocessing . LabelEncoder ( ) . fit_transform ( y . values . reshape ( - 1 , ) )
# this data is truly categorical , with no known concept of ordering
mapping = None
return X , y , mapping


signature:
def basen_to_integer ( self , X , cols , base ) :
    """Convert basen code as integers .
 Parameters
 X : DataFrame
 encoded data
 cols : list - like
 Column names in the DataFrame that be encoded
 base : int
 The base of transform
 Returns
 numerical : DataFrame"""

implementation:
out_cols = X . columns . values . tolist ( )
for col in cols :
    col_list = [ col0 for col0 in out_cols if str ( col0 ) . startswith ( str ( col ) ) ]
    insert_at = out_cols . index ( col_list [ 0 ] )
    if base == 1 :
        value_array = np . array ( [ int ( col0 . split ( '_' ) [ - 1 ] ) for col0 in col_list ] )
    else :
        len0 = len ( col_list )
        value_array = np . array ( [ base ** ( len0 - 1 - i ) for i in range ( len0 ) ] )
    X . insert ( insert_at , col , np . dot ( X [ col_list ] . values , value_array . T ) )
    X . drop ( col_list , axis = 1 , inplace = True )
    out_cols = X . columns . values . tolist ( )
return X


signature:
def col_transform ( self , col , digits ) :
    """The lambda body to transform the column values"""

implementation:
if col is None or float ( col ) < 0.0 :
    return None
else :
    col = self . number_to_base ( int ( col ) , self . base , digits )
    if len ( col ) == digits :
        return col
    else :
        return [ 0 for _ in range ( digits - len ( col ) ) ] + col


implementation:
# first check the type
X = util . convert_input ( X )
self . _dim = X . shape [ 1 ]
# if columns aren ' t passed , just use every string column
if self . cols is None :
    self . cols = util . get_obj_cols ( X )
else :
    self . cols = util . convert_cols_to_list ( self . cols )
X_temp = self . transform ( X , override_return_df = True )
self . feature_names = X_temp . columns . tolist ( )
# drop all output columns with 0 variance .
if self . drop_invariant :
    self . drop_cols = [ ]
    generated_cols = util . get_generated_cols ( X , X_temp , self . cols )
    self . drop_cols = [ x for x in generated_cols if X_temp [ x ] . var ( ) <= 10e-5 ]
    try :
        [ self . feature_names . remove ( x ) for x in self . drop_cols ]
    except KeyError as e :
        if self . verbose > 0 :
            print ( "Could not remove column from feature names." "Not found in generated cols.\n{}" . format ( e ) )
return self


implementation:
if self . _dim is None :
    raise ValueError ( 'Must train encoder before it can be used to transform data.' )
# first check the type
X = util . convert_input ( X )
# then make sure that it is the right size
if X . shape [ 1 ] != self . _dim :
    raise ValueError ( 'Unexpected input dimension %d, expected %d' % ( X . shape [ 1 ] , self . _dim , ) )
if not self . cols :
    return X
X = self . hashing_trick ( X , hashing_method = self . hash_method , N = self . n_components , cols = self . cols )
if self . drop_invariant :
    for col in self . drop_cols :
        X . drop ( col , 1 , inplace = True )
if self . return_df or override_return_df :
    return X
else :
    return X . values


signature:
def hashing_trick ( X_in , hashing_method = 'md5' , N = 2 , cols = None , make_copy = False ) :
    """A basic hashing implementation with configurable dimensionality / precision
 Performs the hashing trick on a pandas dataframe , ` X ` , using the hashing method from hashlib
 identified by ` hashing _ method ` . The number of output dimensions ( ` N ` ) , and columns to hash ( ` cols ` ) are
 also configurable .
 Parameters
 X _ in : pandas dataframe
 description text
 hashing _ method : string , optional
 description text
 N : int , optional
 description text
 cols : list , optional
 description text
 make _ copy : bool , optional
 description text
 Returns
 out : dataframe
 A hashing encoded dataframe .
 References
 Cite the relevant literature , e . g . [ 1 ] _ . You may also cite these
 references in the notes section above .
 . . [ 1 ] Kilian Weinberger ; Anirban Dasgupta ; John Langford ; Alex Smola ; Josh Attenberg ( 2009 ) . Feature Hashing
 for Large Scale Multitask Learning . Proc . ICML ."""

implementation:
try :
    if hashing_method not in hashlib . algorithms_available :
        raise ValueError ( 'Hashing Method: %s Not Available. Please use one from: [%s]' % ( hashing_method , ', ' . join ( [ str ( x ) for x in hashlib . algorithms_available ] ) ) )
except Exception as e :
    try :
        _ = hashlib . new ( hashing_method )
    except Exception as e :
        raise ValueError ( 'Hashing Method: %s Not Found.' )
if make_copy :
    X = X_in . copy ( deep = True )
else :
    X = X_in
if cols is None :
    cols = X . columns . values
def hash_fn ( x ) :
    tmp = [ 0 for _ in range ( N ) ]
    for val in x . values :
        if val is not None :
            hasher = hashlib . new ( hashing_method )
            if sys . version_info [ 0 ] == 2 :
                hasher . update ( str ( val ) )
            else :
                hasher . update ( bytes ( str ( val ) , 'utf-8' ) )
            tmp [ int ( hasher . hexdigest ( ) , 16 ) % N ] += 1
    return pd . Series ( tmp , index = new_cols )
new_cols = [ 'col_%d' % d for d in range ( N ) ]
X_cat = X . loc [ : , cols ]
X_num = X . loc [ : , [ x for x in X . columns . values if x not in cols ] ]
X_cat = X_cat . apply ( hash_fn , axis = 1 )
X_cat . columns = new_cols
X = pd . concat ( [ X_cat , X_num ] , axis = 1 )
return X


signature:
def transform ( self , X , y = None , override_return_df = False ) :
    """Perform the transformation to new categorical data .
 Parameters
 X : array - like , shape = [ n _ samples , n _ features ]
 y : array - like , shape = [ n _ samples ] when transform by leave one out
 None , when transform without target information ( such as transform test set )
 Returns
 p : array , shape = [ n _ samples , n _ numeric + N ]
 Transformed values with encoding applied ."""

implementation:
if self . handle_missing == 'error' :
    if X [ self . cols ] . isnull ( ) . any ( ) . bool ( ) :
        raise ValueError ( 'Columns to be encoded can not contain null' )
if self . _dim is None :
    raise ValueError ( 'Must train encoder before it can be used to transform data.' )
# unite the input into pandas types
X = util . convert_input ( X )
# then make sure that it is the right size
if X . shape [ 1 ] != self . _dim :
    raise ValueError ( 'Unexpected input dimension %d, expected %d' % ( X . shape [ 1 ] , self . _dim , ) )
# if we are encoding the training data , we have to check the target
if y is not None :
    y = util . convert_input_vector ( y , X . index ) . astype ( float )
    if X . shape [ 0 ] != y . shape [ 0 ] :
        raise ValueError ( "The length of X is " + str ( X . shape [ 0 ] ) + " but length of y is " + str ( y . shape [ 0 ] ) + "." )
if not self . cols :
    return X
X = self . transform_leave_one_out ( X , y , mapping = self . mapping )
if self . drop_invariant :
    for col in self . drop_cols :
        X . drop ( col , 1 , inplace = True )
if self . return_df or override_return_df :
    return X
else :
    return X . values


signature:
def transform_leave_one_out ( self , X_in , y , mapping = None ) :
    """Leave one out encoding uses a single column of floats to represent the means of the target variables ."""

implementation:
X = X_in . copy ( deep = True )
random_state_ = check_random_state ( self . random_state )
# Prepare the data
if y is not None : # Convert bools to numbers ( the target must be summable )
    y = y . astype ( 'double' )
    # Cumsum and cumcount do not work nicely with None .
    # This is a terrible workaround that will fail , when the
    # categorical input contains - 999.9
    for cat_col in X . select_dtypes ( 'category' ) . columns . values :
        X [ cat_col ] = X [ cat_col ] . cat . add_categories ( - 999.9 )
    X = X . fillna ( - 999.9 )
for col , colmap in mapping . items ( ) :
    level_notunique = colmap [ 'count' ] > 1
    unique_train = colmap . index
    unseen_values = pd . Series ( [ x for x in X_in [ col ] . unique ( ) if x not in unique_train ] )
    is_nan = X_in [ col ] . isnull ( )
    is_unknown_value = X_in [ col ] . isin ( unseen_values . dropna ( ) )
    if self . handle_unknown == 'error' and is_unknown_value . any ( ) :
        raise ValueError ( 'Columns to be encoded can not contain new values' )
    if y is None : # Replace level with its mean target ; if level occurs only once , use global mean
        level_means = ( ( colmap [ 'sum' ] + self . _mean ) / ( colmap [ 'count' ] + 1 ) ) . where ( level_notunique , self . _mean )
        X [ col ] = X [ col ] . map ( level_means )
    else : # Simulation of CatBoost implementation , which calculates leave - one - out on the fly .
    # The nice thing about this is that it helps to prevent overfitting . The bad thing
    # is that CatBoost uses many iterations over the data . But we run just one iteration .
    # Still , it works better than leave - one - out without any noise .
    # See :
    # https : / / tech . yandex . com / catboost / doc / dg / concepts / algorithm - main - stages _ cat - to - numberic - docpage /
        temp = y . groupby ( X [ col ] ) . agg ( [ 'cumsum' , 'cumcount' ] )
        X [ col ] = ( temp [ 'cumsum' ] - y + self . _mean ) / ( temp [ 'cumcount' ] + 1 )
    if self . handle_unknown == 'value' :
        X . loc [ is_unknown_value , col ] = self . _mean
    elif self . handle_unknown == 'return_nan' :
        X . loc [ is_unknown_value , col ] = np . nan
    if self . handle_missing == 'value' :
        X . loc [ is_nan & unseen_values . isnull ( ) . any ( ) , col ] = self . _mean
    elif self . handle_missing == 'return_nan' :
        X . loc [ is_nan , col ] = np . nan
    if self . sigma is not None and y is not None :
        X [ col ] = X [ col ] * random_state_ . normal ( 1. , self . sigma , X [ col ] . shape [ 0 ] )
return X


signature:
def get_obj_cols ( df ) :
    """Returns names of ' object ' columns in the DataFrame ."""

implementation:
obj_cols = [ ]
for idx , dt in enumerate ( df . dtypes ) :
    if dt == 'object' or is_category ( dt ) :
        obj_cols . append ( df . columns . values [ idx ] )
return obj_cols


signature:
def convert_input ( X ) :
    """Unite data into a DataFrame ."""

implementation:
if not isinstance ( X , pd . DataFrame ) :
    if isinstance ( X , list ) :
        X = pd . DataFrame ( X )
    elif isinstance ( X , ( np . generic , np . ndarray ) ) :
        X = pd . DataFrame ( X )
    elif isinstance ( X , csr_matrix ) :
        X = pd . DataFrame ( X . todense ( ) )
    elif isinstance ( X , pd . Series ) :
        X = pd . DataFrame ( X )
    else :
        raise ValueError ( 'Unexpected input type: %s' % ( str ( type ( X ) ) ) )
    X = X . apply ( lambda x : pd . to_numeric ( x , errors = 'ignore' ) )
return X


signature:
def convert_input_vector ( y , index ) :
    """Unite target data type into a Series .
 If the target is a Series or a DataFrame , we preserve its index .
 But if the target does not contain index attribute , we use the index from the argument ."""

implementation:
if y is None :
    return None
if isinstance ( y , pd . Series ) :
    return y
elif isinstance ( y , np . ndarray ) :
    if len ( np . shape ( y ) ) == 1 : # vector
        return pd . Series ( y , name = 'target' , index = index )
    elif len ( np . shape ( y ) ) == 2 and np . shape ( y ) [ 0 ] == 1 : # single row in a matrix
        return pd . Series ( y [ 0 , : ] , name = 'target' , index = index )
    elif len ( np . shape ( y ) ) == 2 and np . shape ( y ) [ 1 ] == 1 : # single column in a matrix
        return pd . Series ( y [ : , 0 ] , name = 'target' , index = index )
    else :
        raise ValueError ( 'Unexpected input shape: %s' % ( str ( np . shape ( y ) ) ) )
elif np . isscalar ( y ) :
    return pd . Series ( [ y ] , name = 'target' , index = index )
elif isinstance ( y , list ) :
    if len ( y ) == 0 or ( len ( y ) > 0 and not isinstance ( y [ 0 ] , list ) ) : # empty list or a vector
        return pd . Series ( y , name = 'target' , index = index )
    elif len ( y ) > 0 and isinstance ( y [ 0 ] , list ) and len ( y [ 0 ] ) == 1 : # single row in a matrix
        flatten = lambda y : [ item for sublist in y for item in sublist ]
        return pd . Series ( flatten ( y ) , name = 'target' , index = index )
    elif len ( y ) == 1 and isinstance ( y [ 0 ] , list ) : # single column in a matrix
        return pd . Series ( y [ 0 ] , name = 'target' , index = index )
    else :
        raise ValueError ( 'Unexpected input shape' )
elif isinstance ( y , pd . DataFrame ) :
    if len ( list ( y ) ) == 0 : # empty DataFrame
        return pd . Series ( y , name = 'target' )
    if len ( list ( y ) ) == 1 : # a single column
        return y . iloc [ : , 0 ]
    else :
        raise ValueError ( 'Unexpected input shape: %s' % ( str ( y . shape ) ) )
else :
    return pd . Series ( y , name = 'target' , index = index )


signature:
def get_generated_cols ( X_original , X_transformed , to_transform ) :
    """Returns a list of the generated / transformed columns .
 Arguments :
 X _ original : df
 the original ( input ) DataFrame .
 X _ transformed : df
 the transformed ( current ) DataFrame .
 to _ transform : [ str ]
 a list of columns that were transformed ( as in the original DataFrame ) , commonly self . cols .
 Output :
 a list of columns that were transformed ( as in the current DataFrame ) ."""

implementation:
original_cols = list ( X_original . columns )
if len ( to_transform ) > 0 :
    [ original_cols . remove ( c ) for c in to_transform ]
current_cols = list ( X_transformed . columns )
if len ( original_cols ) > 0 :
    [ current_cols . remove ( c ) for c in original_cols ]
return current_cols


signature:
def fit ( self , X , y , ** kwargs ) :
    """Fit encoder according to X and y .
 Parameters
 X : array - like , shape = [ n _ samples , n _ features ]
 Training vectors , where n _ samples is the number of samples
 and n _ features is the number of features .
 y : array - like , shape = [ n _ samples ]
 Target values .
 Returns
 self : encoder
 Returns self ."""

implementation:
# unite the input into pandas types
X = util . convert_input ( X )
y = util . convert_input_vector ( y , X . index )
if X . shape [ 0 ] != y . shape [ 0 ] :
    raise ValueError ( "The length of X is " + str ( X . shape [ 0 ] ) + " but length of y is " + str ( y . shape [ 0 ] ) + "." )
self . _dim = X . shape [ 1 ]
# if columns aren ' t passed , just use every string column
if self . cols is None :
    self . cols = util . get_obj_cols ( X )
else :
    self . cols = util . convert_cols_to_list ( self . cols )
if self . handle_missing == 'error' :
    if X [ self . cols ] . isnull ( ) . any ( ) . bool ( ) :
        raise ValueError ( 'Columns to be encoded can not contain null' )
self . ordinal_encoder = OrdinalEncoder ( verbose = self . verbose , cols = self . cols , handle_unknown = 'value' , handle_missing = 'value' )
self . ordinal_encoder = self . ordinal_encoder . fit ( X )
X_ordinal = self . ordinal_encoder . transform ( X )
self . mapping = self . fit_target_encoding ( X_ordinal , y )
X_temp = self . transform ( X , override_return_df = True )
self . feature_names = list ( X_temp . columns )
if self . drop_invariant :
    self . drop_cols = [ ]
    X_temp = self . transform ( X )
    generated_cols = util . get_generated_cols ( X , X_temp , self . cols )
    self . drop_cols = [ x for x in generated_cols if X_temp [ x ] . var ( ) <= 10e-5 ]
    try :
        [ self . feature_names . remove ( x ) for x in self . drop_cols ]
    except KeyError as e :
        if self . verbose > 0 :
            print ( "Could not remove column from feature names." "Not found in generated cols.\n{}" . format ( e ) )
return self


implementation:
X = X_in . copy ( deep = True )
random_state_ = check_random_state ( self . random_state )
for col , colmap in mapping . items ( ) :
    level_notunique = colmap [ 'count' ] > 1
    unique_train = colmap . index
    unseen_values = pd . Series ( [ x for x in X [ col ] . unique ( ) if x not in unique_train ] )
    is_nan = X [ col ] . isnull ( )
    is_unknown_value = X [ col ] . isin ( unseen_values . dropna ( ) )
    if self . handle_unknown == 'error' and is_unknown_value . any ( ) :
        raise ValueError ( 'Columns to be encoded can not contain new values' )
    if y is None : # Replace level with its mean target ; if level occurs only once , use global mean
        level_means = ( colmap [ 'sum' ] / colmap [ 'count' ] ) . where ( level_notunique , self . _mean )
        X [ col ] = X [ col ] . map ( level_means )
    else : # Replace level with its mean target , calculated excluding this row ' s target
    # The y ( target ) mean for this level is normally just the sum / count ;
    # excluding this row ' s y , it ' s ( sum - y ) / ( count - 1)
        level_means = ( X [ col ] . map ( colmap [ 'sum' ] ) - y ) / ( X [ col ] . map ( colmap [ 'count' ] ) - 1 )
        # The ' where ' fills in singleton levels ( count = 1 - > div by 0 ) with the global mean
        X [ col ] = level_means . where ( X [ col ] . map ( colmap [ 'count' ] [ level_notunique ] ) . notnull ( ) , self . _mean )
    if self . handle_unknown == 'value' :
        X . loc [ is_unknown_value , col ] = self . _mean
    elif self . handle_unknown == 'return_nan' :
        X . loc [ is_unknown_value , col ] = np . nan
    if self . handle_missing == 'value' :
        X . loc [ is_nan & unseen_values . isnull ( ) . any ( ) , col ] = self . _mean
    elif self . handle_missing == 'return_nan' :
        X . loc [ is_nan , col ] = np . nan
    if self . sigma is not None and y is not None :
        X [ col ] = X [ col ] * random_state_ . normal ( 1. , self . sigma , X [ col ] . shape [ 0 ] )
return X


signature:
def score_models ( clf , X , y , encoder , runs = 1 ) :
    """Takes in a classifier that supports multiclass classification , and X and a y , and returns a cross validation score ."""

implementation:
scores = [ ]
X_test = None
for _ in range ( runs ) :
    X_test = encoder ( ) . fit_transform ( X , y )
    # Some models , like logistic regression , like normalized features otherwise they underperform and / or take a long time to converge .
    # To be rigorous , we should have trained the normalization on each fold individually via pipelines .
    # See grid _ search _ example to learn how to do it .
    X_test = StandardScaler ( ) . fit_transform ( X_test )
    scores . append ( cross_validate ( clf , X_test , y , n_jobs = 1 , cv = 5 ) [ 'test_score' ] )
    gc . collect ( )
scores = [ y for z in [ x for x in scores ] for y in z ]
return float ( np . mean ( scores ) ) , float ( np . std ( scores ) ) , scores , X_test . shape [ 1 ]


signature:
def main ( loader , name ) :
    """Here we iterate through the datasets and score them with a classifier using different encodings ."""

implementation:
scores = [ ]
raw_scores_ds = { }
# first get the dataset
X , y , mapping = loader ( )
clf = linear_model . LogisticRegression ( solver = 'lbfgs' , multi_class = 'auto' , max_iter = 200 , random_state = 0 )
# try each encoding method available , which works on multiclass problems
encoders = ( set ( category_encoders . __all__ ) - { 'WOEEncoder' } )
# WoE is currently only for binary targets
for encoder_name in encoders :
    encoder = getattr ( category_encoders , encoder_name )
    start_time = time . time ( )
    score , stds , raw_scores , dim = score_models ( clf , X , y , encoder )
    scores . append ( [ encoder_name , name , dim , score , stds , time . time ( ) - start_time ] )
    raw_scores_ds [ encoder_name ] = raw_scores
    gc . collect ( )
results = pd . DataFrame ( scores , columns = [ 'Encoding' , 'Dataset' , 'Dimensionality' , 'Avg. Score' , 'Score StDev' , 'Elapsed Time' ] )
raw = pd . DataFrame . from_dict ( raw_scores_ds )
ax = raw . plot ( kind = 'box' , return_type = 'axes' )
plt . title ( 'Scores for Encodings on %s Dataset' % ( name , ) )
plt . ylabel ( 'Score (higher is better)' )
for tick in ax . get_xticklabels ( ) :
    tick . set_rotation ( 90 )
plt . grid ( )
plt . tight_layout ( )
plt . show ( )
return results , raw


signature:
def secho ( message , ** kwargs ) :
    """A wrapper around click . secho that disables any coloring being used
 if colors have been disabled ."""

implementation:
# If colors are disabled , remove any color or other style data
# from keyword arguments .
if not settings . color :
    for key in ( 'fg' , 'bg' , 'bold' , 'blink' ) :
        kwargs . pop ( key , None )
# Okay , now call click . secho normally .
return click . secho ( message , ** kwargs )


signature:
def associate_notification_template ( self , job_template , notification_template , status ) :
    """Associate a notification template from this job template .
 = = = = = API DOCS = = = = =
 Associate a notification template from this job template .
 : param job _ template : The job template to associate to .
 : type job _ template : str
 : param notification _ template : The notification template to be associated .
 : type notification _ template : str
 : param status : type of notification this notification template should be associated to .
 : type status : str
 : returns : Dictionary of only one key " changed " , which indicates whether the association succeeded .
 : rtype : dict
 = = = = = API DOCS = = = = ="""

implementation: return self . _assoc ( 'notification_templates_%s' % status , job_template , notification_template )


signature:
def disassociate_notification_template ( self , job_template , notification_template , status ) :
    """Disassociate a notification template from this job template .
 = = = = = API DOCS = = = = =
 Disassociate a notification template from this job template .
 : param job _ template : The job template to disassociate from .
 : type job _ template : str
 : param notification _ template : The notification template to be disassociated .
 : type notification _ template : str
 : param status : type of notification this notification template should be disassociated from .
 : type status : str
 : returns : Dictionary of only one key " changed " , which indicates whether the disassociation succeeded .
 : rtype : dict
 = = = = = API DOCS = = = = ="""

implementation: return self . _disassoc ( 'notification_templates_%s' % status , job_template , notification_template )


signature:
def callback ( self , pk = None , host_config_key = '' , extra_vars = None ) :
    """Contact Tower and request a configuration update using this job template .
 = = = = = API DOCS = = = = =
 Contact Tower and request a provisioning callback using this job template .
 : param pk : Primary key of the job template to run provisioning callback against .
 : type pk : int
 : param host _ config _ key : Key string used to authenticate the callback host .
 : type host _ config _ key : str
 : param extra _ vars : Extra variables that are passed to provisioning callback .
 : type extra _ vars : array of str
 : returns : A dictionary of a single key " changed " , which indicates whether the provisioning callback
 is successful .
 : rtype : dict
 = = = = = API DOCS = = = = ="""

implementation:
url = self . endpoint + '%s/callback/' % pk
if not host_config_key :
    host_config_key = client . get ( url ) . json ( ) [ 'host_config_key' ]
post_data = { 'host_config_key' : host_config_key }
if extra_vars :
    post_data [ 'extra_vars' ] = parser . process_extra_vars ( list ( extra_vars ) , force_json = True )
r = client . post ( url , data = post_data , auth = None )
if r . status_code == 201 :
    return { 'changed' : True }


signature:
def lookup_stdout ( self , pk = None , start_line = None , end_line = None , full = True ) :
    """Internal method that lies to our ` monitor ` method by returning
 a scorecard for the workflow job where the standard out
 would have been expected ."""

implementation:
uj_res = get_resource ( 'unified_job' )
# Filters
# - limit search to jobs spawned as part of this workflow job
# - order in the order in which they should add to the list
# - only include final job states
query_params = ( ( 'unified_job_node__workflow_job' , pk ) , ( 'order_by' , 'finished' ) , ( 'status__in' , 'successful,failed,error' ) )
jobs_list = uj_res . list ( all_pages = True , query = query_params )
if jobs_list [ 'count' ] == 0 :
    return ''
return_content = ResSubcommand ( uj_res ) . _format_human ( jobs_list )
lines = return_content . split ( '\n' )
if not full :
    lines = lines [ : - 1 ]
N = len ( lines )
start_range = start_line
if start_line is None :
    start_range = 0
elif start_line > N :
    start_range = N
end_range = end_line
if end_line is None or end_line > N :
    end_range = N
lines = lines [ start_range : end_range ]
return_content = '\n' . join ( lines )
if len ( lines ) > 0 :
    return_content += '\n'
return return_content


signature:
def launch ( self , workflow_job_template = None , monitor = False , wait = False , timeout = None , extra_vars = None , ** kwargs ) :
    """Launch a new workflow job based on a workflow job template .
 Creates a new workflow job in Ansible Tower , starts it , and
 returns back an ID in order for its status to be monitored .
 = = = = = API DOCS = = = = =
 Launch a new workflow job based on a workflow job template .
 : param workflow _ job _ template : Primary key or name of the workflow job template to launch new job .
 : type workflow _ job _ template : str
 : param monitor : Flag that if set , immediately calls ` ` monitor ` ` on the newly launched workflow job rather
 than exiting with a success .
 : type monitor : bool
 : param wait : Flag that if set , monitor the status of the workflow job , but do not print while job is
 in progress .
 : type wait : bool
 : param timeout : If provided with ` ` monitor ` ` flag set , this attempt will time out after the given number
 of seconds .
 : type timeout : int
 : param extra _ vars : yaml formatted texts that contains extra variables to pass on .
 : type extra _ vars : array of strings
 : param ` * * kwargs ` : Fields needed to create and launch a workflow job .
 : returns : Result of subsequent ` ` monitor ` ` call if ` ` monitor ` ` flag is on ; Result of subsequent ` ` wait ` `
 call if ` ` wait ` ` flag is on ; loaded JSON output of the job launch if none of the two flags are on .
 : rtype : dict
 = = = = = API DOCS = = = = ="""

implementation:
if extra_vars is not None and len ( extra_vars ) > 0 :
    kwargs [ 'extra_vars' ] = parser . process_extra_vars ( extra_vars )
debug . log ( 'Launching the workflow job.' , header = 'details' )
self . _pop_none ( kwargs )
post_response = client . post ( 'workflow_job_templates/{0}/launch/' . format ( workflow_job_template ) , data = kwargs ) . json ( )
workflow_job_id = post_response [ 'id' ]
post_response [ 'changed' ] = True
if monitor :
    return self . monitor ( workflow_job_id , timeout = timeout )
elif wait :
    return self . wait ( workflow_job_id , timeout = timeout )
return post_response


signature:
def parse_args ( self , ctx , args ) :
    """Parse arguments sent to this command .
 The code for this method is taken from MultiCommand :
 https : / / github . com / mitsuhiko / click / blob / master / click / core . py
 It is Copyright ( c ) 2014 by Armin Ronacher .
 See the license :
 https : / / github . com / mitsuhiko / click / blob / master / LICENSE"""

implementation:
if not args and self . no_args_is_help and not ctx . resilient_parsing :
    click . echo ( ctx . get_help ( ) )
    ctx . exit ( )
return super ( ActionSubcommand , self ) . parse_args ( ctx , args )


signature:
def format_options ( self , ctx , formatter ) :
    """Monkey - patch click ' s format _ options method to support option categorization ."""

implementation:
field_opts = [ ]
global_opts = [ ]
local_opts = [ ]
other_opts = [ ]
for param in self . params :
    if param . name in SETTINGS_PARMS :
        opts = global_opts
    elif getattr ( param , 'help' , None ) and param . help . startswith ( '[FIELD]' ) :
        opts = field_opts
        param . help = param . help [ len ( '[FIELD]' ) : ]
    else :
        opts = local_opts
    rv = param . get_help_record ( ctx )
    if rv is None :
        continue
    else :
        opts . append ( rv )
if self . add_help_option :
    help_options = self . get_help_option_names ( ctx )
    if help_options :
        other_opts . append ( [ join_options ( help_options ) [ 0 ] , 'Show this message and exit.' ] )
if field_opts :
    with formatter . section ( 'Field Options' ) :
        formatter . write_dl ( field_opts )
if local_opts :
    with formatter . section ( 'Local Options' ) :
        formatter . write_dl ( local_opts )
if global_opts :
    with formatter . section ( 'Global Options' ) :
        formatter . write_dl ( global_opts )
if other_opts :
    with formatter . section ( 'Other Options' ) :
        formatter . write_dl ( other_opts )


signature:
def list ( self , ** kwargs ) :
    """Return a list of objects .
 = = = = = API DOCS = = = = =
 Retrieve a list of Tower settings .
 : param category : The category slug in which to look up indevidual settings .
 : type category : str
 : param ` * * kwargs ` : Keyword arguments list of available fields used for searching resource objects .
 : returns : A JSON object containing details of all resource objects returned by Tower backend .
 : rtype : dict
 = = = = = API DOCS = = = = ="""

implementation:
self . custom_category = kwargs . get ( 'category' , 'all' )
try :
    result = super ( Resource , self ) . list ( ** kwargs )
except exc . NotFound as e :
    categories = map ( lambda category : category [ 'slug' ] , client . get ( '/settings/' ) . json ( ) [ 'results' ] )
    e . message = '%s is not a valid category.  Choose from [%s]' % ( kwargs [ 'category' ] , ', ' . join ( categories ) )
    raise e
finally :
    self . custom_category = None
return { 'results' : [ { 'id' : k , 'value' : v } for k , v in result . items ( ) ] }


signature:
def get ( self , pk ) :
    """Return one and exactly one object
 = = = = = API DOCS = = = = =
 Return one and exactly one Tower setting .
 : param pk : Primary key of the Tower setting to retrieve
 : type pk : int
 : returns : loaded JSON of the retrieved Tower setting object .
 : rtype : dict
 : raises tower _ cli . exceptions . NotFound : When no specified Tower setting exists .
 = = = = = API DOCS = = = = ="""

implementation:
# The Tower API doesn ' t provide a mechanism for retrieving a single
# setting value at a time , so fetch them all and filter
try :
    return next ( s for s in self . list ( ) [ 'results' ] if s [ 'id' ] == pk )
except StopIteration :
    raise exc . NotFound ( 'The requested object could not be found.' )


signature:
def modify ( self , setting , value ) :
    """Modify an already existing object .
 Positional argument SETTING is the setting name and VALUE is its value ,
 which can be provided directly or obtained from a file name if prefixed with ' @ ' .
 = = = = = API DOCS = = = = =
 Modify an already existing Tower setting .
 : param setting : The name of the Tower setting to be modified .
 : type setting : str
 : param value : The new value of the Tower setting .
 : type value : str
 : returns : A dictionary combining the JSON output of the modified resource , as well as two extra fields :
 " changed " , a flag indicating if the resource is successfully updated ; " id " , an integer which
 is the primary key of the updated object .
 : rtype : dict
 = = = = = API DOCS = = = = ="""

implementation:
prev_value = new_value = self . get ( setting ) [ 'value' ]
answer = OrderedDict ( )
encrypted = '$encrypted$' in six . text_type ( prev_value )
if encrypted or six . text_type ( prev_value ) != six . text_type ( value ) :
    if setting == 'LICENSE' :
        r = client . post ( '/config/' , data = self . coerce_type ( setting , value ) )
        new_value = r . json ( )
    else :
        r = client . patch ( self . endpoint , data = { setting : self . coerce_type ( setting , value ) } )
        new_value = r . json ( ) [ setting ]
    answer . update ( r . json ( ) )
changed = encrypted or ( prev_value != new_value )
answer . update ( { 'changed' : changed , 'id' : setting , 'value' : new_value , } )
return answer


signature:
def _pop_none ( self , kwargs ) :
    """Remove default values ( anything where the value is None ) . click is unfortunately bad at the way it
 sends through unspecified defaults ."""

implementation:
for key , value in copy ( kwargs ) . items ( ) : # options with multiple = True return a tuple
    if value is None or value == ( ) :
        kwargs . pop ( key )
    if hasattr ( value , 'read' ) :
        kwargs [ key ] = value . read ( )


signature:
def _lookup ( self , fail_on_missing = False , fail_on_found = False , include_debug_header = True , ** kwargs ) :
    """= = = = = API DOCS = = = = =
 Attempt to perform a lookup that is expected to return a single result , and return the record .
 This method is a wrapper around ` get ` that strips out non - unique keys , and is used internally by
 ` write ` and ` delete ` .
 : param fail _ on _ missing : Flag that raise exception if no resource is found .
 : type fail _ on _ missing : bool
 : param fail _ on _ found : Flag that raise exception if a resource is found .
 : type fail _ on _ found : bool
 : param include _ debug _ header : Flag determining whether to print debug messages when querying
 Tower backend .
 : type include _ debug _ header : bool
 : param ` * * kwargs ` : Keyword arguments list of available fields used for searching resource .
 : returns : A JSON object containing details of the resource returned by Tower backend .
 : rtype : dict
 : raises tower _ cli . exceptions . BadRequest : When no field are provided in kwargs .
 : raises tower _ cli . exceptions . Found : When a resource is found and fail _ on _ found flag is on .
 : raises tower _ cli . exceptions . NotFound : When no resource is found and fail _ on _ missing flag
 is on .
 = = = = = API DOCS = = = = ="""

implementation:
read_params = { }
for field_name in self . identity :
    if field_name in kwargs :
        read_params [ field_name ] = kwargs [ field_name ]
if 'id' in self . identity and len ( self . identity ) == 1 :
    return { }
if not read_params :
    raise exc . BadRequest ( 'Cannot reliably determine which record to write. Include an ID or unique ' 'fields.' )
try :
    existing_data = self . get ( include_debug_header = include_debug_header , ** read_params )
    if fail_on_found :
        raise exc . Found ( 'A record matching %s already exists, and you requested a failure in that case.' % read_params )
    return existing_data
except exc . NotFound :
    if fail_on_missing :
        raise exc . NotFound ( 'A record matching %s does not exist, and you requested a failure in that case.' % read_params )
    return { }


signature:
def _convert_pagenum ( self , kwargs ) :
    """Convert next and previous from URLs to integers"""

implementation:
for key in ( 'next' , 'previous' ) :
    if not kwargs . get ( key ) :
        continue
    match = re . search ( r'page=(?P<num>[\d]+)' , kwargs [ key ] )
    if match is None and key == 'previous' :
        kwargs [ key ] = 1
        continue
    kwargs [ key ] = int ( match . groupdict ( ) [ 'num' ] )


signature:
def read ( self , pk = None , fail_on_no_results = False , fail_on_multiple_results = False , ** kwargs ) :
    """= = = = = API DOCS = = = = =
 Retrieve and return objects from the Ansible Tower API .
 : param pk : Primary key of the resource to be read . Tower CLI will only attempt to read that object
 if ` ` pk ` ` is provided ( not ` ` None ` ` ) .
 : type pk : int
 : param fail _ on _ no _ results : Flag that if set , zero results is considered a failure case and raises
 an exception ; otherwise , empty list is returned . ( Note : This is always True
 if a primary key is included . )
 : type fail _ on _ no _ results : bool
 : param fail _ on _ multiple _ results : Flag that if set , at most one result is expected , and more results
 constitutes a failure case . ( Note : This is meaningless if a primary
 key is included , as there can never be multiple results . )
 : type fail _ on _ multiple _ results : bool
 : param query : Contains 2 - tuples used as query parameters to filter resulting resource objects .
 : type query : list
 : param ` * * kwargs ` : Keyword arguments which , all together , will be used as query parameters to filter
 resulting resource objects .
 : returns : loaded JSON from Tower backend response body .
 : rtype : dict
 : raises tower _ cli . exceptions . BadRequest : When 2 - tuples in ` ` query ` ` overlaps key - value pairs in
 ` ` * * kwargs ` ` .
 : raises tower _ cli . exceptions . NotFound : When no objects are found and ` ` fail _ on _ no _ results ` ` flag is on .
 : raises tower _ cli . exceptions . MultipleResults : When multiple objects are found and
 ` ` fail _ on _ multiple _ results ` ` flag is on .
 = = = = = API DOCS = = = = ="""

implementation:
# Piece together the URL we will be hitting .
url = self . endpoint
if pk :
    url += '%s/' % pk
# Pop the query parameter off of the keyword arguments ; it will
# require special handling ( below ) .
queries = kwargs . pop ( 'query' , [ ] )
# Remove default values ( anything where the value is None ) .
self . _pop_none ( kwargs )
# Remove fields that are specifically excluded from lookup
for field in self . fields :
    if field . no_lookup and field . name in kwargs :
        kwargs . pop ( field . name )
# If queries were provided , process them .
params = list ( kwargs . items ( ) )
for query in queries :
    params . append ( ( query [ 0 ] , query [ 1 ] ) )
# Make the request to the Ansible Tower API .
r = client . get ( url , params = params )
resp = r . json ( )
# If this was a request with a primary key included , then at the
# point that we got a good result , we know that we ' re done and can
# return the result .
if pk : # Make the results all look the same , for easier parsing
# by other methods .
# Note that the ` get ` method will effectively undo this operation ,
# but that ' s a good thing , because we might use ` get ` without a
# primary key .
    return { 'count' : 1 , 'results' : [ resp ] }
# Did we get zero results back when we shouldn ' t ?
# If so , this is an error , and we need to complain .
if fail_on_no_results and resp [ 'count' ] == 0 :
    raise exc . NotFound ( 'The requested object could not be found.' )
# Did we get more than one result back ?
# If so , this is also an error , and we need to complain .
if fail_on_multiple_results and resp [ 'count' ] >= 2 :
    raise exc . MultipleResults ( 'Expected one result, got %d. Possibly caused by not providing required ' 'fields. Please tighten your criteria.' % resp [ 'count' ] )
# Return the response .
return resp


signature:
def write ( self , pk = None , create_on_missing = False , fail_on_found = False , force_on_exists = True , ** kwargs ) :
    """= = = = = API DOCS = = = = =
 Modify the given object using the Ansible Tower API .
 : param pk : Primary key of the resource to be read . Tower CLI will only attempt to read that object
 if ` ` pk ` ` is provided ( not ` ` None ` ` ) .
 : type pk : int
 : param create _ on _ missing : Flag that if set , a new object is created if ` ` pk ` ` is not set and objects
 matching the appropriate unique criteria is not found .
 : type create _ on _ missing : bool
 : param fail _ on _ found : Flag that if set , the operation fails if an object matching the unique criteria
 already exists .
 : type fail _ on _ found : bool
 : param force _ on _ exists : Flag that if set , then if an object is modified based on matching via unique
 fields ( as opposed to the primary key ) , other fields are updated based on data
 sent ; If unset , then the non - unique values are only written in a creation case .
 : type force _ on _ exists : bool
 : param ` * * kwargs ` : Keyword arguments which , all together , will be used as POST / PATCH body to create / modify
 the resource object . if ` ` pk ` ` is not set , key - value pairs of ` ` * * kwargs ` ` which are
 also in resource ' s identity will be used to lookup existing reosource .
 : returns : A dictionary combining the JSON output of the resource , as well as two extra fields : " changed " ,
 a flag indicating if the resource is created or successfully updated ; " id " , an integer which
 is the primary key of the specified object .
 : rtype : dict
 : raises tower _ cli . exceptions . BadRequest : When required fields are missing in ` ` * * kwargs ` ` when creating
 a new resource object .
 = = = = = API DOCS = = = = ="""

implementation:
existing_data = { }
# Remove default values ( anything where the value is None ) .
self . _pop_none ( kwargs )
# Determine which record we are writing , if we weren ' t given a primary key .
if not pk :
    debug . log ( 'Checking for an existing record.' , header = 'details' )
    existing_data = self . _lookup ( fail_on_found = fail_on_found , fail_on_missing = not create_on_missing , include_debug_header = False , ** kwargs )
    if existing_data :
        pk = existing_data [ 'id' ]
else : # We already know the primary key , but get the existing data .
# This allows us to know whether the write made any changes .
    debug . log ( 'Getting existing record.' , header = 'details' )
    existing_data = self . get ( pk )
# Sanity check : Are we missing required values ?
# If we don ' t have a primary key , then all required values must be set , and if they ' re not , it ' s an error .
missing_fields = [ ]
for i in self . fields :
    if i . key not in kwargs and i . name not in kwargs and i . required :
        missing_fields . append ( i . key or i . name )
if missing_fields and not pk :
    raise exc . BadRequest ( 'Missing required fields: %s' % ', ' . join ( missing_fields ) . replace ( '_' , '-' ) )
# Sanity check : Do we need to do a write at all ?
# If ` force _ on _ exists ` is False and the record was , in fact , found , then no action is required .
if pk and not force_on_exists :
    debug . log ( 'Record already exists, and --force-on-exists is off; do nothing.' , header = 'decision' , nl = 2 )
    answer = OrderedDict ( ( ( 'changed' , False ) , ( 'id' , pk ) ) )
    answer . update ( existing_data )
    return answer
# Similarly , if all existing data matches our write parameters , there ' s no need to do anything .
if all ( [ kwargs [ k ] == existing_data . get ( k , None ) for k in kwargs . keys ( ) ] ) :
    debug . log ( 'All provided fields match existing data; do nothing.' , header = 'decision' , nl = 2 )
    answer = OrderedDict ( ( ( 'changed' , False ) , ( 'id' , pk ) ) )
    answer . update ( existing_data )
    return answer
# Reinsert None for special case of null association
for key in kwargs :
    if kwargs [ key ] == 'null' :
        kwargs [ key ] = None
# Get the URL and method to use for the write .
url = self . endpoint
method = 'POST'
if pk :
    url = self . _get_patch_url ( url , pk )
    method = 'PATCH'
# If debugging is on , print the URL and data being sent .
debug . log ( 'Writing the record.' , header = 'details' )
# Actually perform the write .
r = getattr ( client , method . lower ( ) ) ( url , data = kwargs )
# At this point , we know the write succeeded , and we know that data was changed in the process .
answer = OrderedDict ( ( ( 'changed' , True ) , ( 'id' , r . json ( ) [ 'id' ] ) ) )
answer . update ( r . json ( ) )
return answer


signature:
def delete ( self , pk = None , fail_on_missing = False , ** kwargs ) :
    """Remove the given object .
 If ` fail _ on _ missing ` is True , then the object ' s not being found is considered a failure ; otherwise ,
 a success with no change is reported .
 = = = = = API DOCS = = = = =
 Remove the given object .
 : param pk : Primary key of the resource to be deleted .
 : type pk : int
 : param fail _ on _ missing : Flag that if set , the object ' s not being found is considered a failure ; otherwise ,
 a success with no change is reported .
 : type fail _ on _ missing : bool
 : param ` * * kwargs ` : Keyword arguments used to look up resource object to delete if ` ` pk ` ` is not provided .
 : returns : dictionary of only one field " changed " , which is a flag indicating whether the specified resource
 is successfully deleted .
 : rtype : dict
 = = = = = API DOCS = = = = ="""

implementation:
# If we weren ' t given a primary key , determine which record we ' re deleting .
if not pk :
    existing_data = self . _lookup ( fail_on_missing = fail_on_missing , ** kwargs )
    if not existing_data :
        return { 'changed' : False }
    pk = existing_data [ 'id' ]
# Attempt to delete the record . If it turns out the record doesn ' t exist , handle the 404 appropriately
# ( this is an okay response if ` fail _ on _ missing ` is False ) .
url = '%s%s/' % ( self . endpoint , pk )
debug . log ( 'DELETE %s' % url , fg = 'blue' , bold = True )
try :
    client . delete ( url )
    return { 'changed' : True }
except exc . NotFound :
    if fail_on_missing :
        raise
    return { 'changed' : False }


signature:
def get ( self , pk = None , ** kwargs ) :
    """Return one and exactly one object .
 Lookups may be through a primary key , specified as a positional argument , and / or through filters specified
 through keyword arguments .
 If the number of results does not equal one , raise an exception .
 = = = = = API DOCS = = = = =
 Retrieve one and exactly one object .
 : param pk : Primary key of the resource to be read . Tower CLI will only attempt to read * that * object
 if ` ` pk ` ` is provided ( not ` ` None ` ` ) .
 : type pk : int
 : param ` * * kwargs ` : Keyword arguments used to look up resource object to retrieve if ` ` pk ` ` is not provided .
 : returns : loaded JSON of the retrieved resource object .
 : rtype : dict
 = = = = = API DOCS = = = = ="""

implementation:
if kwargs . pop ( 'include_debug_header' , True ) :
    debug . log ( 'Getting the record.' , header = 'details' )
response = self . read ( pk = pk , fail_on_no_results = True , fail_on_multiple_results = True , ** kwargs )
return response [ 'results' ] [ 0 ]