signature:
def env ( ) :
    """Verify IPMI environment"""

implementation:
ipmi = cij . env_to_dict ( PREFIX , REQUIRED )
if ipmi is None :
    ipmi [ "USER" ] = "admin"
    ipmi [ "PASS" ] = "admin"
    ipmi [ "HOST" ] = "localhost"
    ipmi [ "PORT" ] = "623"
    cij . info ( "ipmi.env: USER: %s, PASS: %s, HOST: %s, PORT: %s" % ( ipmi [ "USER" ] , ipmi [ "PASS" ] , ipmi [ "HOST" ] , ipmi [ "PORT" ] ) )
cij . env_export ( PREFIX , EXPORTED , ipmi )
return 0


signature:
def cmd ( command ) :
    """Send IPMI ' command ' via ipmitool"""

implementation:
env ( )
ipmi = cij . env_to_dict ( PREFIX , EXPORTED + REQUIRED )
command = "ipmitool -U %s -P %s -H %s -p %s %s" % ( ipmi [ "USER" ] , ipmi [ "PASS" ] , ipmi [ "HOST" ] , ipmi [ "PORT" ] , command )
cij . info ( "ipmi.command: %s" % command )
return cij . util . execute ( command , shell = True , echo = True )


signature:
def regex_find ( pattern , content ) :
    """Find the given ' pattern ' in ' content '"""

implementation:
find = re . findall ( pattern , content )
if not find :
    cij . err ( "pattern <%r> is invalid, no matches!" % pattern )
    cij . err ( "content: %r" % content )
    return ''
if len ( find ) >= 2 :
    cij . err ( "pattern <%r> is too simple, matched more than 2!" % pattern )
    cij . err ( "content: %r" % content )
    return ''
return find [ 0 ]


signature:
def execute ( cmd = None , shell = True , echo = True ) :
    """Execute the given ' cmd '
 @ returns ( rcode , stdout , stderr )"""

implementation:
if echo :
    cij . emph ( "cij.util.execute: shell: %r, cmd: %r" % ( shell , cmd ) )
rcode = 1
stdout , stderr = ( "" , "" )
if cmd :
    if shell :
        cmd = " " . join ( cmd )
    proc = Popen ( cmd , stdout = PIPE , stderr = PIPE , shell = shell , close_fds = True )
    stdout , stderr = proc . communicate ( )
    rcode = proc . returncode
if rcode and echo :
    cij . warn ( "cij.util.execute: stdout: %s" % stdout )
    cij . err ( "cij.util.execute: stderr: %s" % stderr )
    cij . err ( "cij.util.execute: rcode: %s" % rcode )
return rcode , stdout , stderr


signature:
def env ( ) :
    """Verify BOARD variables and construct exported variables"""

implementation:
if cij . ssh . env ( ) :
    cij . err ( "board.env: invalid SSH environment" )
    return 1
board = cij . env_to_dict ( PREFIX , REQUIRED )
# Verify REQUIRED variables
if board is None :
    cij . err ( "board.env: invalid BOARD environment" )
    return 1
board [ "CLASS" ] = "_" . join ( [ board [ r ] for r in REQUIRED [ : - 1 ] ] )
board [ "IDENT" ] = "-" . join ( [ board [ "CLASS" ] , board [ "ALIAS" ] ] )
cij . env_export ( PREFIX , EXPORTED , board )
# Export EXPORTED variables
return 0


signature:
def cat_file ( path ) :
    """Cat file and return content"""

implementation:
cmd = [ "cat" , path ]
status , stdout , _ = cij . ssh . command ( cmd , shell = True , echo = True )
if status :
    raise RuntimeError ( "cij.nvme.env: cat %s failed" % path )
return stdout . strip ( )


signature:
def env ( ) :
    """Verify NVME variables and construct exported variables"""

implementation:
if cij . ssh . env ( ) :
    cij . err ( "cij.nvme.env: invalid SSH environment" )
    return 1
nvme = cij . env_to_dict ( PREFIX , REQUIRED )
nvme [ "DEV_PATH" ] = os . path . join ( "/dev" , nvme [ "DEV_NAME" ] )
# get version , chunks , luns and chs
try :
    sysfs = os . path . join ( "/sys/class/block" , nvme [ "DEV_NAME" ] , "lightnvm" )
    nvme [ "LNVM_VERSION" ] = cat_file ( os . path . join ( sysfs , "version" ) )
    if nvme [ "LNVM_VERSION" ] == "2.0" :
        luns = "punits"
        chs = "groups"
    elif nvme [ "LNVM_VERSION" ] == "1.2" :
        luns = "num_luns"
        chs = "num_channels"
    else :
        raise RuntimeError ( "cij.nvme.env: invalid lnvm version: %s" % nvme [ "LNVM_VERSION" ] )
    nvme [ "LNVM_NUM_CHUNKS" ] = cat_file ( os . path . join ( sysfs , "chunks" ) )
    nvme [ "LNVM_NUM_LUNS" ] = cat_file ( os . path . join ( sysfs , luns ) )
    nvme [ "LNVM_NUM_CHS" ] = cat_file ( os . path . join ( sysfs , chs ) )
    nvme [ "LNVM_TOTAL_LUNS" ] = str ( int ( nvme [ "LNVM_NUM_LUNS" ] ) * int ( nvme [ "LNVM_NUM_CHS" ] ) )
    nvme [ "LNVM_TOTAL_CHUNKS" ] = str ( int ( nvme [ "LNVM_TOTAL_LUNS" ] ) * int ( nvme [ "LNVM_NUM_CHUNKS" ] ) )
    # get spec version by identify namespace data struct
    if nvme [ "LNVM_VERSION" ] == "2.0" :
        cmd = [ "nvme" , "id-ctrl" , nvme [ "DEV_PATH" ] , "--raw-binary" ]
        status , stdout , _ = cij . ssh . command ( cmd , shell = True )
        if status :
            raise RuntimeError ( "cij.nvme.env: nvme id-ctrl fail" )
        buff = cij . bin . Buffer ( types = IdentifyCDS , length = 1 )
        buff . memcopy ( stdout )
        if buff [ 0 ] . VS [ 1023 ] == 0x5a :
            nvme [ "SPEC_VERSION" ] = "Denali"
        else :
            nvme [ "SPEC_VERSION" ] = "Spec20"
    else :
        nvme [ "SPEC_VERSION" ] = "Spec12"
    # get chunk meta information
    nvme [ "LNVM_CHUNK_META_LENGTH" ] = str ( get_sizeof_descriptor_table ( nvme [ "SPEC_VERSION" ] ) )
    nvme [ "LNVM_CHUNK_META_SIZE" ] = str ( int ( nvme [ "LNVM_CHUNK_META_LENGTH" ] ) * int ( nvme [ "LNVM_TOTAL_CHUNKS" ] ) )
except StandardError :
    traceback . print_exc ( )
    return 1
cij . env_export ( PREFIX , EXPORTED , nvme )
return 0


signature:
def fmt ( lbaf = 3 ) :
    """Do format for NVMe device"""

implementation:
if env ( ) :
    cij . err ( "cij.nvme.exists: Invalid NVMe ENV." )
    return 1
nvme = cij . env_to_dict ( PREFIX , EXPORTED + REQUIRED )
cmd = [ "nvme" , "format" , nvme [ "DEV_PATH" ] , "-l" , str ( lbaf ) ]
rcode , _ , _ = cij . ssh . command ( cmd , shell = True )
return rcode


signature:
def get_meta ( offset , length , output ) :
    """Get chunk meta of NVMe device"""

implementation:
if env ( ) :
    cij . err ( "cij.nvme.meta: Invalid NVMe ENV." )
    return 1
nvme = cij . env_to_dict ( PREFIX , EXPORTED + REQUIRED )
max_size = 0x40000
with open ( output , "wb" ) as fout :
    for off in range ( offset , length , max_size ) :
        size = min ( length - off , max_size )
        cmd = [ "nvme get-log" , nvme [ "DEV_PATH" ] , "-i 0xca" , "-o 0x%x" % off , "-l 0x%x" % size , "-b" ]
        status , stdout , _ = cij . ssh . command ( cmd , shell = True )
        if status :
            cij . err ( "cij.nvme.meta: Error get chunk meta" )
            return 1
        fout . write ( stdout )
return 0


signature:
def comp_meta ( file_bef , file_aft , mode = "pfail" ) :
    """Compare chunk meta , mode = [ pfail , power , reboot ]"""

implementation:
if env ( ) :
    cij . err ( "cij.nvme.comp_meta: Invalid NVMe ENV." )
    return 1
nvme = cij . env_to_dict ( PREFIX , EXPORTED + REQUIRED )
num_chk = int ( nvme [ "LNVM_TOTAL_CHUNKS" ] )
meta_bef = cij . bin . Buffer ( types = get_descriptor_table ( nvme [ 'SPEC_VERSION' ] ) , length = num_chk )
meta_aft = cij . bin . Buffer ( types = get_descriptor_table ( nvme [ 'SPEC_VERSION' ] ) , length = num_chk )
meta_bef . read ( file_bef )
meta_aft . read ( file_aft )
for chk in range ( num_chk ) :
    ignore = [ "WL" , "RSV0" ]
    # PFAIL : BEFORE IS OPEN CHUNK , WRITE POINTER IS NOT SURE , IGNORE
    if mode == "pfail" and meta_bef [ chk ] . CS == 4 :
        ignore . append ( "WP" )
    # COMPARE CHUNK META
    if meta_bef . compare ( meta_aft , chk , ignore = ignore ) :
        cij . warn ( "META_BUFF_BEF[%s]:" % chk )
        meta_bef . dump ( chk )
        cij . warn ( "META_BUFF_AFT[%s]:" % chk )
        meta_aft . dump ( chk )
        cij . err ( "Error compare, CHUNK: %s" % chk )
        return 1
return 0


signature:
def get_sizeof_descriptor_table ( version = "Denali" ) :
    """Get sizeof DescriptorTable"""

implementation:
if version == "Denali" :
    return sizeof ( DescriptorTableDenali )
elif version == "Spec20" :
    return sizeof ( DescriptorTableSpec20 )
elif version == "Spec12" :
    return 0
else :
    raise RuntimeError ( "Error version!" )


signature:
def env ( ) :
    """Verify LNVM variables and construct exported variables"""

implementation:
if cij . ssh . env ( ) :
    cij . err ( "cij.lnvm.env: invalid SSH environment" )
    return 1
lnvm = cij . env_to_dict ( PREFIX , REQUIRED )
nvme = cij . env_to_dict ( "NVME" , [ "DEV_NAME" ] )
if "BGN" not in lnvm . keys ( ) :
    cij . err ( "cij.lnvm.env: invalid LNVM_BGN" )
    return 1
if "END" not in lnvm . keys ( ) :
    cij . err ( "cij.lnvm.env: invalid LNVM_END" )
    return 1
if "DEV_TYPE" not in lnvm . keys ( ) :
    cij . err ( "cij.lnvm.env: invalid LNVM_DEV_TYPE" )
    return 1
lnvm [ "DEV_NAME" ] = "%sb%03de%03d" % ( nvme [ "DEV_NAME" ] , int ( lnvm [ "BGN" ] ) , int ( lnvm [ "END" ] ) )
lnvm [ "DEV_PATH" ] = "/dev/%s" % lnvm [ "DEV_NAME" ]
cij . env_export ( PREFIX , EXPORTED , lnvm )
return 0


signature:
def create ( ) :
    """Create LNVM device"""

implementation:
if env ( ) :
    cij . err ( "cij.lnvm.create: Invalid LNVM ENV" )
    return 1
nvme = cij . env_to_dict ( "NVME" , [ "DEV_NAME" ] )
lnvm = cij . env_to_dict ( PREFIX , EXPORTED + REQUIRED )
cij . emph ( "lnvm.create: LNVM_DEV_NAME: %s" % lnvm [ "DEV_NAME" ] )
cmd = [ "nvme lnvm create -d %s -n %s -t %s -b %s -e %s -f" % ( nvme [ "DEV_NAME" ] , lnvm [ "DEV_NAME" ] , lnvm [ "DEV_TYPE" ] , lnvm [ "BGN" ] , lnvm [ "END" ] ) ]
rcode , _ , _ = cij . ssh . command ( cmd , shell = True )
if rcode :
    cij . err ( "cij.lnvm.create: FAILED" )
    return 1
return 0


signature:
def dump ( buf , indent = 0 , skip = "" ) :
    """Dump UnionType / StructType to STDOUT"""

implementation:
if not isinstance ( type ( buf ) , ( type ( Union ) , type ( Structure ) ) ) :
    raise RuntimeError ( "Error type(%s)" % type ( buf ) )
for field in getattr ( buf , '_fields_' ) :
    name , types = field [ 0 ] , field [ 1 ]
    if name in skip :
        return
    value = getattr ( buf , name )
    if isinstance ( types , ( type ( Union ) , type ( Structure ) ) ) :
        cij . info ( "%s%s:" % ( " " * indent , name ) )
        dump ( value , indent + 2 , skip )
    elif isinstance ( types , type ( Array ) ) :
        for i , item in enumerate ( value ) :
            name_index = "%s[%s]" % ( name , i )
            if isinstance ( types , ( type ( Union ) , type ( Structure ) ) ) :
                cij . info ( "%s%s:" % ( " " * indent , name_index ) )
                dump ( item , indent + 2 , skip )
            else :
                cij . info ( "%s%-12s: 0x%x" % ( " " * indent , name_index , item ) )
    else :
        cij . info ( "%s%-12s: 0x%x" % ( " " * indent , name , value ) )


signature:
def compare ( buf_a , buf_b , ignore ) :
    """Compare of two Buffer item"""

implementation:
for field in getattr ( buf_a , '_fields_' ) :
    name , types = field [ 0 ] , field [ 1 ]
    if name in ignore :
        continue
    val_a = getattr ( buf_a , name )
    val_b = getattr ( buf_b , name )
    if isinstance ( types , ( type ( Union ) , type ( Structure ) ) ) :
        if compare ( val_a , val_b , ignore ) :
            return 1
    elif isinstance ( types , type ( Array ) ) :
        for i , _ in enumerate ( val_a ) :
            if isinstance ( types , ( type ( Union ) , type ( Structure ) ) ) :
                if compare ( val_a [ i ] , val_b [ i ] , ignore ) :
                    return 1
            else :
                if val_a [ i ] != val_b [ i ] :
                    return 1
    else :
        if val_a != val_b :
            return 1
return 0


signature:
def memcopy ( self , stream , offset = 0 , length = float ( "inf" ) ) :
    """Copy stream to buffer"""

implementation:
data = [ ord ( i ) for i in list ( stream ) ]
size = min ( length , len ( data ) , self . m_size )
buff = cast ( self . m_buf , POINTER ( c_uint8 ) )
for i in range ( size ) :
    buff [ offset + i ] = data [ i ]


signature:
def write ( self , path ) :
    """Write buffer to file"""

implementation:
with open ( path , "wb" ) as fout :
    fout . write ( self . m_buf )


signature:
def read ( self , path ) :
    """Read file to buffer"""

implementation:
with open ( path , "rb" ) as fout :
    memmove ( self . m_buf , fout . read ( self . m_size ) , self . m_size )


signature:
def dump ( self , offset = 0 , length = 1 ) :
    """Dump item"""

implementation:
for i in range ( offset , offset + length ) :
    if "ctypes" in str ( self . m_types ) :
        cij . info ( "Buff[%s]: %s" % ( i , self . m_buf [ i ] ) )
    else :
        cij . info ( "Buff[%s]:" % i )
        dump ( self . m_buf [ i ] , 2 )


signature:
def compare ( self , buf , offset = 0 , length = 1 , ignore = "" ) :
    """Compare buffer"""

implementation:
for i in range ( offset , offset + length ) :
    if isinstance ( self . m_types , ( type ( Union ) , type ( Structure ) ) ) :
        if compare ( self . m_buf [ i ] , buf [ i ] , ignore = ignore ) :
            return 1
    elif self . m_buf [ i ] != buf [ i ] :
        return 1
return 0


signature:
def power_on ( self , interval = 200 ) :
    """230v power on"""

implementation:
if self . __power_on_port is None :
    cij . err ( "cij.usb.relay: Invalid USB_RELAY_POWER_ON" )
    return 1
return self . __press ( self . __power_on_port , interval = interval )


signature:
def power_off ( self , interval = 200 ) :
    """230v power off"""

implementation:
if self . __power_off_port is None :
    cij . err ( "cij.usb.relay: Invalid USB_RELAY_POWER_OFF" )
    return 1
return self . __press ( self . __power_off_port , interval = interval )


signature:
def power_btn ( self , interval = 200 ) :
    """TARGET power button"""

implementation:
if self . __power_btn_port is None :
    cij . err ( "cij.usb.relay: Invalid USB_RELAY_POWER_BTN" )
    return 1
return self . __press ( self . __power_btn_port , interval = interval )


signature:
def get_chunk_information ( self , chk , lun , chunk_name ) :
    """Get chunk information"""

implementation:
cmd = [ "nvm_cmd rprt_lun" , self . envs , "%d %d > %s" % ( chk , lun , chunk_name ) ]
status , _ , _ = cij . ssh . command ( cmd , shell = True )
return status


signature:
def is_bad_chunk ( self , chk , yml ) :
    """Check the chunk is offline or not"""

implementation:
cs = self . get_chunk_status ( chk , yml )
if cs >= 8 :
    return True
return False


signature:
def is_free_chunk ( self , chk ) :
    """Check the chunk is free or not"""

implementation:
cs = self . get_chunk_status ( chk )
if cs & 0x1 != 0 :
    return True
return False


signature:
def is_closed_chunk ( self , chk ) :
    """Check the chunk is free or not"""

implementation:
cs = self . get_chunk_status ( chk )
if cs & 0x2 != 0 :
    return True
return False


signature:
def is_open_chunk ( self , chk ) :
    """Check the chunk is free or not"""

implementation:
cs = self . get_chunk_status ( chk )
if cs & 0x4 != 0 :
    return True
return False


signature:
def vblk_erase ( self , address ) :
    """nvm _ vblk erase"""

implementation:
cmd = [ "nvm_vblk erase" , self . envs , "0x%x" % address ]
status , _ , _ = cij . ssh . command ( cmd , shell = True )
return status


signature:
def slc_erase ( self , address , BE_ID = 0x1 , PMODE = 0x0100 ) :
    """slc erase"""

implementation:
cmd = [ "NVM_CLI_BE_ID=0x%x" % BE_ID , "NVM_CLI_PMODE=0x%x" % PMODE , "nvm_cmd erase" , self . envs , "0x%x" % address ]
status , _ , _ = cij . ssh . command ( cmd , shell = True )
return status


signature:
def env ( ) :
    """Verify BLOCK variables and construct exported variables"""

implementation:
if cij . ssh . env ( ) :
    cij . err ( "cij.block.env: invalid SSH environment" )
    return 1
block = cij . env_to_dict ( PREFIX , REQUIRED )
block [ "DEV_PATH" ] = "/dev/%s" % block [ "DEV_NAME" ]
cij . env_export ( PREFIX , EXPORTED , block )
return 0


signature:
def script_run ( trun , script ) :
    """Execute a script or testcase"""

implementation:
if trun [ "conf" ] [ "VERBOSE" ] :
    cij . emph ( "rnr:script:run { script: %s }" % script )
    cij . emph ( "rnr:script:run:evars: %s" % script [ "evars" ] )
launchers = { ".py" : "python" , ".sh" : "source" }
ext = os . path . splitext ( script [ "fpath" ] ) [ - 1 ]
if not ext in launchers . keys ( ) :
    cij . err ( "rnr:script:run { invalid script[\"fpath\"]: %r }" % script [ "fpath" ] )
    return 1
launch = launchers [ ext ]
with open ( script [ "log_fpath" ] , "a" ) as log_fd :
    log_fd . write ( "# script_fpath: %r\n" % script [ "fpath" ] )
    log_fd . flush ( )
    bgn = time . time ( )
    cmd = [ 'bash' , '-c' , 'CIJ_ROOT=$(cij_root) && ' 'source $CIJ_ROOT/modules/cijoe.sh && ' 'source %s && ' 'CIJ_TEST_RES_ROOT="%s" %s %s ' % ( trun [ "conf" ] [ "ENV_FPATH" ] , script [ "res_root" ] , launch , script [ "fpath" ] ) ]
    if trun [ "conf" ] [ "VERBOSE" ] > 1 :
        cij . emph ( "rnr:script:run { cmd: %r }" % " " . join ( cmd ) )
    evars = os . environ . copy ( )
    evars . update ( { k : str ( script [ "evars" ] [ k ] ) for k in script [ "evars" ] } )
    process = Popen ( cmd , stdout = log_fd , stderr = STDOUT , cwd = script [ "res_root" ] , env = evars )
    process . wait ( )
    script [ "rcode" ] = process . returncode
    script [ "wallc" ] = time . time ( ) - bgn
if trun [ "conf" ] [ "VERBOSE" ] :
    cij . emph ( "rnr:script:run { wallc: %02f }" % script [ "wallc" ] )
    cij . emph ( "rnr:script:run { rcode: %r } " % script [ "rcode" ] , script [ "rcode" ] )
return script [ "rcode" ]


signature:
def hook_setup ( parent , hook_fpath ) :
    """Setup hook"""

implementation:
hook = copy . deepcopy ( HOOK )
hook [ "name" ] = os . path . splitext ( os . path . basename ( hook_fpath ) ) [ 0 ]
hook [ "name" ] = hook [ "name" ] . replace ( "_enter" , "" ) . replace ( "_exit" , "" )
hook [ "res_root" ] = parent [ "res_root" ]
hook [ "fpath_orig" ] = hook_fpath
hook [ "fname" ] = "hook_%s" % os . path . basename ( hook [ "fpath_orig" ] )
hook [ "fpath" ] = os . sep . join ( [ hook [ "res_root" ] , hook [ "fname" ] ] )
hook [ "log_fpath" ] = os . sep . join ( [ hook [ "res_root" ] , "%s.log" % hook [ "fname" ] ] )
hook [ "evars" ] . update ( copy . deepcopy ( parent [ "evars" ] ) )
shutil . copyfile ( hook [ "fpath_orig" ] , hook [ "fpath" ] )
return hook


signature:
def hooks_setup ( trun , parent , hnames = None ) :
    """Setup test - hooks
 @ returns dict of hook filepaths { " enter " : [ ] , " exit " : [ ] }"""

implementation:
hooks = { "enter" : [ ] , "exit" : [ ] }
if hnames is None : # Nothing to do , just return the struct
    return hooks
for hname in hnames : # Fill out paths
    for med in HOOK_PATTERNS :
        for ptn in HOOK_PATTERNS [ med ] :
            fpath = os . sep . join ( [ trun [ "conf" ] [ "HOOKS" ] , ptn % hname ] )
            if not os . path . exists ( fpath ) :
                continue
            hook = hook_setup ( parent , fpath )
            if not hook :
                continue
            hooks [ med ] . append ( hook )
    if not hooks [ "enter" ] + hooks [ "exit" ] :
        cij . err ( "rnr:hooks_setup:FAIL { hname: %r has no files }" % hname )
        return None
return hooks


signature:
def trun_to_file ( trun , fpath = None ) :
    """Dump the given trun to file"""

implementation:
if fpath is None :
    fpath = yml_fpath ( trun [ "conf" ] [ "OUTPUT" ] )
with open ( fpath , 'w' ) as yml_file :
    data = yaml . dump ( trun , explicit_start = True , default_flow_style = False )
    yml_file . write ( data )


signature:
def trun_emph ( trun ) :
    """Print essential info on"""

implementation:
if trun [ "conf" ] [ "VERBOSE" ] > 1 : # Print environment variables
    cij . emph ( "rnr:CONF {" )
    for cvar in sorted ( trun [ "conf" ] . keys ( ) ) :
        cij . emph ( "  % 16s: %r" % ( cvar , trun [ "conf" ] [ cvar ] ) )
    cij . emph ( "}" )
if trun [ "conf" ] [ "VERBOSE" ] :
    cij . emph ( "rnr:INFO {" )
    cij . emph ( "  OUTPUT: %r" % trun [ "conf" ] [ "OUTPUT" ] )
    cij . emph ( "  yml_fpath: %r" % yml_fpath ( trun [ "conf" ] [ "OUTPUT" ] ) )
    cij . emph ( "}" )


signature:
def tcase_setup ( trun , parent , tcase_fname ) :
    """Create and initialize a testcase"""

implementation:
# pylint : disable = locally - disabled , unused - argument
case = copy . deepcopy ( TESTCASE )
case [ "fname" ] = tcase_fname
case [ "fpath_orig" ] = os . sep . join ( [ trun [ "conf" ] [ "TESTCASES" ] , case [ "fname" ] ] )
if not os . path . exists ( case [ "fpath_orig" ] ) :
    cij . err ( 'rnr:tcase_setup: !case["fpath_orig"]: %r' % case [ "fpath_orig" ] )
    return None
case [ "name" ] = os . path . splitext ( case [ "fname" ] ) [ 0 ]
case [ "ident" ] = "/" . join ( [ parent [ "ident" ] , case [ "fname" ] ] )
case [ "res_root" ] = os . sep . join ( [ parent [ "res_root" ] , case [ "fname" ] ] )
case [ "aux_root" ] = os . sep . join ( [ case [ "res_root" ] , "_aux" ] )
case [ "log_fpath" ] = os . sep . join ( [ case [ "res_root" ] , "run.log" ] )
case [ "fpath" ] = os . sep . join ( [ case [ "res_root" ] , case [ "fname" ] ] )
case [ "evars" ] . update ( copy . deepcopy ( parent [ "evars" ] ) )
# Initalize
os . makedirs ( case [ "res_root" ] )
# Create DIRS
os . makedirs ( case [ "aux_root" ] )
shutil . copyfile ( case [ "fpath_orig" ] , case [ "fpath" ] )
# Copy testcase
# Initialize hooks
case [ "hooks" ] = hooks_setup ( trun , case , parent . get ( "hooks_pr_tcase" ) )
return case


signature:
def tsuite_exit ( trun , tsuite ) :
    """Triggers when exiting the given testsuite"""

implementation:
if trun [ "conf" ] [ "VERBOSE" ] :
    cij . emph ( "rnr:tsuite:exit" )
rcode = 0
for hook in reversed ( tsuite [ "hooks" ] [ "exit" ] ) : # EXIT - hooks
    rcode = script_run ( trun , hook )
    if rcode :
        break
if trun [ "conf" ] [ "VERBOSE" ] :
    cij . emph ( "rnr:tsuite:exit { rcode: %r } " % rcode , rcode )
return rcode


signature:
def tsuite_enter ( trun , tsuite ) :
    """Triggers when entering the given testsuite"""

implementation:
if trun [ "conf" ] [ "VERBOSE" ] :
    cij . emph ( "rnr:tsuite:enter { name: %r }" % tsuite [ "name" ] )
rcode = 0
for hook in tsuite [ "hooks" ] [ "enter" ] : # ENTER - hooks
    rcode = script_run ( trun , hook )
    if rcode :
        break
if trun [ "conf" ] [ "VERBOSE" ] :
    cij . emph ( "rnr:tsuite:enter { rcode: %r } " % rcode , rcode )
return rcode


signature:
def tsuite_setup ( trun , declr , enum ) :
    """Creates and initialized a TESTSUITE struct and site - effects such as creating
 output directories and forwarding initialization of testcases"""

implementation:
suite = copy . deepcopy ( TESTSUITE )
# Setup the test - suite
suite [ "name" ] = declr . get ( "name" )
if suite [ "name" ] is None :
    cij . err ( "rnr:tsuite_setup: no testsuite is given" )
    return None
suite [ "alias" ] = declr . get ( "alias" )
suite [ "ident" ] = "%s_%d" % ( suite [ "name" ] , enum )
suite [ "res_root" ] = os . sep . join ( [ trun [ "conf" ] [ "OUTPUT" ] , suite [ "ident" ] ] )
suite [ "aux_root" ] = os . sep . join ( [ suite [ "res_root" ] , "_aux" ] )
suite [ "evars" ] . update ( copy . deepcopy ( trun [ "evars" ] ) )
suite [ "evars" ] . update ( copy . deepcopy ( declr . get ( "evars" , { } ) ) )
# Initialize
os . makedirs ( suite [ "res_root" ] )
os . makedirs ( suite [ "aux_root" ] )
# Setup testsuite - hooks
suite [ "hooks" ] = hooks_setup ( trun , suite , declr . get ( "hooks" ) )
# Forward from declaration
suite [ "hooks_pr_tcase" ] = declr . get ( "hooks_pr_tcase" , [ ] )
suite [ "fname" ] = "%s.suite" % suite [ "name" ]
suite [ "fpath" ] = os . sep . join ( [ trun [ "conf" ] [ "TESTSUITES" ] , suite [ "fname" ] ] )
# Load testcases from . suite file OR from declaration
tcase_fpaths = [ ]
# Load testcase fpaths
if os . path . exists ( suite [ "fpath" ] ) : # From suite - file
    suite_lines = ( l . strip ( ) for l in open ( suite [ "fpath" ] ) . read ( ) . splitlines ( ) )
    tcase_fpaths . extend ( ( l for l in suite_lines if len ( l ) > 1 and l [ 0 ] != "#" ) )
else : # From declaration
    tcase_fpaths . extend ( declr . get ( "testcases" , [ ] ) )
# NOTE : fix duplicates ; allow them
# NOTE : Currently hot - fixed here
if len ( set ( tcase_fpaths ) ) != len ( tcase_fpaths ) :
    cij . err ( "rnr:suite: failed: duplicate tcase in suite not supported" )
    return None
for tcase_fname in tcase_fpaths : # Setup testcases
    tcase = tcase_setup ( trun , suite , tcase_fname )
    if not tcase :
        cij . err ( "rnr:suite: failed: tcase_setup" )
        return None
    suite [ "testcases" ] . append ( tcase )
return suite


signature:
def tcase_enter ( trun , tsuite , tcase ) :
    """setup res _ root and aux _ root , log info and run tcase - enter - hooks
 @ returns 0 when all hooks succeed , some value othervise"""

implementation:
# pylint : disable = locally - disabled , unused - argument
if trun [ "conf" ] [ "VERBOSE" ] :
    cij . emph ( "rnr:tcase:enter" )
    cij . emph ( "rnr:tcase:enter { fname: %r }" % tcase [ "fname" ] )
    cij . emph ( "rnr:tcase:enter { log_fpath: %r }" % tcase [ "log_fpath" ] )
rcode = 0
for hook in tcase [ "hooks" ] [ "enter" ] : # tcase ENTER - hooks
    rcode = script_run ( trun , hook )
    if rcode :
        break
if trun [ "conf" ] [ "VERBOSE" ] :
    cij . emph ( "rnr:tcase:exit: { rcode: %r }" % rcode , rcode )
return rcode


signature:
def trun_exit ( trun ) :
    """Triggers when exiting the given testrun"""

implementation:
if trun [ "conf" ] [ "VERBOSE" ] :
    cij . emph ( "rnr:trun:exit" )
rcode = 0
for hook in reversed ( trun [ "hooks" ] [ "exit" ] ) : # EXIT - hooks
    rcode = script_run ( trun , hook )
    if rcode :
        break
if trun [ "conf" ] [ "VERBOSE" ] :
    cij . emph ( "rnr:trun::exit { rcode: %r }" % rcode , rcode )
return rcode


signature:
def trun_enter ( trun ) :
    """Triggers when entering the given testrun"""

implementation:
if trun [ "conf" ] [ "VERBOSE" ] :
    cij . emph ( "rnr:trun::enter" )
trun [ "stamp" ] [ "begin" ] = int ( time . time ( ) )
# Record start timestamp
rcode = 0
for hook in trun [ "hooks" ] [ "enter" ] : # ENTER - hooks
    rcode = script_run ( trun , hook )
    if rcode :
        break
if trun [ "conf" ] [ "VERBOSE" ] :
    cij . emph ( "rnr:trun::enter { rcode: %r }" % rcode , rcode )
return rcode


signature:
def trun_setup ( conf ) :
    """Setup the testrunner data - structure , embedding the parsed environment
 variables and command - line arguments and continues with setup for testplans ,
 testsuites , and testcases"""

implementation:
declr = None
try :
    with open ( conf [ "TESTPLAN_FPATH" ] ) as declr_fd :
        declr = yaml . safe_load ( declr_fd )
except AttributeError as exc :
    cij . err ( "rnr: %r" % exc )
if not declr :
    return None
trun = copy . deepcopy ( TRUN )
trun [ "ver" ] = cij . VERSION
trun [ "conf" ] = copy . deepcopy ( conf )
trun [ "res_root" ] = conf [ "OUTPUT" ]
trun [ "aux_root" ] = os . sep . join ( [ trun [ "res_root" ] , "_aux" ] )
trun [ "evars" ] . update ( copy . deepcopy ( declr . get ( "evars" , { } ) ) )
os . makedirs ( trun [ "aux_root" ] )
hook_names = declr . get ( "hooks" , [ ] )
if "lock" not in hook_names :
    hook_names = [ "lock" ] + hook_names
if hook_names [ 0 ] != "lock" :
    return None
# Setup top - level hooks
trun [ "hooks" ] = hooks_setup ( trun , trun , hook_names )
for enum , declr in enumerate ( declr [ "testsuites" ] ) : # Setup testsuites
    tsuite = tsuite_setup ( trun , declr , enum )
    if tsuite is None :
        cij . err ( "main::FAILED: setting up tsuite: %r" % tsuite )
        return 1
    trun [ "testsuites" ] . append ( tsuite )
    trun [ "progress" ] [ "UNKN" ] += len ( tsuite [ "testcases" ] )
return trun


signature:
def main ( conf ) :
    """CIJ Test Runner main entry point"""

implementation:
fpath = yml_fpath ( conf [ "OUTPUT" ] )
if os . path . exists ( fpath ) : # YAML exists , we exit , it might be RUNNING !
    cij . err ( "main:FAILED { fpath: %r }, exists" % fpath )
    return 1
trun = trun_setup ( conf )
# Construct ' trun ' from ' conf '
if not trun :
    return 1
trun_to_file ( trun )
# Persist trun
trun_emph ( trun )
# Print trun before run
tr_err = 0
tr_ent_err = trun_enter ( trun )
for tsuite in ( ts for ts in trun [ "testsuites" ] if not tr_ent_err ) :
    ts_err = 0
    ts_ent_err = tsuite_enter ( trun , tsuite )
    for tcase in ( tc for tc in tsuite [ "testcases" ] if not ts_ent_err ) :
        tc_err = tcase_enter ( trun , tsuite , tcase )
        if not tc_err :
            tc_err += script_run ( trun , tcase )
            tc_err += tcase_exit ( trun , tsuite , tcase )
        tcase [ "status" ] = "FAIL" if tc_err else "PASS"
        trun [ "progress" ] [ tcase [ "status" ] ] += 1
        # Update progress
        trun [ "progress" ] [ "UNKN" ] -= 1
        ts_err += tc_err
        # Accumulate errors
        trun_to_file ( trun )
        # Persist trun
    if not ts_ent_err :
        ts_err += tsuite_exit ( trun , tsuite )
    ts_err += ts_ent_err
    # Accumulate errors
    tr_err += ts_err
    tsuite [ "status" ] = "FAIL" if ts_err else "PASS"
    cij . emph ( "rnr:tsuite %r" % tsuite [ "status" ] , tsuite [ "status" ] != "PASS" )
if not tr_ent_err :
    trun_exit ( trun )
tr_err += tr_ent_err
trun [ "status" ] = "FAIL" if tr_err else "PASS"
trun [ "stamp" ] [ "end" ] = int ( time . time ( ) ) + 1
# END STAMP
trun_to_file ( trun )
# PERSIST
cij . emph ( "rnr:main:progress %r" % trun [ "progress" ] )
cij . emph ( "rnr:main:trun %r" % trun [ "status" ] , trun [ "status" ] != "PASS" )
return trun [ "progress" ] [ "UNKN" ] + trun [ "progress" ] [ "FAIL" ]


signature:
def get_chunk_meta ( self , meta_file ) :
    """Get chunk meta table"""

implementation:
chunks = self . envs [ "CHUNKS" ]
if cij . nvme . get_meta ( 0 , chunks * self . envs [ "CHUNK_META_SIZEOF" ] , meta_file ) :
    raise RuntimeError ( "cij.liblight.get_chunk_meta: fail" )
chunk_meta = cij . bin . Buffer ( types = self . envs [ "CHUNK_META_STRUCT" ] , length = chunks )
chunk_meta . read ( meta_file )
return chunk_meta


signature:
def get_chunk_meta_item ( self , chunk_meta , grp , pug , chk ) :
    """Get item of chunk meta table"""

implementation:
num_chk = self . envs [ "NUM_CHK" ]
num_pu = self . envs [ "NUM_PU" ]
index = grp * num_pu * num_chk + pug * num_chk + chk
return chunk_meta [ index ]


signature:
def is_bad_chunk ( self , chunk_meta , grp , pug , chk ) :
    """Check the chunk is offline or not"""

implementation:
meta = self . get_chunk_meta_item ( chunk_meta , grp , pug , chk )
if meta . CS & 0x8 != 0 :
    return True
return False


signature:
def s20_to_gen ( self , pugrp , punit , chunk , sectr ) :
    """S20 unit to generic address"""

implementation:
cmd = [ "nvm_addr s20_to_gen" , self . envs [ "DEV_PATH" ] , "%d %d %d %d" % ( pugrp , punit , chunk , sectr ) ]
status , stdout , _ = cij . ssh . command ( cmd , shell = True )
if status :
    raise RuntimeError ( "cij.liblight.s20_to_gen: cmd fail" )
return int ( re . findall ( r"val: ([0-9a-fx]+)" , stdout ) [ 0 ] , 16 )


signature:
def gen_to_dev ( self , address ) :
    """Generic address to device address"""

implementation:
cmd = [ "nvm_addr gen2dev" , self . envs [ "DEV_PATH" ] , "0x{:x}" . format ( address ) ]
status , stdout , _ = cij . ssh . command ( cmd , shell = True )
if status :
    raise RuntimeError ( "cij.liblight.gen_to_dev: cmd fail" )
return int ( re . findall ( r"dev: ([0-9a-fx]+)" , stdout ) [ 0 ] , 16 )


signature:
def vblk_write ( self , address , meta = False ) :
    """nvm _ vblk write"""

implementation:
cmd = list ( )
if meta :
    cmd . append ( "NVM_CLI_META_MODE=1" )
cmd += [ "nvm_vblk write" , self . envs [ "DEV_PATH" ] , "0x%x" % address ]
status , _ , _ = cij . ssh . command ( cmd , shell = True )
return status


signature:
def vector_read ( self , address_list , file_name = None ) :
    """nvm _ cmd read"""

implementation:
address = [ "0x{:x}" . format ( i ) for i in address_list ]
cmd = [ "nvm_cmd read" , self . envs [ "DEV_PATH" ] , " " . join ( address ) ]
if file_name :
    cmd += [ "-o {}" . format ( file_name ) ]
status , _ , _ = cij . ssh . command ( cmd , shell = True )
return status


signature:
def scalar_write ( self , address , block_count , data_file , meta_file ) :
    """nvme write"""

implementation:
cmd = [ "nvme" , "write" , self . envs [ "DEV_PATH" ] , "-s 0x{:x}" . format ( address ) , "-c {}" . format ( block_count - 1 ) , "-d {}" . format ( data_file ) , "-M {}" . format ( meta_file ) , "-z 0x{:x}" . format ( block_count * self . envs [ "NBYTES" ] ) , "-y 0x{:x}" . format ( block_count * self . envs [ "NBYTES_OOB" ] ) ]
status , _ , _ = cij . ssh . command ( cmd , shell = True )
return status


signature:
def __run ( self , shell = True , echo = True ) :
    """Run DMESG job"""

implementation:
if env ( ) :
    return 1
cij . emph ( "cij.dmesg.start: shell: %r, cmd: %r" % ( shell , self . __prefix + self . __suffix ) )
return cij . ssh . command ( self . __prefix , shell , echo , self . __suffix )


signature:
def start ( self ) :
    """Start DMESG job in thread"""

implementation:
self . __thread = Thread ( target = self . __run , args = ( True , False ) )
self . __thread . setDaemon ( True )
self . __thread . start ( )


signature:
def terminate ( self ) :
    """Terminate DMESG job"""

implementation:
if self . __thread :
    cmd = [ "who am i" ]
    status , output , _ = cij . util . execute ( cmd , shell = True , echo = True )
    if status :
        cij . warn ( "cij.dmesg.terminate: who am i failed" )
        return 1
    tty = output . split ( ) [ 1 ]
    cmd = [ "pkill -f '{}' -t '{}'" . format ( " " . join ( self . __prefix ) , tty ) ]
    status , _ , _ = cij . util . execute ( cmd , shell = True , echo = True )
    if status :
        cij . warn ( "cij.dmesg.terminate: pkill failed" )
        return 1
    self . __thread . join ( )
    self . __thread = None
return 0


signature:
def generate_rt_pic ( process_data , para_meter , scale ) :
    """generate rater pic"""

implementation:
pic_path = para_meter [ 'filename' ] + '.png'
plt . figure ( figsize = ( 5.6 * scale , 3.2 * scale ) )
for key in process_data . keys ( ) :
    plt . plot ( process_data [ key ] [ : , 0 ] , process_data [ key ] [ : , 1 ] , label = str ( key ) )
plt . title ( para_meter [ 'title' ] )
plt . xlabel ( para_meter [ 'x_axis_name' ] )
plt . ylabel ( para_meter [ 'y_axis_name' ] )
plt . legend ( loc = 'upper left' )
plt . savefig ( pic_path )
return pic_path


signature:
def generate_steady_rt_pic ( process_data , para_meter , scale , steady_time ) :
    """generate rate steady"""

implementation:
pic_path_steady = para_meter [ 'filename' ] + '_steady.png'
plt . figure ( figsize = ( 4 * scale , 2.5 * scale ) )
for key in process_data . keys ( ) :
    if len ( process_data [ key ] ) < steady_time :
        steady_time = len ( process_data [ key ] )
    plt . scatter ( process_data [ key ] [ - 1 * steady_time : , 0 ] , process_data [ key ] [ - 1 * steady_time : , 1 ] , label = str ( key ) , s = 10 )
    steady_value = np . mean ( process_data [ key ] [ - 1 * steady_time : , 1 ] )
    steady_value_5 = steady_value * ( 1 + 0.05 )
    steady_value_10 = steady_value * ( 1 + 0.1 )
    steady_value_ng_5 = steady_value * ( 1 - 0.05 )
    steady_value_ng_10 = steady_value * ( 1 - 0.1 )
    plt . plot ( process_data [ key ] [ - 1 * steady_time : , 0 ] , [ steady_value ] * steady_time , 'b' )
    plt . plot ( process_data [ key ] [ - 1 * steady_time : , 0 ] , [ steady_value_5 ] * steady_time , 'g' )
    plt . plot ( process_data [ key ] [ - 1 * steady_time : , 0 ] , [ steady_value_ng_5 ] * steady_time , 'g' )
    plt . plot ( process_data [ key ] [ - 1 * steady_time : , 0 ] , [ steady_value_10 ] * steady_time , 'r' )
    plt . plot ( process_data [ key ] [ - 1 * steady_time : , 0 ] , [ steady_value_ng_10 ] * steady_time , 'r' )
plt . title ( para_meter [ 'title' ] + '(steady)' )
plt . xlabel ( para_meter [ 'x_axis_name' ] + '(steady)' )
plt . ylabel ( para_meter [ 'y_axis_name' ] + '(steady)' )
plt . legend ( loc = 'upper left' )
plt . savefig ( pic_path_steady )
return pic_path_steady


signature:
def process_rt_data ( source_data , is_bw = False ) :
    """process data"""

implementation:
print ( "source_data length:" , len ( source_data ) )
filter_data = { }
for index in range ( 2 ) :
    filter_mask = source_data [ : , 2 ] == index
    if np . any ( filter_mask ) :
        filter_data [ index ] = sum_data ( round_data ( source_data [ filter_mask ] ) , is_bw )
return filter_data


signature:
def sum_data ( filter_data , is_bw ) :
    """caculate sum"""

implementation:
for index in range ( len ( filter_data ) - 1 ) :
    if filter_data [ index ] [ 0 ] > filter_data [ index + 1 ] [ 0 ] :
        max_index = index + 1
        break
else :
    max_index = len ( filter_data )
print ( "max_index: " , max_index + 1 )
num_jobs = int ( round ( len ( filter_data ) * 1.0 / max_index ) )
print ( "num_jobs: " , num_jobs )
dict_time = Counter ( filter_data [ : , 0 ] )
list_sum = [ ]
for time_index in range ( 1 , max_index + 1 ) :
    if dict_time . get ( time_index * 1000 , 0 ) != num_jobs :
        print ( "[WARNING] Time %d, number of data %d != num_jobs %d" % ( time_index * 1000 , dict_time . get ( time_index * 1000 , 0 ) , num_jobs ) )
        continue
    filter_mask = ( filter_data [ : , 0 ] == time_index * 1000 )
    sum_rst = np . sum ( filter_data [ filter_mask ] [ : , 1 ] )
    if is_bw :
        sum_rst = sum_rst / 1024
    list_sum . append ( [ time_index , sum_rst ] )
return np . array ( list_sum )


signature:
def round_data ( filter_data ) :
    """round the data"""

implementation:
for index , _ in enumerate ( filter_data ) :
    filter_data [ index ] [ 0 ] = round ( filter_data [ index ] [ 0 ] / 100.0 ) * 100.0
return filter_data


signature:
def import_source ( self , sheet , source , delimiter = "," ) :
    """Function :
 Save original data into specific sheet , and try to translate data to float type
 Input :
 sheet : Must be a non exists sheet
 source : File path of source"""

implementation:
# check input parameters
if ' ' in sheet :
    raise RuntimeError ( "Error sheet name: %s" % sheet )
if not source . endswith ( "txt" ) and not source . endswith ( "csv" ) :
    raise RuntimeError ( "Error source name: %s" % source )
self . source_sheet = sheet
source_data = np . loadtxt ( source , dtype = str , delimiter = delimiter )
self . source_data = { "title" : source_data [ 0 ] . tolist ( ) , "data" : source_data [ 1 : ] }
cell_format_title = self . workbook . add_format ( { 'bold' : True , 'font_name' : u'等线' , 'bg_color' : '#c5d9f1' , 'rotation' : 45 } )
cell_format = self . workbook . add_format ( { 'bold' : False , 'font_name' : u'等线' , 'num_format' : 0 } )
worksheet = self . workbook . add_worksheet ( sheet )
worksheet . write_row ( 'A1' , self . source_data [ 'title' ] , cell_format_title )
_ , col_num = self . source_data [ 'data' ] . shape
for i in range ( col_num ) :
    try :
        data_array = self . source_data [ 'data' ] [ : , i ] . astype ( float )
    except ValueError :
        data_array = self . source_data [ 'data' ] [ : , i ]
    worksheet . write_column ( 1 , i , data_array . tolist ( ) , cell_format )


signature:
def generate_chart ( self , properties ) :
    """Function :
 Generate and save chart to specific sheet .
 Input :
 sheet : If already exists , new chart will be added below .
 Otherwise , it would create a new sheet ;
 x _ axis : Specify x axis ;
 y _ axis : Specify y axis ;
 series : Specify series ;
 filters : dict type , use to filter useful data from original data ;
 title : if None , the chart will create without title ;
 x _ axis _ name : if None , use x _ axis instead ;
 y _ axis _ name : if None , use y _ axis instead ;"""

implementation:
# check input parameters
if not { 'x_axis' , 'y_axis' , 'series' , 'filters' } . issubset ( set ( properties . keys ( ) ) ) :
    raise RuntimeError ( "Error properties: %s" % properties . keys ( ) )
# generate chart
mask = self . __filter_data ( properties [ 'filters' ] )
chart = self . __generate_chart ( mask , properties )
sheet = properties [ 'sheet' ]
# Add work sheet
if sheet in self . sheet_dict . keys ( ) :
    self . sheet_dict [ sheet ] += 1
    worksheet = self . workbook . get_worksheet_by_name ( sheet )
else :
    self . sheet_dict [ sheet ] = 1
    worksheet = self . workbook . add_worksheet ( sheet )
worksheet . insert_chart ( 'B%d' % ( 5 + ( self . sheet_dict [ sheet ] - 1 ) * 35 ) , chart )


signature:
def gen_config_sheet ( self , sheetname , plist ) :
    """generate configuration"""

implementation:
worksheet_cfg = self . workbook . add_worksheet ( sheetname )
cell_format = self . workbook . add_format ( { 'bold' : False , 'font_name' : u'等线' } )
cell_format_title = self . workbook . add_format ( { 'border' : 0 , 'align' : 'center' , 'bg_color' : '#c5d9f1' , 'font_size' : 12 , 'font_name' : u'等线' , 'bold' : False } )
worksheet_cfg . set_column ( 'A:E' , 40 , cell_format )
worksheet_cfg . write_row ( 'A1' , plist [ 0 ] , cell_format_title )
for i in range ( 1 , len ( plist ) ) :
    worksheet_cfg . write_row ( 'A%d' % ( i + 1 ) , plist [ i ] , cell_format )


signature:
def gen_data_sheet ( self , datafile , para_meter , scale = 1.75 , steady_time = 300 ) :
    """datafile , sheetname , x _ axis _ name , y _ axis _ name , title ,
 Function :
 Turn realtime bw data into picture , and save into specific sheet
 Input :
 sheetname : If already exists , new chart will be added continuely .
 Otherwise , it would create new sheet ;
 x _ axis _ name : x _ axis name ;
 y _ axis _ name : y _ axis name ;
 title : picure name ;
 scale ; size of picture ."""

implementation:
filename = os . path . splitext ( os . path . split ( datafile ) [ 1 ] ) [ 0 ] [ : - 5 ]
para_meter [ 'filename' ] = filename
source_data = np . loadtxt ( datafile , dtype = int , delimiter = ',' ) [ : , : 3 ]
is_bw = 'bw' in para_meter [ 'title' ] . lower ( )
file_data = process_rt_data ( source_data , is_bw )
pic_path = generate_rt_pic ( file_data , para_meter , scale )
pic_path_steady = generate_steady_rt_pic ( file_data , para_meter , scale , steady_time )
if para_meter [ 'sheetname' ] in self . sheetname_dict . keys ( ) :
    self . sheetname_dict [ para_meter [ 'sheetname' ] ] = self . sheetname_dict [ para_meter [ 'sheetname' ] ] + 1
    chart_sheet = self . workbook . get_worksheet_by_name ( para_meter [ 'sheetname' ] )
else :
    self . sheetname_dict [ para_meter [ 'sheetname' ] ] = 1
    chart_sheet = self . workbook . add_worksheet ( para_meter [ 'sheetname' ] )
chart_sheet . insert_image ( 'B%d' % ( 5 + ( self . sheetname_dict [ para_meter [ 'sheetname' ] ] - 1 ) * 30 ) , pic_path )
chart_sheet . insert_image ( 'Q%d' % ( 5 + ( self . sheetname_dict [ para_meter [ 'sheetname' ] ] - 1 ) * 30 ) , pic_path_steady )
self . __insert_value ( chart_sheet , file_data , 5 + ( self . sheetname_dict [ para_meter [ 'sheetname' ] ] - 1 ) * 30 , steady_time )
self . pic_list . append ( pic_path )
self . pic_list . append ( pic_path_steady )


signature:
def env ( ) :
    """Verify PCI variables and construct exported variables"""

implementation:
if cij . ssh . env ( ) :
    cij . err ( "cij.pci.env: invalid SSH environment" )
    return 1
pci = cij . env_to_dict ( PREFIX , REQUIRED )
pci [ "BUS_PATH" ] = "/sys/bus/pci"
pci [ "DEV_PATH" ] = os . sep . join ( [ pci [ "BUS_PATH" ] , "devices" , pci [ "DEV_NAME" ] ] )
cij . env_export ( PREFIX , EXPORTED , pci )
return 0


signature:
def info ( txt ) :
    """Print , emphasized ' neutral ' , the given ' txt ' message"""

implementation:
print ( "%s# %s%s%s" % ( PR_EMPH_CC , get_time_stamp ( ) , txt , PR_NC ) )
sys . stdout . flush ( )


signature:
def good ( txt ) :
    """Print , emphasized ' good ' , the given ' txt ' message"""

implementation:
print ( "%s# %s%s%s" % ( PR_GOOD_CC , get_time_stamp ( ) , txt , PR_NC ) )
sys . stdout . flush ( )


signature:
def warn ( txt ) :
    """Print , emphasized ' warning ' , the given ' txt ' message"""

implementation:
print ( "%s# %s%s%s" % ( PR_WARN_CC , get_time_stamp ( ) , txt , PR_NC ) )
sys . stdout . flush ( )


signature:
def err ( txt ) :
    """Print , emphasized ' error ' , the given ' txt ' message"""

implementation:
print ( "%s# %s%s%s" % ( PR_ERR_CC , get_time_stamp ( ) , txt , PR_NC ) )
sys . stdout . flush ( )


signature:
def emph ( txt , rval = None ) :
    """Print , emphasized based on rval"""

implementation:
if rval is None : # rval is not specified , use ' neutral '
    info ( txt )
elif rval == 0 : # rval is 0 , by convention , this is ' good '
    good ( txt )
else : # any other value , considered ' bad '
    err ( txt )


signature:
def env_to_dict ( prefix , names ) :
    """Construct dict from environment variables named : PREFIX _ NAME
 @ returns dict of names"""

implementation:
env = { }
for name in names :
    env [ name ] = ENV . get ( "_" . join ( [ prefix , name ] ) )
    if env [ name ] is None :
        return None
return env


signature:
def env_export ( prefix , exported , env ) :
    """Define the list of ' exported ' variables with ' prefix ' with values from ' env '"""

implementation:
for exp in exported :
    ENV [ "_" . join ( [ prefix , exp ] ) ] = env [ exp ]


signature:
def env ( ) :
    """Verify NVME variables and construct exported variables"""

implementation:
if cij . ssh . env ( ) :
    cij . err ( "cij.nvm.env: invalid SSH environment" )
    return 1
nvm = cij . env_to_dict ( PREFIX , REQUIRED )
if "nvme" in nvm [ "DEV_NAME" ] :
    nvm [ "DEV_PATH" ] = "/dev/%s" % nvm [ "DEV_NAME" ]
else :
    nvm [ "DEV_PATH" ] = "traddr:%s" % nvm [ "DEV_NAME" ]
cij . env_export ( PREFIX , EXPORTED , nvm )
return 0


signature:
def exists ( ) :
    """Verify that the ENV defined NVMe device exists"""

implementation:
if env ( ) :
    cij . err ( "cij.nvm.exists: Invalid NVMe ENV." )
    return 1
nvm = cij . env_to_dict ( PREFIX , EXPORTED + REQUIRED )
cmd = [ '[[ -b "%s" ]]' % nvm [ "DEV_PATH" ] ]
rcode , _ , _ = cij . ssh . command ( cmd , shell = True , echo = False )
return rcode


signature:
def dev_get_rprt ( dev_name , pugrp = None , punit = None ) :
    """Get - log - page chunk information
 If the pugrp and punit is set , then provide report only for that pugrp / punit
 @ returns the first chunk in the given state if one exists , None otherwise"""

implementation:
cmd = [ "nvm_cmd" , "rprt_all" , dev_name ]
if not ( pugrp is None and punit is None ) :
    cmd = [ "nvm_cmd" , "rprt_lun" , dev_name , str ( pugrp ) , str ( punit ) ]
_ , _ , _ , struct = cij . test . command_to_struct ( cmd )
if not struct :
    return None
return struct [ "rprt_descr" ]


signature:
def dev_get_chunk ( dev_name , state , pugrp = None , punit = None ) :
    """Get a chunk - descriptor for the first chunk in the given state .
 If the pugrp and punit is set , then search only that pugrp / punit
 @ returns the first chunk in the given state if one exists , None otherwise"""

implementation:
rprt = dev_get_rprt ( dev_name , pugrp , punit )
if not rprt :
    return None
return next ( ( d for d in rprt if d [ "cs" ] == state ) , None )


signature:
def pkill ( ) :
    """Kill all of FIO processes"""

implementation:
if env ( ) :
    return 1
cmd = [ "ps -aux | grep fio | grep -v grep" ]
status , _ , _ = cij . ssh . command ( cmd , shell = True , echo = False )
if not status :
    status , _ , _ = cij . ssh . command ( [ "pkill -f fio" ] , shell = True )
    if status :
        return 1
return 0


signature:
def __parse_parms ( self ) :
    """Translate dict parameters to string"""

implementation:
args = list ( )
for key , val in self . __parm . items ( ) :
    key = key . replace ( "FIO_" , "" ) . lower ( )
    if key == "runtime" :
        args . append ( "--time_based" )
    if val is None :
        args . append ( "--%s" % key )
    else :
        args . append ( "--%s=%s" % ( key , val ) )
return args


signature:
def import_parms ( self , args ) :
    """Import external dict to internal dict"""

implementation:
for key , val in args . items ( ) :
    self . set_parm ( key , val )


signature:
def get_parm ( self , key ) :
    """Get parameter of FIO"""

implementation:
if key in self . __parm . keys ( ) :
    return self . __parm [ key ]
return None


signature:
def start ( self ) :
    """Run FIO job in thread"""

implementation:
self . __thread = Threads ( target = self . run , args = ( True , True , False ) )
self . __thread . setDaemon ( True )
self . __thread . start ( )


signature:
def run ( self , shell = True , cmdline = False , echo = True ) :
    """Run FIO job"""

implementation:
if env ( ) :
    return 1
cmd = [ "fio" ] + self . __parse_parms ( )
if cmdline :
    cij . emph ( "cij.fio.run: shell: %r, cmd: %r" % ( shell , cmd ) )
return cij . ssh . command ( cmd , shell , echo )


signature:
def extract_hook_names ( ent ) :
    """Extract hook names from the given entity"""

implementation:
hnames = [ ]
for hook in ent [ "hooks" ] [ "enter" ] + ent [ "hooks" ] [ "exit" ] :
    hname = os . path . basename ( hook [ "fpath_orig" ] )
    hname = os . path . splitext ( hname ) [ 0 ]
    hname = hname . strip ( )
    hname = hname . replace ( "_enter" , "" )
    hname = hname . replace ( "_exit" , "" )
    if hname in hnames :
        continue
    hnames . append ( hname )
hnames . sort ( )
return hnames


signature:
def tcase_comment ( tcase ) :
    """Extract testcase comment section / testcase description
 @ returns the testcase - comment from the tcase [ " fpath " ] as a list of strings"""

implementation:
src = open ( tcase [ "fpath" ] ) . read ( )
if len ( src ) < 3 :
    cij . err ( "rprtr::tcase_comment: invalid src, tcase: %r" % tcase [ "name" ] )
    return None
ext = os . path . splitext ( tcase [ "fpath" ] ) [ - 1 ]
if ext not in [ ".sh" , ".py" ] :
    cij . err ( "rprtr::tcase_comment: invalid ext: %r, tcase: %r" % ( ext , tcase [ "name" ] ) )
    return None
comment = [ ]
for line in src . splitlines ( ) [ 2 : ] :
    if ext == ".sh" and not line . startswith ( "#" ) :
        break
    elif ext == ".py" and not '"""' in line :
        break
    comment . append ( line )
return comment


signature:
def tcase_parse_descr ( tcase ) :
    """Parse descriptions from the the given tcase"""

implementation:
descr_short = "SHORT"
descr_long = "LONG"
try :
    comment = tcase_comment ( tcase )
except ( IOError , OSError , ValueError ) as exc :
    comment = [ ]
    cij . err ( "tcase_parse_descr: failed: %r, tcase: %r" % ( exc , tcase ) )
comment = [ l for l in comment if l . strip ( ) ]
# Remove empty lines
for line_number , line in enumerate ( comment ) :
    if line . startswith ( "#" ) :
        comment [ line_number ] = line [ 1 : ]
if comment :
    descr_short = comment [ 0 ]
if len ( comment ) > 1 :
    descr_long = "\n" . join ( comment [ 1 : ] )
return descr_short , descr_long


signature:
def runlogs_to_html ( run_root ) :
    """Returns content of the given ' fpath ' with HTML annotations , currently simply
 a conversion of ANSI color codes to HTML elements"""

implementation:
if not os . path . isdir ( run_root ) :
    return "CANNOT_LOCATE_LOGFILES"
hook_enter = [ ]
hook_exit = [ ]
tcase = [ ]
for fpath in glob . glob ( os . sep . join ( [ run_root , "*.log" ] ) ) :
    if "exit" in fpath :
        hook_exit . append ( fpath )
        continue
    if "hook" in fpath :
        hook_enter . append ( fpath )
        continue
    tcase . append ( fpath )
content = ""
for fpath in hook_enter + tcase + hook_exit :
    content += "# BEGIN: run-log from log_fpath: %s\n" % fpath
    content += open ( fpath , "r" ) . read ( )
    content += "# END: run-log from log_fpath: %s\n\n" % fpath
return content


signature:
def src_to_html ( fpath ) :
    """Returns content of the given ' fpath ' with HTML annotations for syntax
 highlighting"""

implementation:
if not os . path . exists ( fpath ) :
    return "COULD-NOT-FIND-TESTCASE-SRC-AT-FPATH:%r" % fpath
# NOTE : Do SYNTAX highlight ?
return open ( fpath , "r" ) . read ( )


signature:
def aux_listing ( aux_root ) :
    """Listing"""

implementation:
listing = [ ]
for root , _ , fnames in os . walk ( aux_root ) :
    count = len ( aux_root . split ( os . sep ) )
    prefix = root . split ( os . sep ) [ count : ]
    for fname in fnames :
        listing . append ( os . sep . join ( prefix + [ fname ] ) )
return listing


signature:
def process_tsuite ( tsuite ) :
    """Goes through the tsuite and processes " * . log " """

implementation:
# scoop of output from all run - logs
tsuite [ "log_content" ] = runlogs_to_html ( tsuite [ "res_root" ] )
tsuite [ "aux_list" ] = aux_listing ( tsuite [ "aux_root" ] )
tsuite [ "hnames" ] = extract_hook_names ( tsuite )
return True


signature:
def process_tcase ( tcase ) :
    """Goes through the trun and processes " run . log " """

implementation:
tcase [ "src_content" ] = src_to_html ( tcase [ "fpath" ] )
tcase [ "log_content" ] = runlogs_to_html ( tcase [ "res_root" ] )
tcase [ "aux_list" ] = aux_listing ( tcase [ "aux_root" ] )
tcase [ "descr_short" ] , tcase [ "descr_long" ] = tcase_parse_descr ( tcase )
tcase [ "hnames" ] = extract_hook_names ( tcase )
return True


signature:
def process_trun ( trun ) :
    """Goes through the trun and processes " run . log " """

implementation:
trun [ "log_content" ] = runlogs_to_html ( trun [ "res_root" ] )
trun [ "aux_list" ] = aux_listing ( trun [ "aux_root" ] )
trun [ "hnames" ] = extract_hook_names ( trun )
return True


signature:
def postprocess ( trun ) :
    """Perform postprocessing of the given test run"""

implementation:
plog = [ ]
plog . append ( ( "trun" , process_trun ( trun ) ) )
for tsuite in trun [ "testsuites" ] :
    plog . append ( ( "tsuite" , process_tsuite ( tsuite ) ) )
    for tcase in tsuite [ "testcases" ] :
        plog . append ( ( "tcase" , process_tcase ( tcase ) ) )
for task , success in plog :
    if not success :
        cij . err ( "rprtr::postprocess: FAILED for %r" % task )
return sum ( ( success for task , success in plog ) )


signature:
def rehome ( old , new , struct ) :
    """Replace all absolute paths to " re - home " it"""

implementation:
if old == new :
    return
if isinstance ( struct , list ) :
    for item in struct :
        rehome ( old , new , item )
elif isinstance ( struct , dict ) :
    for key , val in struct . iteritems ( ) :
        if isinstance ( val , ( dict , list ) ) :
            rehome ( old , new , val )
        elif "conf" in key :
            continue
        elif "orig" in key :
            continue
        elif "root" in key or "path" in key :
            struct [ key ] = struct [ key ] . replace ( old , new )


signature:
def main ( args ) :
    """Main entry point"""

implementation:
trun = cij . runner . trun_from_file ( args . trun_fpath )
rehome ( trun [ "conf" ] [ "OUTPUT" ] , args . output , trun )
postprocess ( trun )
cij . emph ( "main: reports are uses tmpl_fpath: %r" % args . tmpl_fpath )
cij . emph ( "main: reports are here args.output: %r" % args . output )
html_fpath = os . sep . join ( [ args . output , "%s.html" % args . tmpl_name ] )
cij . emph ( "html_fpath: %r" % html_fpath )
try : # Create and store HTML report
    with open ( html_fpath , 'w' ) as html_file :
        html_file . write ( dset_to_html ( trun , args . tmpl_fpath ) )
except ( IOError , OSError , ValueError ) as exc :
    import traceback
    traceback . print_exc ( )
    cij . err ( "rprtr:main: exc: %s" % exc )
    return 1
return 0


signature:
def env ( ) :
    """Verify SSH variables and construct exported variables"""

implementation:
ssh = cij . env_to_dict ( PREFIX , REQUIRED )
if "KEY" in ssh :
    ssh [ "KEY" ] = cij . util . expand_path ( ssh [ "KEY" ] )
if cij . ENV . get ( "SSH_PORT" ) is None :
    cij . ENV [ "SSH_PORT" ] = "22"
    cij . warn ( "cij.ssh.env: SSH_PORT was not set, assigned: %r" % ( cij . ENV . get ( "SSH_PORT" ) ) )
if cij . ENV . get ( "SSH_CMD_TIME" ) is None :
    cij . ENV [ "SSH_CMD_TIME" ] = "1"
    cij . warn ( "cij.ssh.env: SSH_CMD_TIME was not set, assigned: %r" % ( cij . ENV . get ( "SSH_CMD_TIME" ) ) )
return 0


signature:
def command ( cmd , shell = True , echo = True , suffix = None ) :
    """SSH : Run the given command over SSH as defined in environment"""

implementation:
if env ( ) :
    cij . err ( "cij.ssh.command: Invalid SSH environment" )
    return 1
prefix = [ ]
if cij . ENV . get ( "SSH_CMD_TIME" ) == "1" :
    prefix . append ( "/usr/bin/time" )
if cij . ENV . get ( "SSH_CMD_TIMEOUT" ) :
    prefix . append ( "timeout" )
    prefix . append ( cij . ENV . get ( "SSH_CMD_TIMEOUT" ) )
prefix . append ( "ssh" )
args = [ ]
if cij . ENV . get ( "SSH_KEY" ) :
    args . append ( "-i" )
    args . append ( cij . ENV . get ( "SSH_KEY" ) )
if cij . ENV . get ( "SSH_PORT" ) :
    args . append ( "-p" )
    args . append ( cij . ENV . get ( "SSH_PORT" ) )
args . append ( "@" . join ( [ cij . ENV . get ( "SSH_USER" ) , cij . ENV . get ( "SSH_HOST" ) ] ) )
wrapped = prefix + args + [ "'%s'" % " " . join ( cmd ) ]
if suffix :
    wrapped += suffix
return cij . util . execute ( wrapped , shell , echo )


signature:
def pull ( src , dst , folder = False ) :
    """SSH : pull data from remote linux"""

implementation:
if env ( ) :
    cij . err ( "cij.ssh.pull: Invalid SSH environment" )
    return 1
args = [ ]
if cij . ENV . get ( "SSH_KEY" ) :
    args . append ( "-i" )
    args . append ( cij . ENV . get ( "SSH_KEY" ) )
if cij . ENV . get ( "SSH_PORT" ) :
    args . append ( "-P" )
    args . append ( cij . ENV . get ( "SSH_PORT" ) )
if folder :
    args . append ( "-r" )
target = "%s:%s" % ( "@" . join ( [ cij . ENV . get ( "SSH_USER" ) , cij . ENV . get ( "SSH_HOST" ) ] ) , src )
wrapped = [ "scp" , " " . join ( args ) , target , dst ]
return cij . util . execute ( wrapped , shell = True , echo = True )


signature:
def wait ( timeout = 300 ) :
    """Wait util target connected"""

implementation:
if env ( ) :
    cij . err ( "cij.ssh.wait: Invalid SSH environment" )
    return 1
timeout_backup = cij . ENV . get ( "SSH_CMD_TIMEOUT" )
try :
    time_start = time . time ( )
    cij . ENV [ "SSH_CMD_TIMEOUT" ] = "3"
    while True :
        time_current = time . time ( )
        if ( time_current - time_start ) > timeout :
            cij . err ( "cij.ssh.wait: Timeout" )
            return 1
        status , _ , _ = command ( [ "exit" ] , shell = True , echo = False )
        if not status :
            break
    cij . info ( "cij.ssh.wait: Time elapsed: %d seconds" % ( time_current - time_start ) )
finally :
    if timeout_backup is None :
        del cij . ENV [ "SSH_CMD_TIMEOUT" ]
    else :
        cij . ENV [ "SSH_CMD_TIMEOUT" ] = timeout_backup
return 0


signature:
def reboot ( timeout = 300 , extra = "" ) :
    """Reboot target"""

implementation:
if env ( ) :
    cij . err ( "cij.ssh.reboot: Invalid SSH environment" )
    return 1
timeout_backup = cij . ENV . get ( "SSH_CMD_TIMEOUT" )
try :
    time_start = time . time ( )
    status , last_uptime , _ = command ( [ "/usr/bin/uptime -s" ] , shell = True , echo = False )
    if status :
        return 1
    cij . ENV [ "SSH_CMD_TIMEOUT" ] = "3"
    cij . info ( "cij.ssh.reboot: Target: %s" % cij . ENV . get ( "SSH_HOST" ) )
    command ( [ "reboot %s" % extra ] , shell = True , echo = False )
    while True :
        time_current = time . time ( )
        if ( time_current - time_start ) > timeout :
            cij . err ( "cij.ssh.reboot: Timeout" )
            return 1
        status , current_uptime , _ = command ( [ "/usr/bin/uptime -s" ] , shell = True , echo = False )
        if not status and current_uptime != last_uptime :
            break
    cij . info ( "cij.ssh.reboot: Time elapsed: %d seconds" % ( time_current - time_start ) )
finally :
    if timeout_backup is None :
        del cij . ENV [ "SSH_CMD_TIMEOUT" ]
    else :
        cij . ENV [ "SSH_CMD_TIMEOUT" ] = timeout_backup
return 0


signature:
def assert_that ( val , description = '' ) :
    """Factory method for the assertion builder with value to be tested and optional description ."""

implementation:
global _soft_ctx
if _soft_ctx :
    return AssertionBuilder ( val , description , 'soft' )
return AssertionBuilder ( val , description )


signature:
def contents_of ( f , encoding = 'utf-8' ) :
    """Helper to read the contents of the given file or path into a string with the given encoding .
 Encoding defaults to ' utf - 8 ' , other useful encodings are ' ascii ' and ' latin - 1 ' ."""

implementation:
try :
    contents = f . read ( )
except AttributeError :
    try :
        with open ( f , 'r' ) as fp :
            contents = fp . read ( )
    except TypeError :
        raise ValueError ( 'val must be file or path, but was type <%s>' % type ( f ) . __name__ )
    except OSError :
        if not isinstance ( f , str_types ) :
            raise ValueError ( 'val must be file or path, but was type <%s>' % type ( f ) . __name__ )
        raise
if sys . version_info [ 0 ] == 3 and type ( contents ) is bytes : # in PY3 force decoding of bytes to target encoding
    return contents . decode ( encoding , 'replace' )
elif sys . version_info [ 0 ] == 2 and encoding == 'ascii' : # in PY2 force encoding back to ascii
    return contents . encode ( 'ascii' , 'replace' )
else : # in all other cases , try to decode to target encoding
    try :
        return contents . decode ( encoding , 'replace' )
    except AttributeError :
        pass
# if all else fails , just return the contents " as is "
return contents


signature:
def soft_fail ( msg = '' ) :
    """Adds error message to soft errors list if within soft assertions context .
 Either just force test failure with the given message ."""

implementation:
global _soft_ctx
if _soft_ctx :
    global _soft_err
    _soft_err . append ( 'Fail: %s!' % msg if msg else 'Fail!' )
    return
fail ( msg )


signature:
def is_equal_to ( self , other , ** kwargs ) :
    """Asserts that val is equal to other ."""

implementation:
if self . _check_dict_like ( self . val , check_values = False , return_as_bool = True ) and self . _check_dict_like ( other , check_values = False , return_as_bool = True ) :
    if self . _dict_not_equal ( self . val , other , ignore = kwargs . get ( 'ignore' ) , include = kwargs . get ( 'include' ) ) :
        self . _dict_err ( self . val , other , ignore = kwargs . get ( 'ignore' ) , include = kwargs . get ( 'include' ) )
else :
    if self . val != other :
        self . _err ( 'Expected <%s> to be equal to <%s>, but was not.' % ( self . val , other ) )
return self


signature:
def is_not_equal_to ( self , other ) :
    """Asserts that val is not equal to other ."""

implementation:
if self . val == other :
    self . _err ( 'Expected <%s> to be not equal to <%s>, but was.' % ( self . val , other ) )
return self


signature:
def is_same_as ( self , other ) :
    """Asserts that the val is identical to other , via ' is ' compare ."""

implementation:
if self . val is not other :
    self . _err ( 'Expected <%s> to be identical to <%s>, but was not.' % ( self . val , other ) )
return self


signature:
def is_not_same_as ( self , other ) :
    """Asserts that the val is not identical to other , via ' is ' compare ."""

implementation:
if self . val is other :
    self . _err ( 'Expected <%s> to be not identical to <%s>, but was.' % ( self . val , other ) )
return self


signature:
def is_type_of ( self , some_type ) :
    """Asserts that val is of the given type ."""

implementation:
if type ( some_type ) is not type and not issubclass ( type ( some_type ) , type ) :
    raise TypeError ( 'given arg must be a type' )
if type ( self . val ) is not some_type :
    if hasattr ( self . val , '__name__' ) :
        t = self . val . __name__
    elif hasattr ( self . val , '__class__' ) :
        t = self . val . __class__ . __name__
    else :
        t = 'unknown'
    self . _err ( 'Expected <%s:%s> to be of type <%s>, but was not.' % ( self . val , t , some_type . __name__ ) )
return self


signature:
def is_instance_of ( self , some_class ) :
    """Asserts that val is an instance of the given class ."""

implementation:
try :
    if not isinstance ( self . val , some_class ) :
        if hasattr ( self . val , '__name__' ) :
            t = self . val . __name__
        elif hasattr ( self . val , '__class__' ) :
            t = self . val . __class__ . __name__
        else :
            t = 'unknown'
        self . _err ( 'Expected <%s:%s> to be instance of class <%s>, but was not.' % ( self . val , t , some_class . __name__ ) )
except TypeError :
    raise TypeError ( 'given arg must be a class' )
return self


signature:
def is_length ( self , length ) :
    """Asserts that val is the given length ."""

implementation:
if type ( length ) is not int :
    raise TypeError ( 'given arg must be an int' )
if length < 0 :
    raise ValueError ( 'given arg must be a positive int' )
if len ( self . val ) != length :
    self . _err ( 'Expected <%s> to be of length <%d>, but was <%d>.' % ( self . val , length , len ( self . val ) ) )
return self


signature:
def contains ( self , * items ) :
    """Asserts that val contains the given item or items ."""

implementation:
if len ( items ) == 0 :
    raise ValueError ( 'one or more args must be given' )
elif len ( items ) == 1 :
    if items [ 0 ] not in self . val :
        if self . _check_dict_like ( self . val , return_as_bool = True ) :
            self . _err ( 'Expected <%s> to contain key <%s>, but did not.' % ( self . val , items [ 0 ] ) )
        else :
            self . _err ( 'Expected <%s> to contain item <%s>, but did not.' % ( self . val , items [ 0 ] ) )
else :
    missing = [ ]
    for i in items :
        if i not in self . val :
            missing . append ( i )
    if missing :
        if self . _check_dict_like ( self . val , return_as_bool = True ) :
            self . _err ( 'Expected <%s> to contain keys %s, but did not contain key%s %s.' % ( self . val , self . _fmt_items ( items ) , '' if len ( missing ) == 0 else 's' , self . _fmt_items ( missing ) ) )
        else :
            self . _err ( 'Expected <%s> to contain items %s, but did not contain %s.' % ( self . val , self . _fmt_items ( items ) , self . _fmt_items ( missing ) ) )
return self


signature:
def does_not_contain ( self , * items ) :
    """Asserts that val does not contain the given item or items ."""

implementation:
if len ( items ) == 0 :
    raise ValueError ( 'one or more args must be given' )
elif len ( items ) == 1 :
    if items [ 0 ] in self . val :
        self . _err ( 'Expected <%s> to not contain item <%s>, but did.' % ( self . val , items [ 0 ] ) )
else :
    found = [ ]
    for i in items :
        if i in self . val :
            found . append ( i )
    if found :
        self . _err ( 'Expected <%s> to not contain items %s, but did contain %s.' % ( self . val , self . _fmt_items ( items ) , self . _fmt_items ( found ) ) )
return self


signature:
def contains_only ( self , * items ) :
    """Asserts that val contains only the given item or items ."""

implementation:
if len ( items ) == 0 :
    raise ValueError ( 'one or more args must be given' )
else :
    extra = [ ]
    for i in self . val :
        if i not in items :
            extra . append ( i )
    if extra :
        self . _err ( 'Expected <%s> to contain only %s, but did contain %s.' % ( self . val , self . _fmt_items ( items ) , self . _fmt_items ( extra ) ) )
    missing = [ ]
    for i in items :
        if i not in self . val :
            missing . append ( i )
    if missing :
        self . _err ( 'Expected <%s> to contain only %s, but did not contain %s.' % ( self . val , self . _fmt_items ( items ) , self . _fmt_items ( missing ) ) )
return self


signature:
def contains_sequence ( self , * items ) :
    """Asserts that val contains the given sequence of items in order ."""

implementation:
if len ( items ) == 0 :
    raise ValueError ( 'one or more args must be given' )
else :
    try :
        for i in xrange ( len ( self . val ) - len ( items ) + 1 ) :
            for j in xrange ( len ( items ) ) :
                if self . val [ i + j ] != items [ j ] :
                    break
            else :
                return self
    except TypeError :
        raise TypeError ( 'val is not iterable' )
self . _err ( 'Expected <%s> to contain sequence %s, but did not.' % ( self . val , self . _fmt_items ( items ) ) )


signature:
def contains_duplicates ( self ) :
    """Asserts that val is iterable and contains duplicate items ."""

implementation:
try :
    if len ( self . val ) != len ( set ( self . val ) ) :
        return self
except TypeError :
    raise TypeError ( 'val is not iterable' )
self . _err ( 'Expected <%s> to contain duplicates, but did not.' % self . val )


signature:
def does_not_contain_duplicates ( self ) :
    """Asserts that val is iterable and does not contain any duplicate items ."""

implementation:
try :
    if len ( self . val ) == len ( set ( self . val ) ) :
        return self
except TypeError :
    raise TypeError ( 'val is not iterable' )
self . _err ( 'Expected <%s> to not contain duplicates, but did.' % self . val )


signature:
def is_empty ( self ) :
    """Asserts that val is empty ."""

implementation:
if len ( self . val ) != 0 :
    if isinstance ( self . val , str_types ) :
        self . _err ( 'Expected <%s> to be empty string, but was not.' % self . val )
    else :
        self . _err ( 'Expected <%s> to be empty, but was not.' % self . val )
return self


signature:
def is_not_empty ( self ) :
    """Asserts that val is not empty ."""

implementation:
if len ( self . val ) == 0 :
    if isinstance ( self . val , str_types ) :
        self . _err ( 'Expected not empty string, but was empty.' )
    else :
        self . _err ( 'Expected not empty, but was empty.' )
return self


signature:
def is_in ( self , * items ) :
    """Asserts that val is equal to one of the given items ."""

implementation:
if len ( items ) == 0 :
    raise ValueError ( 'one or more args must be given' )
else :
    for i in items :
        if self . val == i :
            return self
self . _err ( 'Expected <%s> to be in %s, but was not.' % ( self . val , self . _fmt_items ( items ) ) )


signature:
def is_nan ( self ) :
    """Asserts that val is real number and NaN ( not a number ) ."""

implementation:
self . _validate_number ( )
self . _validate_real ( )
if not math . isnan ( self . val ) :
    self . _err ( 'Expected <%s> to be <NaN>, but was not.' % self . val )
return self


signature:
def is_not_nan ( self ) :
    """Asserts that val is real number and not NaN ( not a number ) ."""

implementation:
self . _validate_number ( )
self . _validate_real ( )
if math . isnan ( self . val ) :
    self . _err ( 'Expected not <NaN>, but was.' )
return self


signature:
def is_inf ( self ) :
    """Asserts that val is real number and Inf ( infinity ) ."""

implementation:
self . _validate_number ( )
self . _validate_real ( )
if not math . isinf ( self . val ) :
    self . _err ( 'Expected <%s> to be <Inf>, but was not.' % self . val )
return self


signature:
def is_not_inf ( self ) :
    """Asserts that val is real number and not Inf ( infinity ) ."""

implementation:
self . _validate_number ( )
self . _validate_real ( )
if math . isinf ( self . val ) :
    self . _err ( 'Expected not <Inf>, but was.' )
return self


signature:
def is_less_than ( self , other ) :
    """Asserts that val is numeric and is less than other ."""

implementation:
self . _validate_compareable ( other )
if self . val >= other :
    if type ( self . val ) is datetime . datetime :
        self . _err ( 'Expected <%s> to be less than <%s>, but was not.' % ( self . val . strftime ( '%Y-%m-%d %H:%M:%S' ) , other . strftime ( '%Y-%m-%d %H:%M:%S' ) ) )
    else :
        self . _err ( 'Expected <%s> to be less than <%s>, but was not.' % ( self . val , other ) )
return self


signature:
def is_between ( self , low , high ) :
    """Asserts that val is numeric and is between low and high ."""

implementation:
val_type = type ( self . val )
self . _validate_between_args ( val_type , low , high )
if self . val < low or self . val > high :
    if val_type is datetime . datetime :
        self . _err ( 'Expected <%s> to be between <%s> and <%s>, but was not.' % ( self . val . strftime ( '%Y-%m-%d %H:%M:%S' ) , low . strftime ( '%Y-%m-%d %H:%M:%S' ) , high . strftime ( '%Y-%m-%d %H:%M:%S' ) ) )
    else :
        self . _err ( 'Expected <%s> to be between <%s> and <%s>, but was not.' % ( self . val , low , high ) )
return self


signature:
def is_close_to ( self , other , tolerance ) :
    """Asserts that val is numeric and is close to other within tolerance ."""

implementation:
self . _validate_close_to_args ( self . val , other , tolerance )
if self . val < ( other - tolerance ) or self . val > ( other + tolerance ) :
    if type ( self . val ) is datetime . datetime :
        tolerance_seconds = tolerance . days * 86400 + tolerance . seconds + tolerance . microseconds / 1000000
        h , rem = divmod ( tolerance_seconds , 3600 )
        m , s = divmod ( rem , 60 )
        self . _err ( 'Expected <%s> to be close to <%s> within tolerance <%d:%02d:%02d>, but was not.' % ( self . val . strftime ( '%Y-%m-%d %H:%M:%S' ) , other . strftime ( '%Y-%m-%d %H:%M:%S' ) , h , m , s ) )
    else :
        self . _err ( 'Expected <%s> to be close to <%s> within tolerance <%s>, but was not.' % ( self . val , other , tolerance ) )
return self


signature:
def is_equal_to_ignoring_case ( self , other ) :
    """Asserts that val is case - insensitive equal to other ."""

implementation:
if not isinstance ( self . val , str_types ) :
    raise TypeError ( 'val is not a string' )
if not isinstance ( other , str_types ) :
    raise TypeError ( 'given arg must be a string' )
if self . val . lower ( ) != other . lower ( ) :
    self . _err ( 'Expected <%s> to be case-insensitive equal to <%s>, but was not.' % ( self . val , other ) )
return self


signature:
def contains_ignoring_case ( self , * items ) :
    """Asserts that val is string and contains the given item or items ."""

implementation:
if len ( items ) == 0 :
    raise ValueError ( 'one or more args must be given' )
if isinstance ( self . val , str_types ) :
    if len ( items ) == 1 :
        if not isinstance ( items [ 0 ] , str_types ) :
            raise TypeError ( 'given arg must be a string' )
        if items [ 0 ] . lower ( ) not in self . val . lower ( ) :
            self . _err ( 'Expected <%s> to case-insensitive contain item <%s>, but did not.' % ( self . val , items [ 0 ] ) )
    else :
        missing = [ ]
        for i in items :
            if not isinstance ( i , str_types ) :
                raise TypeError ( 'given args must all be strings' )
            if i . lower ( ) not in self . val . lower ( ) :
                missing . append ( i )
        if missing :
            self . _err ( 'Expected <%s> to case-insensitive contain items %s, but did not contain %s.' % ( self . val , self . _fmt_items ( items ) , self . _fmt_items ( missing ) ) )
elif isinstance ( self . val , Iterable ) :
    missing = [ ]
    for i in items :
        if not isinstance ( i , str_types ) :
            raise TypeError ( 'given args must all be strings' )
        found = False
        for v in self . val :
            if not isinstance ( v , str_types ) :
                raise TypeError ( 'val items must all be strings' )
            if i . lower ( ) == v . lower ( ) :
                found = True
                break
        if not found :
            missing . append ( i )
    if missing :
        self . _err ( 'Expected <%s> to case-insensitive contain items %s, but did not contain %s.' % ( self . val , self . _fmt_items ( items ) , self . _fmt_items ( missing ) ) )
else :
    raise TypeError ( 'val is not a string or iterable' )
return self


signature:
def starts_with ( self , prefix ) :
    """Asserts that val is string or iterable and starts with prefix ."""

implementation:
if prefix is None :
    raise TypeError ( 'given prefix arg must not be none' )
if isinstance ( self . val , str_types ) :
    if not isinstance ( prefix , str_types ) :
        raise TypeError ( 'given prefix arg must be a string' )
    if len ( prefix ) == 0 :
        raise ValueError ( 'given prefix arg must not be empty' )
    if not self . val . startswith ( prefix ) :
        self . _err ( 'Expected <%s> to start with <%s>, but did not.' % ( self . val , prefix ) )
elif isinstance ( self . val , Iterable ) :
    if len ( self . val ) == 0 :
        raise ValueError ( 'val must not be empty' )
    first = next ( iter ( self . val ) )
    if first != prefix :
        self . _err ( 'Expected %s to start with <%s>, but did not.' % ( self . val , prefix ) )
else :
    raise TypeError ( 'val is not a string or iterable' )
return self


signature:
def ends_with ( self , suffix ) :
    """Asserts that val is string or iterable and ends with suffix ."""

implementation:
if suffix is None :
    raise TypeError ( 'given suffix arg must not be none' )
if isinstance ( self . val , str_types ) :
    if not isinstance ( suffix , str_types ) :
        raise TypeError ( 'given suffix arg must be a string' )
    if len ( suffix ) == 0 :
        raise ValueError ( 'given suffix arg must not be empty' )
    if not self . val . endswith ( suffix ) :
        self . _err ( 'Expected <%s> to end with <%s>, but did not.' % ( self . val , suffix ) )
elif isinstance ( self . val , Iterable ) :
    if len ( self . val ) == 0 :
        raise ValueError ( 'val must not be empty' )
    last = None
    for last in self . val :
        pass
    if last != suffix :
        self . _err ( 'Expected %s to end with <%s>, but did not.' % ( self . val , suffix ) )
else :
    raise TypeError ( 'val is not a string or iterable' )
return self


signature:
def matches ( self , pattern ) :
    """Asserts that val is string and matches regex pattern ."""

implementation:
if not isinstance ( self . val , str_types ) :
    raise TypeError ( 'val is not a string' )
if not isinstance ( pattern , str_types ) :
    raise TypeError ( 'given pattern arg must be a string' )
if len ( pattern ) == 0 :
    raise ValueError ( 'given pattern arg must not be empty' )
if re . search ( pattern , self . val ) is None :
    self . _err ( 'Expected <%s> to match pattern <%s>, but did not.' % ( self . val , pattern ) )
return self


signature:
def is_alpha ( self ) :
    """Asserts that val is non - empty string and all characters are alphabetic ."""

implementation:
if not isinstance ( self . val , str_types ) :
    raise TypeError ( 'val is not a string' )
if len ( self . val ) == 0 :
    raise ValueError ( 'val is empty' )
if not self . val . isalpha ( ) :
    self . _err ( 'Expected <%s> to contain only alphabetic chars, but did not.' % self . val )
return self


signature:
def is_digit ( self ) :
    """Asserts that val is non - empty string and all characters are digits ."""

implementation:
if not isinstance ( self . val , str_types ) :
    raise TypeError ( 'val is not a string' )
if len ( self . val ) == 0 :
    raise ValueError ( 'val is empty' )
if not self . val . isdigit ( ) :
    self . _err ( 'Expected <%s> to contain only digits, but did not.' % self . val )
return self


signature:
def is_lower ( self ) :
    """Asserts that val is non - empty string and all characters are lowercase ."""

implementation:
if not isinstance ( self . val , str_types ) :
    raise TypeError ( 'val is not a string' )
if len ( self . val ) == 0 :
    raise ValueError ( 'val is empty' )
if self . val != self . val . lower ( ) :
    self . _err ( 'Expected <%s> to contain only lowercase chars, but did not.' % self . val )
return self


signature:
def is_upper ( self ) :
    """Asserts that val is non - empty string and all characters are uppercase ."""

implementation:
if not isinstance ( self . val , str_types ) :
    raise TypeError ( 'val is not a string' )
if len ( self . val ) == 0 :
    raise ValueError ( 'val is empty' )
if self . val != self . val . upper ( ) :
    self . _err ( 'Expected <%s> to contain only uppercase chars, but did not.' % self . val )
return self


signature:
def is_unicode ( self ) :
    """Asserts that val is a unicode string ."""

implementation:
if type ( self . val ) is not unicode :
    self . _err ( 'Expected <%s> to be unicode, but was <%s>.' % ( self . val , type ( self . val ) . __name__ ) )
return self


signature:
def is_subset_of ( self , * supersets ) :
    """Asserts that val is iterable and a subset of the given superset or flattened superset if multiple supersets are given ."""

implementation:
if not isinstance ( self . val , Iterable ) :
    raise TypeError ( 'val is not iterable' )
if len ( supersets ) == 0 :
    raise ValueError ( 'one or more superset args must be given' )
missing = [ ]
if hasattr ( self . val , 'keys' ) and callable ( getattr ( self . val , 'keys' ) ) and hasattr ( self . val , '__getitem__' ) : # flatten superset dicts
    superdict = { }
    for l , j in enumerate ( supersets ) :
        self . _check_dict_like ( j , check_values = False , name = 'arg #%d' % ( l + 1 ) )
        for k in j . keys ( ) :
            superdict . update ( { k : j [ k ] } )
    for i in self . val . keys ( ) :
        if i not in superdict :
            missing . append ( { i : self . val [ i ] } )
            # bad key
        elif self . val [ i ] != superdict [ i ] :
            missing . append ( { i : self . val [ i ] } )
            # bad val
    if missing :
        self . _err ( 'Expected <%s> to be subset of %s, but %s %s missing.' % ( self . val , self . _fmt_items ( superdict ) , self . _fmt_items ( missing ) , 'was' if len ( missing ) == 1 else 'were' ) )
else : # flatten supersets
    superset = set ( )
    for j in supersets :
        try :
            for k in j :
                superset . add ( k )
        except Exception :
            superset . add ( j )
    for i in self . val :
        if i not in superset :
            missing . append ( i )
    if missing :
        self . _err ( 'Expected <%s> to be subset of %s, but %s %s missing.' % ( self . val , self . _fmt_items ( superset ) , self . _fmt_items ( missing ) , 'was' if len ( missing ) == 1 else 'were' ) )
return self


signature:
def contains_key ( self , * keys ) :
    """Asserts the val is a dict and contains the given key or keys . Alias for contains ( ) ."""

implementation:
self . _check_dict_like ( self . val , check_values = False , check_getitem = False )
return self . contains ( * keys )


signature:
def does_not_contain_key ( self , * keys ) :
    """Asserts the val is a dict and does not contain the given key or keys . Alias for does _ not _ contain ( ) ."""

implementation:
self . _check_dict_like ( self . val , check_values = False , check_getitem = False )
return self . does_not_contain ( * keys )


signature:
def contains_value ( self , * values ) :
    """Asserts that val is a dict and contains the given value or values ."""

implementation:
self . _check_dict_like ( self . val , check_getitem = False )
if len ( values ) == 0 :
    raise ValueError ( 'one or more value args must be given' )
missing = [ ]
for v in values :
    if v not in self . val . values ( ) :
        missing . append ( v )
if missing :
    self . _err ( 'Expected <%s> to contain values %s, but did not contain %s.' % ( self . val , self . _fmt_items ( values ) , self . _fmt_items ( missing ) ) )
return self


signature:
def does_not_contain_value ( self , * values ) :
    """Asserts that val is a dict and does not contain the given value or values ."""

implementation:
self . _check_dict_like ( self . val , check_getitem = False )
if len ( values ) == 0 :
    raise ValueError ( 'one or more value args must be given' )
else :
    found = [ ]
    for v in values :
        if v in self . val . values ( ) :
            found . append ( v )
    if found :
        self . _err ( 'Expected <%s> to not contain values %s, but did contain %s.' % ( self . val , self . _fmt_items ( values ) , self . _fmt_items ( found ) ) )
return self


signature:
def contains_entry ( self , * args , ** kwargs ) :
    """Asserts that val is a dict and contains the given entry or entries ."""

implementation:
self . _check_dict_like ( self . val , check_values = False )
entries = list ( args ) + [ { k : v } for k , v in kwargs . items ( ) ]
if len ( entries ) == 0 :
    raise ValueError ( 'one or more entry args must be given' )
missing = [ ]
for e in entries :
    if type ( e ) is not dict :
        raise TypeError ( 'given entry arg must be a dict' )
    if len ( e ) != 1 :
        raise ValueError ( 'given entry args must contain exactly one key-value pair' )
    k = next ( iter ( e ) )
    if k not in self . val :
        missing . append ( e )
        # bad key
    elif self . val [ k ] != e [ k ] :
        missing . append ( e )
        # bad val
if missing :
    self . _err ( 'Expected <%s> to contain entries %s, but did not contain %s.' % ( self . val , self . _fmt_items ( entries ) , self . _fmt_items ( missing ) ) )
return self


signature:
def is_before ( self , other ) :
    """Asserts that val is a date and is before other date ."""

implementation:
if type ( self . val ) is not datetime . datetime :
    raise TypeError ( 'val must be datetime, but was type <%s>' % type ( self . val ) . __name__ )
if type ( other ) is not datetime . datetime :
    raise TypeError ( 'given arg must be datetime, but was type <%s>' % type ( other ) . __name__ )
if self . val >= other :
    self . _err ( 'Expected <%s> to be before <%s>, but was not.' % ( self . val . strftime ( '%Y-%m-%d %H:%M:%S' ) , other . strftime ( '%Y-%m-%d %H:%M:%S' ) ) )
return self


signature:
def exists ( self ) :
    """Asserts that val is a path and that it exists ."""

implementation:
if not isinstance ( self . val , str_types ) :
    raise TypeError ( 'val is not a path' )
if not os . path . exists ( self . val ) :
    self . _err ( 'Expected <%s> to exist, but was not found.' % self . val )
return self


signature:
def is_file ( self ) :
    """Asserts that val is an existing path to a file ."""

implementation:
self . exists ( )
if not os . path . isfile ( self . val ) :
    self . _err ( 'Expected <%s> to be a file, but was not.' % self . val )
return self


signature:
def is_directory ( self ) :
    """Asserts that val is an existing path to a directory ."""

implementation:
self . exists ( )
if not os . path . isdir ( self . val ) :
    self . _err ( 'Expected <%s> to be a directory, but was not.' % self . val )
return self


signature:
def is_named ( self , filename ) :
    """Asserts that val is an existing path to a file and that file is named filename ."""

implementation:
self . is_file ( )
if not isinstance ( filename , str_types ) :
    raise TypeError ( 'given filename arg must be a path' )
val_filename = os . path . basename ( os . path . abspath ( self . val ) )
if val_filename != filename :
    self . _err ( 'Expected filename <%s> to be equal to <%s>, but was not.' % ( val_filename , filename ) )
return self


signature:
def is_child_of ( self , parent ) :
    """Asserts that val is an existing path to a file and that file is a child of parent ."""

implementation:
self . is_file ( )
if not isinstance ( parent , str_types ) :
    raise TypeError ( 'given parent directory arg must be a path' )
val_abspath = os . path . abspath ( self . val )
parent_abspath = os . path . abspath ( parent )
if not val_abspath . startswith ( parent_abspath ) :
    self . _err ( 'Expected file <%s> to be a child of <%s>, but was not.' % ( val_abspath , parent_abspath ) )
return self


signature:
def extracting ( self , * names , ** kwargs ) :
    """Asserts that val is collection , then extracts the named properties or named zero - arg methods into a list ( or list of tuples if multiple names are given ) ."""

implementation:
if not isinstance ( self . val , Iterable ) :
    raise TypeError ( 'val is not iterable' )
if isinstance ( self . val , str_types ) :
    raise TypeError ( 'val must not be string' )
if len ( names ) == 0 :
    raise ValueError ( 'one or more name args must be given' )
def _extract ( x , name ) :
    if self . _check_dict_like ( x , check_values = False , return_as_bool = True ) :
        if name in x :
            return x [ name ]
        else :
            raise ValueError ( 'item keys %s did not contain key <%s>' % ( list ( x . keys ( ) ) , name ) )
    elif isinstance ( x , Iterable ) :
        self . _check_iterable ( x , name = 'item' )
        return x [ name ]
    elif hasattr ( x , name ) :
        attr = getattr ( x , name )
        if callable ( attr ) :
            try :
                return attr ( )
            except TypeError :
                raise ValueError ( 'val method <%s()> exists, but is not zero-arg method' % name )
        else :
            return attr
    else :
        raise ValueError ( 'val does not have property or zero-arg method <%s>' % name )
def _filter ( x ) :
    if 'filter' in kwargs :
        if isinstance ( kwargs [ 'filter' ] , str_types ) :
            return bool ( _extract ( x , kwargs [ 'filter' ] ) )
        elif self . _check_dict_like ( kwargs [ 'filter' ] , check_values = False , return_as_bool = True ) :
            for k in kwargs [ 'filter' ] :
                if isinstance ( k , str_types ) :
                    if _extract ( x , k ) != kwargs [ 'filter' ] [ k ] :
                        return False
            return True
        elif callable ( kwargs [ 'filter' ] ) :
            return kwargs [ 'filter' ] ( x )
        return False
    return True
def _sort ( x ) :
    if 'sort' in kwargs :
        if isinstance ( kwargs [ 'sort' ] , str_types ) :
            return _extract ( x , kwargs [ 'sort' ] )
        elif isinstance ( kwargs [ 'sort' ] , Iterable ) :
            items = [ ]
            for k in kwargs [ 'sort' ] :
                if isinstance ( k , str_types ) :
                    items . append ( _extract ( x , k ) )
            return tuple ( items )
        elif callable ( kwargs [ 'sort' ] ) :
            return kwargs [ 'sort' ] ( x )
    return 0
extracted = [ ]
for i in sorted ( self . val , key = lambda x : _sort ( x ) ) :
    if _filter ( i ) :
        items = [ _extract ( i , name ) for name in names ]
        extracted . append ( tuple ( items ) if len ( items ) > 1 else items [ 0 ] )
return AssertionBuilder ( extracted , self . description , self . kind )


signature:
def raises ( self , ex ) :
    """Asserts that val is callable and that when called raises the given error ."""

implementation:
if not callable ( self . val ) :
    raise TypeError ( 'val must be callable' )
if not issubclass ( ex , BaseException ) :
    raise TypeError ( 'given arg must be exception' )
return AssertionBuilder ( self . val , self . description , self . kind , ex )


signature:
def when_called_with ( self , * some_args , ** some_kwargs ) :
    """Asserts the val callable when invoked with the given args and kwargs raises the expected exception ."""

implementation:
if not self . expected :
    raise TypeError ( 'expected exception not set, raises() must be called first' )
try :
    self . val ( * some_args , ** some_kwargs )
except BaseException as e :
    if issubclass ( type ( e ) , self . expected ) : # chain on with exception message as val
        return AssertionBuilder ( str ( e ) , self . description , self . kind )
    else : # got exception , but wrong type , so raise
        self . _err ( 'Expected <%s> to raise <%s> when called with (%s), but raised <%s>.' % ( self . val . __name__ , self . expected . __name__ , self . _fmt_args_kwargs ( * some_args , ** some_kwargs ) , type ( e ) . __name__ ) )
# didn ' t fail as expected , so raise
self . _err ( 'Expected <%s> to raise <%s> when called with (%s).' % ( self . val . __name__ , self . expected . __name__ , self . _fmt_args_kwargs ( * some_args , ** some_kwargs ) ) )


signature:
def _err ( self , msg ) :
    """Helper to raise an AssertionError , and optionally prepend custom description ."""

implementation:
out = '%s%s' % ( '[%s] ' % self . description if len ( self . description ) > 0 else '' , msg )
if self . kind == 'warn' :
    print ( out )
    return self
elif self . kind == 'soft' :
    global _soft_err
    _soft_err . append ( out )
    return self
else :
    raise AssertionError ( out )


signature:
def _fmt_args_kwargs ( self , * some_args , ** some_kwargs ) :
    """Helper to convert the given args and kwargs into a string ."""

implementation:
if some_args :
    out_args = str ( some_args ) . lstrip ( '(' ) . rstrip ( ',)' )
if some_kwargs :
    out_kwargs = ', ' . join ( [ str ( i ) . lstrip ( '(' ) . rstrip ( ')' ) . replace ( ', ' , ': ' ) for i in [ ( k , some_kwargs [ k ] ) for k in sorted ( some_kwargs . keys ( ) ) ] ] )
if some_args and some_kwargs :
    return out_args + ', ' + out_kwargs
elif some_args :
    return out_args
elif some_kwargs :
    return out_kwargs
else :
    return ''


signature:
def generate_words ( files ) :
    """Transform list of files to list of words ,
 removing new line character
 and replace name entity ' < NE > . . . < / NE > ' and abbreviation ' < AB > . . . < / AB > ' symbol"""

implementation:
repls = { '<NE>' : '' , '</NE>' : '' , '<AB>' : '' , '</AB>' : '' }
words_all = [ ]
for i , file in enumerate ( files ) :
    lines = open ( file , 'r' )
    for line in lines :
        line = reduce ( lambda a , kv : a . replace ( * kv ) , repls . items ( ) , line )
        words = [ word for word in line . split ( "|" ) if word is not '\n' ]
        words_all . extend ( words )
return words_all


signature:
def create_char_dataframe ( words ) :
    """Give list of input tokenized words ,
 create dataframe of characters where first character of
 the word is tagged as 1 , otherwise 0
 Example
 [ ' กิน ' , ' หมด ' ] to dataframe of
 [ { ' char ' : ' ก ' , ' type ' : . . . , ' target ' : 1 } , . . . ,
 { ' char ' : ' ด ' , ' type ' : . . . , ' target ' : 0 } ]"""

implementation:
char_dict = [ ]
for word in words :
    for i , char in enumerate ( word ) :
        if i == 0 :
            char_dict . append ( { 'char' : char , 'type' : CHAR_TYPE_FLATTEN . get ( char , 'o' ) , 'target' : True } )
        else :
            char_dict . append ( { 'char' : char , 'type' : CHAR_TYPE_FLATTEN . get ( char , 'o' ) , 'target' : False } )
return pd . DataFrame ( char_dict )


signature:
def generate_best_dataset ( best_path , output_path = 'cleaned_data' , create_val = False ) :
    """Generate CSV file for training and testing data
 Input
 best _ path : str , path to BEST folder which contains unzipped subfolder
 ' article ' , ' encyclopedia ' , ' news ' , ' novel '
 cleaned _ data : str , path to output folder , the cleaned data will be saved
 in the given folder name where training set will be stored in ` train ` folder
 and testing set will be stored on ` test ` folder
 create _ val : boolean , True or False , if True , divide training set into training set and
 validation set in ` val ` folder"""

implementation:
if not os . path . isdir ( output_path ) :
    os . mkdir ( output_path )
if not os . path . isdir ( os . path . join ( output_path , 'train' ) ) :
    os . makedirs ( os . path . join ( output_path , 'train' ) )
if not os . path . isdir ( os . path . join ( output_path , 'test' ) ) :
    os . makedirs ( os . path . join ( output_path , 'test' ) )
if not os . path . isdir ( os . path . join ( output_path , 'val' ) ) and create_val :
    os . makedirs ( os . path . join ( output_path , 'val' ) )
for article_type in article_types :
    files = glob ( os . path . join ( best_path , article_type , '*.txt' ) )
    files_train , files_test = train_test_split ( files , random_state = 0 , test_size = 0.1 )
    if create_val :
        files_train , files_val = train_test_split ( files_train , random_state = 0 , test_size = 0.1 )
        val_words = generate_words ( files_val )
        val_df = create_char_dataframe ( val_words )
        val_df . to_csv ( os . path . join ( output_path , 'val' , 'df_best_{}_val.csv' . format ( article_type ) ) , index = False )
    train_words = generate_words ( files_train )
    test_words = generate_words ( files_test )
    train_df = create_char_dataframe ( train_words )
    test_df = create_char_dataframe ( test_words )
    train_df . to_csv ( os . path . join ( output_path , 'train' , 'df_best_{}_train.csv' . format ( article_type ) ) , index = False )
    test_df . to_csv ( os . path . join ( output_path , 'test' , 'df_best_{}_test.csv' . format ( article_type ) ) , index = False )
    print ( "Save {} to CSV file" . format ( article_type ) )


signature:
def prepare_feature ( best_processed_path , option = 'train' ) :
    """Transform processed path into feature matrix and output array
 Input
 best _ processed _ path : str , path to processed BEST dataset
 option : str , ' train ' or ' test '"""

implementation:
# padding for training and testing set
n_pad = 21
n_pad_2 = int ( ( n_pad - 1 ) / 2 )
pad = [ { 'char' : ' ' , 'type' : 'p' , 'target' : True } ]
df_pad = pd . DataFrame ( pad * n_pad_2 )
df = [ ]
for article_type in article_types :
    df . append ( pd . read_csv ( os . path . join ( best_processed_path , option , 'df_best_{}_{}.csv' . format ( article_type , option ) ) ) )
df = pd . concat ( df )
df = pd . concat ( ( df_pad , df , df_pad ) )
# pad with empty string feature
df [ 'char' ] = df [ 'char' ] . map ( lambda x : CHARS_MAP . get ( x , 80 ) )
df [ 'type' ] = df [ 'type' ] . map ( lambda x : CHAR_TYPES_MAP . get ( x , 4 ) )
df_pad = create_n_gram_df ( df , n_pad = n_pad )
char_row = [ 'char' + str ( i + 1 ) for i in range ( n_pad_2 ) ] + [ 'char-' + str ( i + 1 ) for i in range ( n_pad_2 ) ] + [ 'char' ]
type_row = [ 'type' + str ( i + 1 ) for i in range ( n_pad_2 ) ] + [ 'type-' + str ( i + 1 ) for i in range ( n_pad_2 ) ] + [ 'type' ]
x_char = df_pad [ char_row ] . as_matrix ( )
x_type = df_pad [ type_row ] . as_matrix ( )
y = df_pad [ 'target' ] . astype ( int ) . as_matrix ( )
return x_char , x_type , y