signature:
def corrmtx ( x_input , m , method = 'autocorrelation' ) :
    r"""Correlation matrix
 This function is used by PSD estimator functions . It generates
 the correlation matrix from a correlation data set and a maximum lag .
 : param array x : autocorrelation samples ( 1D )
 : param int m : the maximum lag
 Depending on the choice of the method , the correlation matrix has different
 sizes , but the number of rows is always m + 1.
 Method can be :
 * ' autocorrelation ' : ( default ) X is the ( n + m ) - by - ( m + 1 ) rectangular Toeplitz
 matrix derived using prewindowed and postwindowed data .
 * ' prewindowed ' : X is the n - by - ( m + 1 ) rectangular Toeplitz matrix derived
 using prewindowed data only .
 * ' postwindowed ' : X is the n - by - ( m + 1 ) rectangular Toeplitz matrix that
 derived using postwindowed data only .
 * ' covariance ' : X is the ( n - m ) - by - ( m + 1 ) rectangular Toeplitz matrix
 derived using nonwindowed data .
 * ' modified ' : X is the 2 ( n - m ) - by - ( m + 1 ) modified rectangular Toeplitz
 matrix that generates an autocorrelation estimate for the length n data
 vector x , derived using forward and backward prediction error estimates .
 : return :
 * the autocorrelation matrix
 * R , the ( m + 1 ) - by - ( m + 1 ) autocorrelation matrix estimate ` ` R = X ' * X ` ` .
 . . rubric : : Algorithm details :
 The * * autocorrelation * * matrix is a : math : ` ( N + p ) \ times ( p + 1 ) ` rectangular Toeplilz
 data matrix :
 . . math : : X _ p = \ begin { pmatrix } L _ p \ \ T _ p \ \ Up \ end { pmatrix }
 where the lower triangular : math : ` p \ times ( p + 1 ) ` matrix : math : ` L _ p ` is
 . . math : : L _ p =
 \ begin { pmatrix }
 x [ 1 ] & \ cdots & 0 & 0 \ \
 \ vdots & \ ddots & \ vdots & \ vdots \ \
 x [ p ] & \ cdots & x [ 1 ] & 0
 \ end { pmatrix }
 where the rectangular : math : ` ( N - p ) \ times ( p + 1 ) ` matrix : math : ` T _ p ` is
 . . math : : T _ p =
 \ begin { pmatrix }
 x [ p + 1 ] & \ cdots & x [ 1 ] \ \
 \ vdots & \ ddots & \ vdots \ \
 x [ N - p ] & \ cdots & x [ p + 1 ] \ \
 \ vdots & \ ddots & \ vdots \ \
 x [ N ] & \ cdots & x [ N - p ]
 \ end { pmatrix }
 and where the upper triangular : math : ` p \ times ( p + 1 ) ` matrix : math : ` U _ p ` is
 . . math : : U _ p =
 \ begin { pmatrix }
 0 & x [ N ] & \ cdots & x [ N - p + 1 ] \ \
 \ vdots & \ vdots & \ ddots & \ vdots \ \
 0 & 0 & \ cdots & x [ N ]
 \ end { pmatrix }
 From this definition , the prewindowed matrix is
 . . math : : X _ p = \ begin { pmatrix } L _ p \ \ T _ p \ end { pmatrix }
 the postwindowed matrix is
 . . math : : X _ p = \ begin { pmatrix } T _ p \ \ U _ p \ end { pmatrix }
 the covariance matrix is :
 . . math : : X _ p = \ begin { pmatrix } T _ p \ end { pmatrix }
 and the modified covariance matrix is :
 . . math : : X _ p = \ begin { pmatrix } T _ p \ \ T _ p ^ * \ end { pmatrix }"""

implementation:
valid_methods = [ 'autocorrelation' , 'prewindowed' , 'postwindowed' , 'covariance' , 'modified' ]
if method not in valid_methods :
    raise ValueError ( "Invalid method. Try one of %s" % valid_methods )
from scipy . linalg import toeplitz
# create the relevant matrices that will be useful to create
# the correlation matrices
N = len ( x_input )
# FIXME : do we need a copy ?
if isinstance ( x_input , list ) :
    x = numpy . array ( x_input )
else :
    x = x_input . copy ( )
if x . dtype == complex :
    complex_type = True
else :
    complex_type = False
# Compute the Lp , Up and Tp matrices according to the requested method
if method in [ 'autocorrelation' , 'prewindowed' ] :
    Lp = toeplitz ( x [ 0 : m ] , [ 0 ] * ( m + 1 ) )
Tp = toeplitz ( x [ m : N ] , x [ m : : - 1 ] )
if method in [ 'autocorrelation' , 'postwindowed' ] :
    Up = toeplitz ( [ 0 ] * ( m + 1 ) , numpy . insert ( x [ N : N - m - 1 : - 1 ] , 0 , 0 ) )
# Create the output matrix
if method == 'autocorrelation' :
    if complex_type == True :
        C = numpy . zeros ( ( N + m , m + 1 ) , dtype = complex )
    else :
        C = numpy . zeros ( ( N + m , m + 1 ) )
    for i in range ( 0 , m ) :
        C [ i ] = Lp [ i ]
    for i in range ( m , N ) :
        C [ i ] = Tp [ i - m ]
    for i in range ( N , N + m ) :
        C [ i ] = Up [ i - N ]
elif method == 'prewindowed' :
    if complex_type == True :
        C = numpy . zeros ( ( N , m + 1 ) , dtype = complex )
    else :
        C = numpy . zeros ( ( N , m + 1 ) )
    for i in range ( 0 , m ) :
        C [ i ] = Lp [ i ]
    for i in range ( m , N ) :
        C [ i ] = Tp [ i - m ]
elif method == 'postwindowed' :
    if complex_type == True :
        C = numpy . zeros ( ( N , m + 1 ) , dtype = complex )
    else :
        C = numpy . zeros ( ( N , m + 1 ) )
    for i in range ( 0 , N - m ) :
        C [ i ] = Tp [ i ]
    for i in range ( N - m , N ) :
        C [ i ] = Up [ i - N + m ]
elif method == 'covariance' :
    return Tp
elif method == 'modified' :
    if complex_type == True :
        C = numpy . zeros ( ( 2 * ( N - m ) , m + 1 ) , dtype = complex )
    else :
        C = numpy . zeros ( ( 2 * ( N - m ) , m + 1 ) )
    for i in range ( 0 , N - m ) :
        C [ i ] = Tp [ i ]
    Tp = numpy . fliplr ( Tp . conj ( ) )
    for i in range ( N - m , 2 * ( N - m ) ) :
        C [ i ] = Tp [ i - N + m ]
return C


signature:
def csvd ( A ) :
    """SVD decomposition using numpy . linalg . svd
 : param A : a M by N matrix
 : return :
 * U , a M by M matrix
 * S the N eigen values
 * V a N by N matrix
 See : func : ` numpy . linalg . svd ` for a detailed documentation .
 Should return the same as in [ Marple ] _ , CSVD routine .
 U , S , V = numpy . linalg . svd ( A )
 U , S , V = cvsd ( A )"""

implementation:
U , S , V = numpy . linalg . svd ( A )
return U , S , V


signature:
def compatible_staticpath ( path ) :
    """Try to return a path to static the static files compatible all
 the way back to Django 1.2 . If anyone has a cleaner or better
 way to do this let me know !"""

implementation:
if VERSION >= ( 1 , 10 ) : # Since Django 1.10 , forms . Media automatically invoke static
# lazily on the path if it is relative .
    return path
try : # > = 1.4
    from django . templatetags . static import static
    return static ( path )
except ImportError :
    pass
try : # > = 1.3
    return '%s/%s' % ( settings . STATIC_URL . rstrip ( '/' ) , path )
except AttributeError :
    pass
try :
    return '%s/%s' % ( settings . PAGEDOWN_URL . rstrip ( '/' ) , path )
except AttributeError :
    pass
return '%s/%s' % ( settings . MEDIA_URL . rstrip ( '/' ) , path )


signature:
def main ( argv = None ) :
    """Main command line interface ."""

implementation:
if argv is None :
    argv = sys . argv [ 1 : ]
cli = CommandLineTool ( )
return cli . run ( argv )


signature:
def pass_from_pipe ( cls ) :
    """Return password from pipe if not on TTY , else False ."""

implementation:
is_pipe = not sys . stdin . isatty ( )
return is_pipe and cls . strip_last_newline ( sys . stdin . read ( ) )


signature:
def _load_plugins ( ) :
    """Locate all setuptools entry points by the name ' keyring backends '
 and initialize them .
 Any third - party library may register an entry point by adding the
 following to their setup . py : :
 entry _ points = {
 ' keyring . backends ' : [
 ' plugin _ name = mylib . mymodule : initialize _ func ' ,
 ` plugin _ name ` can be anything , and is only used to display the name
 of the plugin at initialization time .
 ` initialize _ func ` is optional , but will be invoked if callable ."""

implementation:
group = 'keyring.backends'
entry_points = entrypoints . get_group_all ( group = group )
for ep in entry_points :
    try :
        log . info ( 'Loading %s' , ep . name )
        init_func = ep . load ( )
        if callable ( init_func ) :
            init_func ( )
    except Exception :
        log . exception ( "Error initializing plugin %s." % ep )


signature:
def get_all_keyring ( ) :
    """Return a list of all implemented keyrings that can be constructed without
 parameters ."""

implementation:
_load_plugins ( )
viable_classes = KeyringBackend . get_viable_backends ( )
rings = util . suppress_exceptions ( viable_classes , exceptions = TypeError )
return list ( rings )


signature:
def name ( cls ) :
    """The keyring name , suitable for display .
 The name is derived from module and class name ."""

implementation:
parent , sep , mod_name = cls . __module__ . rpartition ( '.' )
mod_name = mod_name . replace ( '_' , ' ' )
return ' ' . join ( [ mod_name , cls . __name__ ] )


signature:
def get_credential ( self , service , username ) :
    """Gets the username and password for the service .
 Returns a Credential instance .
 The * username * argument is optional and may be omitted by
 the caller or ignored by the backend . Callers must use the
 returned username ."""

implementation:
# The default implementation requires a username here .
if username is not None :
    password = self . get_password ( service , username )
    if password is not None :
        return credentials . SimpleCredential ( username , password , )
return None


signature:
def get_password ( self , service , username ) :
    """Get password of the username for the service"""

implementation:
if not self . connected ( service ) : # the user pressed " cancel " when prompted to unlock their keyring .
    raise KeyringLocked ( "Failed to unlock the keyring!" )
if not self . iface . hasEntry ( self . handle , service , username , self . appid ) :
    return None
password = self . iface . readPassword ( self . handle , service , username , self . appid )
return str ( password )


signature:
def set_password ( self , service , username , password ) :
    """Set password for the username of the service"""

implementation:
if not self . connected ( service ) : # the user pressed " cancel " when prompted to unlock their keyring .
    raise PasswordSetError ( "Cancelled by user" )
self . iface . writePassword ( self . handle , service , username , password , self . appid )


signature:
def delete_password ( self , service , username ) :
    """Delete the password for the username of the service ."""

implementation:
if not self . connected ( service ) : # the user pressed " cancel " when prompted to unlock their keyring .
    raise PasswordDeleteError ( "Cancelled by user" )
if not self . iface . hasEntry ( self . handle , service , username , self . appid ) :
    raise PasswordDeleteError ( "Password not found" )
self . iface . removeEntry ( self . handle , service , username , self . appid )


signature:
def _get_env ( self , env_var ) :
    """Helper to read an environment variable"""

implementation:
value = os . environ . get ( env_var )
if not value :
    raise ValueError ( 'Missing environment variable:%s' % env_var )
return value


signature:
def get_preferred_collection ( self ) :
    """If self . preferred _ collection contains a D - Bus path ,
 the collection at that address is returned . Otherwise ,
 the default collection is returned ."""

implementation:
bus = secretstorage . dbus_init ( )
try :
    if hasattr ( self , 'preferred_collection' ) :
        collection = secretstorage . Collection ( bus , self . preferred_collection )
    else :
        collection = secretstorage . get_default_collection ( bus )
except exceptions . SecretStorageException as e :
    raise InitError ( "Failed to create the collection: %s." % e )
if collection . is_locked ( ) :
    collection . unlock ( )
    if collection . is_locked ( ) : # User dismissed the prompt
        raise KeyringLocked ( "Failed to unlock the collection!" )
return collection


implementation:
collection = self . get_preferred_collection ( )
items = collection . search_items ( { "username" : username , "service" : service } )
for item in items :
    if hasattr ( item , 'unlock' ) :
        item . unlock ( )
    if item . is_locked ( ) : # User dismissed the prompt
        raise KeyringLocked ( 'Failed to unlock the item!' )
    return item . get_secret ( ) . decode ( 'utf-8' )


implementation:
collection = self . get_preferred_collection ( )
attributes = { "application" : self . appid , "service" : service , "username" : username }
label = "Password for '{}' on '{}'" . format ( username , service )
collection . create_item ( label , attributes , password , replace = True )


signature:
def delete_password ( self , service , username ) :
    """Delete the stored password ( only the first one )"""

implementation:
collection = self . get_preferred_collection ( )
items = collection . search_items ( { "username" : username , "service" : service } )
for item in items :
    return item . delete ( )
raise PasswordDeleteError ( "No such password!" )


signature:
def unpack ( word ) :
    r"""> > > PackedAttributes . unpack ( 0)
 > > > PackedAttributes . unpack ( ' \ x00 \ x00 \ x00 \ x01 ' )
 > > > PackedAttributes . unpack ( ' abcd ' )
 1633837924"""

implementation:
if not isinstance ( word , str ) :
    return word
val , = struct . unpack ( '!I' , word . encode ( 'ascii' ) )
return val


signature:
def backends ( cls ) :
    """Discover all keyrings for chaining ."""

implementation:
allowed = ( keyring for keyring in filter ( backend . _limit , backend . get_all_keyring ( ) ) if not isinstance ( keyring , ChainerBackend ) and keyring . priority > 0 )
return sorted ( allowed , key = backend . by_priority , reverse = True )


signature:
def set_keyring ( keyring ) :
    """Set current keyring backend ."""

implementation:
global _keyring_backend
if not isinstance ( keyring , backend . KeyringBackend ) :
    raise TypeError ( "The keyring must be a subclass of KeyringBackend" )
_keyring_backend = keyring


signature:
def disable ( ) :
    """Configure the null keyring as the default ."""

implementation:
root = platform . config_root ( )
try :
    os . makedirs ( root )
except OSError :
    pass
filename = os . path . join ( root , 'keyringrc.cfg' )
if os . path . exists ( filename ) :
    msg = "Refusing to overwrite {filename}" . format ( ** locals ( ) )
    raise RuntimeError ( msg )
with open ( filename , 'w' ) as file :
    file . write ( '[backend]\ndefault-keyring=keyring.backends.null.Keyring' )


signature:
def init_backend ( limit = None ) :
    """Load a keyring specified in the config file or infer the best available .
 Limit , if supplied , should be a callable taking a backend and returning
 True if that backend should be included for consideration ."""

implementation:
# save the limit for the chainer to honor
backend . _limit = limit
# get all keyrings passing the limit filter
keyrings = filter ( limit , backend . get_all_keyring ( ) )
set_keyring ( load_env ( ) or load_config ( ) or max ( keyrings , default = fail . Keyring ( ) , key = backend . by_priority ) )


signature:
def _load_keyring_class ( keyring_name ) :
    """Load the keyring class indicated by name .
 These popular names are tested to ensure their presence .
 > > > popular _ names = [
 . . . ' keyring . backends . Windows . WinVaultKeyring ' ,
 . . . ' keyring . backends . OS _ X . Keyring ' ,
 . . . ' keyring . backends . kwallet . DBusKeyring ' ,
 . . . ' keyring . backends . SecretService . Keyring ' ,
 > > > list ( map ( _ load _ keyring _ class , popular _ names ) )
 These legacy names are retained for compatibility .
 > > > legacy _ names = [
 > > > list ( map ( _ load _ keyring _ class , legacy _ names ) )"""

implementation:
module_name , sep , class_name = keyring_name . rpartition ( '.' )
__import__ ( module_name )
module = sys . modules [ module_name ]
return getattr ( module , class_name )


signature:
def load_config ( ) :
    """Load a keyring using the config file in the config root ."""

implementation:
filename = 'keyringrc.cfg'
keyring_cfg = os . path . join ( platform . config_root ( ) , filename )
if not os . path . exists ( keyring_cfg ) :
    return
config = configparser . RawConfigParser ( )
config . read ( keyring_cfg )
_load_keyring_path ( config )
# load the keyring class name , and then load this keyring
try :
    if config . has_section ( "backend" ) :
        keyring_name = config . get ( "backend" , "default-keyring" ) . strip ( )
    else :
        raise configparser . NoOptionError ( 'backend' , 'default-keyring' )
except ( configparser . NoOptionError , ImportError ) :
    logger = logging . getLogger ( 'keyring' )
    logger . warning ( "Keyring config file contains incorrect values.\n" + "Config file: %s" % keyring_cfg )
    return
return load_keyring ( keyring_name )


signature:
def _load_keyring_path ( config ) :
    "load the keyring - path option ( if present )"

implementation:
try :
    path = config . get ( "backend" , "keyring-path" ) . strip ( )
    sys . path . insert ( 0 , path )
except ( configparser . NoOptionError , configparser . NoSectionError ) :
    pass


signature:
def _data_root_Linux ( ) :
    """Use freedesktop . org Base Dir Specfication to determine storage
 location ."""

implementation:
fallback = os . path . expanduser ( '~/.local/share' )
root = os . environ . get ( 'XDG_DATA_HOME' , None ) or fallback
return os . path . join ( root , 'python_keyring' )


signature:
def _check_old_config_root ( ) :
    """Prior versions of keyring would search for the config
 in XDG _ DATA _ HOME , but should probably have been
 searching for config in XDG _ CONFIG _ HOME . If the
 config exists in the former but not in the latter ,
 raise a RuntimeError to force the change ."""

implementation:
# disable the check - once is enough and avoids infinite loop
globals ( ) [ '_check_old_config_root' ] = lambda : None
config_file_new = os . path . join ( _config_root_Linux ( ) , 'keyringrc.cfg' )
config_file_old = os . path . join ( _data_root_Linux ( ) , 'keyringrc.cfg' )
if os . path . isfile ( config_file_old ) and not os . path . isfile ( config_file_new ) :
    msg = ( "Keyring config exists only in the old location " "{config_file_old} and should be moved to {config_file_new} " "to work with this version of keyring." )
    raise RuntimeError ( msg . format ( ** locals ( ) ) )


signature:
def _config_root_Linux ( ) :
    """Use freedesktop . org Base Dir Specfication to determine config
 location ."""

implementation:
_check_old_config_root ( )
fallback = os . path . expanduser ( '~/.local/share' )
key = 'XDG_CONFIG_HOME'
root = os . environ . get ( key , None ) or fallback
return os . path . join ( root , 'python_keyring' )


signature:
def make_formatter ( format_name ) :
    """Returns a callable that outputs the data . Defaults to print ."""

implementation:
if "json" in format_name :
    from json import dumps
    import datetime
    def jsonhandler ( obj ) :
        obj . isoformat ( ) if isinstance ( obj , ( datetime . datetime , datetime . date ) ) else obj
    if format_name == "prettyjson" :
        def jsondumps ( data ) :
            return dumps ( data , default = jsonhandler , indent = 2 , separators = ( ',' , ': ' ) )
    else :
        def jsondumps ( data ) :
            return dumps ( data , default = jsonhandler )
    def jsonify ( data ) :
        if isinstance ( data , dict ) :
            print ( jsondumps ( data ) )
        elif isinstance ( data , list ) :
            print ( jsondumps ( [ device . _asdict ( ) for device in data ] ) )
        else :
            print ( dumps ( { 'result' : data } ) )
    return jsonify
else :
    def printer ( data ) :
        if isinstance ( data , dict ) :
            print ( data )
        else :
            for row in data :
                print ( row )
    return printer


signature:
def argparser ( ) :
    """Constructs the ArgumentParser for the CLI"""

implementation:
parser = ArgumentParser ( prog = 'pynetgear' )
parser . add_argument ( "--format" , choices = [ 'json' , 'prettyjson' , 'py' ] , default = 'prettyjson' )
router_args = parser . add_argument_group ( "router connection config" )
router_args . add_argument ( "--host" , help = "Hostname for the router" )
router_args . add_argument ( "--user" , help = "Account for login" )
router_args . add_argument ( "--port" , help = "Port exposed on the router" )
router_args . add_argument ( "--login-v2" , help = "Force the use of the cookie-based authentication" , dest = "force_login_v2" , default = False , action = "store_true" )
router_args . add_argument ( "--password" , help = "Not required with a wired connection." + "Optionally, set the PYNETGEAR_PASSWORD environment variable" )
router_args . add_argument ( "--url" , help = "Overrides host:port and ssl with url to router" )
router_args . add_argument ( "--no-ssl" , dest = "ssl" , default = True , action = "store_false" , help = "Connect with https" )
subparsers = parser . add_subparsers ( description = "Runs subcommand against the specified router" , dest = "subcommand" )
block_parser = subparsers . add_parser ( "block_device" , help = "Blocks a device from connecting by mac address" )
block_parser . add_argument ( "--mac-addr" )
allow_parser = subparsers . add_parser ( "allow_device" , help = "Allows a device with the mac address to connect" )
allow_parser . add_argument ( "--mac-addr" )
subparsers . add_parser ( "login" , help = "Attempts to login to router." )
attached_devices = subparsers . add_parser ( "attached_devices" , help = "Outputs all attached devices" )
attached_devices . add_argument ( "-v" , "--verbose" , action = "store_true" , default = False , help = "Choose between verbose and slower or terse and fast." )
subparsers . add_parser ( "traffic_meter" , help = "Output router's traffic meter data" )
return parser


signature:
def run_subcommand ( netgear , args ) :
    """Runs the subcommand configured in args on the netgear session"""

implementation:
subcommand = args . subcommand
if subcommand == "block_device" or subcommand == "allow_device" :
    return netgear . allow_block_device ( args . mac_addr , BLOCK if subcommand == "block_device" else ALLOW )
if subcommand == "attached_devices" :
    if args . verbose :
        return netgear . get_attached_devices_2 ( )
    else :
        return netgear . get_attached_devices ( )
if subcommand == 'traffic_meter' :
    return netgear . get_traffic_meter ( )
if subcommand == 'login' :
    return netgear . login ( )
print ( "Unknown subcommand" )


signature:
def main ( ) :
    """Scan for devices and print results ."""

implementation:
args = argparser ( ) . parse_args ( sys . argv [ 1 : ] )
password = os . environ . get ( 'PYNETGEAR_PASSWORD' ) or args . password
netgear = Netgear ( password , args . host , args . user , args . port , args . ssl , args . url , args . force_login_v2 )
results = run_subcommand ( netgear , args )
formatter = make_formatter ( args . format )
if results is None :
    print ( "Error communicating with the Netgear router" )
else :
    formatter ( results )


signature:
def autodetect_url ( ) :
    """Try to autodetect the base URL of the router SOAP service .
 Returns None if it can ' t be found ."""

implementation:
for url in [ "http://routerlogin.net:5000" , "https://routerlogin.net" , "http://routerlogin.net" ] :
    try :
        r = requests . get ( url + "/soap/server_sa/" , headers = _get_soap_headers ( "Test:1" , "test" ) , verify = False )
        if r . status_code == 200 :
            return url
    except requests . exceptions . RequestException :
        pass
return None


signature:
def _xml_get ( e , name ) :
    """Returns the value of the subnode " name " of element e .
 Returns None if the subnode doesn ' t exist"""

implementation:
r = e . find ( name )
if r is not None :
    return r . text
return None


signature:
def _convert ( value , to_type , default = None ) :
    """Convert value to to _ type , returns default if fails ."""

implementation:
try :
    return default if value is None else to_type ( value )
except ValueError : # If value could not be converted
    return default


signature:
def login ( self ) :
    """Login to the router .
 Will be called automatically by other actions ."""

implementation:
if not self . force_login_v2 :
    v1_result = self . login_v1 ( )
    if v1_result :
        return v1_result
return self . login_v2 ( )


signature:
def get_attached_devices ( self ) :
    """Return list of connected devices to the router .
 Returns None if error occurred ."""

implementation:
_LOGGER . info ( "Get attached devices" )
success , response = self . _make_request ( SERVICE_DEVICE_INFO , "GetAttachDevice" )
if not success :
    _LOGGER . error ( "Get attached devices failed" )
    return None
success , node = _find_node ( response . text , ".//GetAttachDeviceResponse/NewAttachDevice" )
if not success :
    return None
devices = [ ]
# Netgear inserts a double - encoded value for " unknown " devices
decoded = node . text . strip ( ) . replace ( UNKNOWN_DEVICE_ENCODED , UNKNOWN_DEVICE_DECODED )
if not decoded or decoded == "0" :
    _LOGGER . error ( "Can't parse attached devices string" )
    _LOGGER . debug ( node . text . strip ( ) )
    return devices
entries = decoded . split ( "@" )
# First element is the total device count
entry_count = None
if len ( entries ) > 1 :
    entry_count = _convert ( entries . pop ( 0 ) , int )
if entry_count is not None and entry_count != len ( entries ) :
    _LOGGER . info ( """Number of devices should \
                 be: %d but is: %d""" , entry_count , len ( entries ) )
for entry in entries :
    info = entry . split ( ";" )
    if len ( info ) == 0 :
        continue
    # Not all routers will report those
    signal = None
    link_type = None
    link_rate = None
    allow_or_block = None
    if len ( info ) >= 8 :
        allow_or_block = info [ 7 ]
    if len ( info ) >= 7 :
        link_type = info [ 4 ]
        link_rate = _convert ( info [ 5 ] , int )
        signal = _convert ( info [ 6 ] , int )
    if len ( info ) < 4 :
        _LOGGER . warning ( "Unexpected entry: %s" , info )
        continue
    ipv4 , name , mac = info [ 1 : 4 ]
    devices . append ( Device ( name , ipv4 , mac , link_type , signal , link_rate , allow_or_block , None , None , None , None ) )
return devices


signature:
def get_attached_devices_2 ( self ) :
    """Return list of connected devices to the router with details .
 This call is slower and probably heavier on the router load .
 Returns None if error occurred ."""

implementation:
_LOGGER . info ( "Get attached devices 2" )
success , response = self . _make_request ( SERVICE_DEVICE_INFO , "GetAttachDevice2" )
if not success :
    return None
success , devices_node = _find_node ( response . text , ".//GetAttachDevice2Response/NewAttachDevice" )
if not success :
    return None
xml_devices = devices_node . findall ( "Device" )
devices = [ ]
for d in xml_devices :
    ip = _xml_get ( d , 'IP' )
    name = _xml_get ( d , 'Name' )
    mac = _xml_get ( d , 'MAC' )
    signal = _convert ( _xml_get ( d , 'SignalStrength' ) , int )
    link_type = _xml_get ( d , 'ConnectionType' )
    link_rate = _xml_get ( d , 'Linkspeed' )
    allow_or_block = _xml_get ( d , 'AllowOrBlock' )
    device_type = _convert ( _xml_get ( d , 'DeviceType' ) , int )
    device_model = _xml_get ( d , 'DeviceModel' )
    ssid = _xml_get ( d , 'SSID' )
    conn_ap_mac = _xml_get ( d , 'ConnAPMAC' )
    devices . append ( Device ( name , ip , mac , link_type , signal , link_rate , allow_or_block , device_type , device_model , ssid , conn_ap_mac ) )
return devices


signature:
def config_start ( self ) :
    """Start a configuration session .
 For managing router admin functionality ( ie allowing / blocking devices )"""

implementation:
_LOGGER . info ( "Config start" )
success , _ = self . _make_request ( SERVICE_DEVICE_CONFIG , "ConfigurationStarted" , { "NewSessionID" : SESSION_ID } )
self . config_started = success
return success


signature:
def config_finish ( self ) :
    """End of a configuration session .
 Tells the router we ' re done managing admin functionality ."""

implementation:
_LOGGER . info ( "Config finish" )
if not self . config_started :
    return True
success , _ = self . _make_request ( SERVICE_DEVICE_CONFIG , "ConfigurationFinished" , { "NewStatus" : "ChangesApplied" } )
self . config_started = not success
return success


signature:
def allow_block_device ( self , mac_addr , device_status = BLOCK ) :
    """Allow or Block a device via its Mac Address .
 Pass in the mac address for the device that you want to set . Pass in the
 device _ status you wish to set the device to : Allow ( allow device to access the
 network ) or Block ( block the device from accessing the network ) ."""

implementation:
_LOGGER . info ( "Allow block device" )
if self . config_started :
    _LOGGER . error ( "Inconsistant configuration state, configuration already started" )
    return False
if not self . config_start ( ) :
    _LOGGER . error ( "Could not start configuration" )
    return False
success , _ = self . _make_request ( SERVICE_DEVICE_CONFIG , "SetBlockDeviceByMAC" , { "NewAllowOrBlock" : device_status , "NewMACAddress" : mac_addr } )
if not success :
    _LOGGER . error ( "Could not successfully call allow/block device" )
    return False
if not self . config_finish ( ) :
    _LOGGER . error ( "Inconsistant configuration state, configuration already finished" )
    return False
return True


signature:
def _make_request ( self , service , method , params = None , body = "" , need_auth = True ) :
    """Make an API request to the router ."""

implementation:
# If we have no cookie ( v2 ) or never called login before ( v1)
# and we need auth , the request will fail for sure .
if need_auth and not self . cookie :
    if not self . login ( ) :
        return False , None
headers = self . _get_headers ( service , method , need_auth )
if not body :
    if not params :
        params = ""
    if isinstance ( params , dict ) :
        _map = params
        params = ""
        for k in _map :
            params += "<" + k + ">" + _map [ k ] + "</" + k + ">\n"
    body = CALL_BODY . format ( service = SERVICE_PREFIX + service , method = method , params = params )
message = SOAP_REQUEST . format ( session_id = SESSION_ID , body = body )
try :
    response = requests . post ( self . soap_url , headers = headers , data = message , timeout = 30 , verify = False )
    if need_auth and _is_unauthorized_response ( response ) : # let ' s discard the cookie because it probably expired ( v2)
    # or the IP - bound ( ? ) session expired ( v1)
        self . cookie = None
        _LOGGER . warning ( "Unauthorized response, let's login and retry..." )
        if self . login ( ) : # reset headers with new cookie first
            headers = self . _get_headers ( service , method , need_auth )
            response = requests . post ( self . soap_url , headers = headers , data = message , timeout = 30 , verify = False )
    success = _is_valid_response ( response )
    if not success :
        _LOGGER . error ( "Invalid response" )
        _LOGGER . debug ( "%s\n%s\n%s" , response . status_code , str ( response . headers ) , response . text )
    return success , response
except requests . exceptions . RequestException :
    _LOGGER . exception ( "Error talking to API" )
    # Maybe one day we will distinguish between
    # different errors . .
    return False , None


signature:
def ip2long ( ip ) :
    """Wrapper function for IPv4 and IPv6 converters .
 : arg ip : IPv4 or IPv6 address"""

implementation:
try :
    return int ( binascii . hexlify ( socket . inet_aton ( ip ) ) , 16 )
except socket . error :
    return int ( binascii . hexlify ( socket . inet_pton ( socket . AF_INET6 , ip ) ) , 16 )


signature:
def str2fp ( data ) :
    """Convert bytes data to file handle object ( StringIO or BytesIO ) .
 : arg data : String data to transform"""

implementation: return BytesIO ( bytearray ( data , const . ENCODING ) ) if const . PY3 else StringIO ( data )


signature:
def _setup_segments ( self ) :
    """Parses the database file to determine what kind of database is
 being used and setup segment sizes and start points that will
 be used by the seek * ( ) methods later ."""

implementation:
self . _databaseType = const . COUNTRY_EDITION
self . _recordLength = const . STANDARD_RECORD_LENGTH
self . _databaseSegments = const . COUNTRY_BEGIN
filepos = self . _fp . tell ( )
self . _fp . seek ( - 3 , os . SEEK_END )
for i in range ( const . STRUCTURE_INFO_MAX_SIZE ) :
    chars = chr ( 255 ) * 3
    delim = self . _fp . read ( 3 )
    if PY3 and type ( delim ) is bytes :
        delim = delim . decode ( ENCODING )
    if PY2 :
        chars = chars . decode ( ENCODING )
        if type ( delim ) is str :
            delim = delim . decode ( ENCODING )
    if delim == chars :
        byte = self . _fp . read ( 1 )
        self . _databaseType = ord ( byte )
        # Compatibility with databases from April 2003 and earlier
        if self . _databaseType >= 106 :
            self . _databaseType -= 105
        if self . _databaseType == const . REGION_EDITION_REV0 :
            self . _databaseSegments = const . STATE_BEGIN_REV0
        elif self . _databaseType == const . REGION_EDITION_REV1 :
            self . _databaseSegments = const . STATE_BEGIN_REV1
        elif self . _databaseType in ( const . CITY_EDITION_REV0 , const . CITY_EDITION_REV1 , const . CITY_EDITION_REV1_V6 , const . ORG_EDITION , const . ISP_EDITION , const . NETSPEED_EDITION_REV1 , const . NETSPEED_EDITION_REV1_V6 , const . ASNUM_EDITION , const . ASNUM_EDITION_V6 ) :
            self . _databaseSegments = 0
            buf = self . _fp . read ( const . SEGMENT_RECORD_LENGTH )
            if PY3 and type ( buf ) is bytes :
                buf = buf . decode ( ENCODING )
            for j in range ( const . SEGMENT_RECORD_LENGTH ) :
                self . _databaseSegments += ( ord ( buf [ j ] ) << ( j * 8 ) )
            LONG_RECORDS = ( const . ORG_EDITION , const . ISP_EDITION )
            if self . _databaseType in LONG_RECORDS :
                self . _recordLength = const . ORG_RECORD_LENGTH
        break
    else :
        self . _fp . seek ( - 4 , os . SEEK_CUR )
self . _fp . seek ( filepos , os . SEEK_SET )


signature:
def _seek_country ( self , ipnum ) :
    """Using the record length and appropriate start points , seek to the
 country that corresponds to the converted IP address integer .
 Return offset of record .
 : arg ipnum : Result of ip2long conversion"""

implementation:
try :
    offset = 0
    seek_depth = 127 if len ( str ( ipnum ) ) > 10 else 31
    for depth in range ( seek_depth , - 1 , - 1 ) :
        if self . _flags & const . MEMORY_CACHE :
            startIndex = 2 * self . _recordLength * offset
            endIndex = startIndex + ( 2 * self . _recordLength )
            buf = self . _memory [ startIndex : endIndex ]
        else :
            startIndex = 2 * self . _recordLength * offset
            readLength = 2 * self . _recordLength
            try :
                self . _lock . acquire ( )
                self . _fp . seek ( startIndex , os . SEEK_SET )
                buf = self . _fp . read ( readLength )
            finally :
                self . _lock . release ( )
        if PY3 and type ( buf ) is bytes :
            buf = buf . decode ( ENCODING )
        x = [ 0 , 0 ]
        for i in range ( 2 ) :
            for j in range ( self . _recordLength ) :
                byte = buf [ self . _recordLength * i + j ]
                x [ i ] += ord ( byte ) << ( j * 8 )
        if ipnum & ( 1 << depth ) :
            if x [ 1 ] >= self . _databaseSegments :
                self . _netmask = seek_depth - depth + 1
                return x [ 1 ]
            offset = x [ 1 ]
        else :
            if x [ 0 ] >= self . _databaseSegments :
                self . _netmask = seek_depth - depth + 1
                return x [ 0 ]
            offset = x [ 0 ]
except ( IndexError , UnicodeDecodeError ) :
    pass
raise GeoIPError ( 'Corrupt database' )


signature:
def _get_org ( self , ipnum ) :
    """Seek and return organization or ISP name for ipnum .
 Return org / isp name .
 : arg ipnum : Result of ip2long conversion"""

implementation:
seek_org = self . _seek_country ( ipnum )
if seek_org == self . _databaseSegments :
    return None
read_length = ( 2 * self . _recordLength - 1 ) * self . _databaseSegments
try :
    self . _lock . acquire ( )
    self . _fp . seek ( seek_org + read_length , os . SEEK_SET )
    buf = self . _fp . read ( const . MAX_ORG_RECORD_LENGTH )
finally :
    self . _lock . release ( )
if PY3 and type ( buf ) is bytes :
    buf = buf . decode ( ENCODING )
return buf [ : buf . index ( chr ( 0 ) ) ]


signature:
def _get_region ( self , ipnum ) :
    """Seek and return the region information .
 Returns dict containing country _ code and region _ code .
 : arg ipnum : Result of ip2long conversion"""

implementation:
region_code = None
country_code = None
seek_country = self . _seek_country ( ipnum )
def get_region_code ( offset ) :
    region1 = chr ( offset // 26 + 65 )
    region2 = chr ( offset % 26 + 65 )
    return '' . join ( [ region1 , region2 ] )
if self . _databaseType == const . REGION_EDITION_REV0 :
    seek_region = seek_country - const . STATE_BEGIN_REV0
    if seek_region >= 1000 :
        country_code = 'US'
        region_code = get_region_code ( seek_region - 1000 )
    else :
        country_code = const . COUNTRY_CODES [ seek_region ]
elif self . _databaseType == const . REGION_EDITION_REV1 :
    seek_region = seek_country - const . STATE_BEGIN_REV1
    if seek_region < const . US_OFFSET :
        pass
    elif seek_region < const . CANADA_OFFSET :
        country_code = 'US'
        region_code = get_region_code ( seek_region - const . US_OFFSET )
    elif seek_region < const . WORLD_OFFSET :
        country_code = 'CA'
        region_code = get_region_code ( seek_region - const . CANADA_OFFSET )
    else :
        index = ( seek_region - const . WORLD_OFFSET ) // const . FIPS_RANGE
        if index < len ( const . COUNTRY_CODES ) :
            country_code = const . COUNTRY_CODES [ index ]
elif self . _databaseType in const . CITY_EDITIONS :
    rec = self . _get_record ( ipnum )
    region_code = rec . get ( 'region_code' )
    country_code = rec . get ( 'country_code' )
return { 'country_code' : country_code , 'region_code' : region_code }


signature:
def _get_record ( self , ipnum ) :
    """Populate location dict for converted IP .
 Returns dict with numerous location properties .
 : arg ipnum : Result of ip2long conversion"""

implementation:
seek_country = self . _seek_country ( ipnum )
if seek_country == self . _databaseSegments :
    return { }
read_length = ( 2 * self . _recordLength - 1 ) * self . _databaseSegments
try :
    self . _lock . acquire ( )
    self . _fp . seek ( seek_country + read_length , os . SEEK_SET )
    buf = self . _fp . read ( const . FULL_RECORD_LENGTH )
finally :
    self . _lock . release ( )
if PY3 and type ( buf ) is bytes :
    buf = buf . decode ( ENCODING )
record = { 'dma_code' : 0 , 'area_code' : 0 , 'metro_code' : None , 'postal_code' : None }
latitude = 0
longitude = 0
char = ord ( buf [ 0 ] )
record [ 'country_code' ] = const . COUNTRY_CODES [ char ]
record [ 'country_code3' ] = const . COUNTRY_CODES3 [ char ]
record [ 'country_name' ] = const . COUNTRY_NAMES [ char ]
record [ 'continent' ] = const . CONTINENT_NAMES [ char ]
def read_data ( buf , pos ) :
    cur = pos
    while buf [ cur ] != '\0' :
        cur += 1
    return cur , buf [ pos : cur ] if cur > pos else None
offset , record [ 'region_code' ] = read_data ( buf , 1 )
offset , record [ 'city' ] = read_data ( buf , offset + 1 )
offset , record [ 'postal_code' ] = read_data ( buf , offset + 1 )
offset = offset + 1
for j in range ( 3 ) :
    latitude += ( ord ( buf [ offset + j ] ) << ( j * 8 ) )
for j in range ( 3 ) :
    longitude += ( ord ( buf [ offset + j + 3 ] ) << ( j * 8 ) )
record [ 'latitude' ] = ( latitude / 10000.0 ) - 180.0
record [ 'longitude' ] = ( longitude / 10000.0 ) - 180.0
if self . _databaseType in ( const . CITY_EDITION_REV1 , const . CITY_EDITION_REV1_V6 ) :
    if record [ 'country_code' ] == 'US' :
        dma_area = 0
        for j in range ( 3 ) :
            dma_area += ord ( buf [ offset + j + 6 ] ) << ( j * 8 )
        record [ 'dma_code' ] = int ( floor ( dma_area / 1000 ) )
        record [ 'area_code' ] = dma_area % 1000
        record [ 'metro_code' ] = const . DMA_MAP . get ( record [ 'dma_code' ] )
params = ( record [ 'country_code' ] , record [ 'region_code' ] )
record [ 'time_zone' ] = time_zone_by_country_and_region ( * params )
return record


signature:
def _gethostbyname ( self , hostname ) :
    """Hostname lookup method , supports both IPv4 and IPv6."""

implementation:
if self . _databaseType in const . IPV6_EDITIONS :
    response = socket . getaddrinfo ( hostname , 0 , socket . AF_INET6 )
    family , socktype , proto , canonname , sockaddr = response [ 0 ]
    address , port , flow , scope = sockaddr
    return address
else :
    return socket . gethostbyname ( hostname )


signature:
def id_by_name ( self , hostname ) :
    """Returns the database ID for specified hostname .
 The id might be useful as array index . 0 is unknown .
 : arg hostname : Hostname to get ID from ."""

implementation:
addr = self . _gethostbyname ( hostname )
return self . id_by_addr ( addr )


signature:
def id_by_addr ( self , addr ) :
    """Returns the database ID for specified address .
 The ID might be useful as array index . 0 is unknown .
 : arg addr : IPv4 or IPv6 address ( eg . 203.0.113.30)"""

implementation:
if self . _databaseType in ( const . PROXY_EDITION , const . NETSPEED_EDITION_REV1 , const . NETSPEED_EDITION_REV1_V6 ) :
    raise GeoIPError ( 'Invalid database type; this database is not supported' )
ipv = 6 if addr . find ( ':' ) >= 0 else 4
if ipv == 4 and self . _databaseType not in ( const . COUNTRY_EDITION , const . NETSPEED_EDITION ) :
    raise GeoIPError ( 'Invalid database type; this database supports IPv6 addresses, not IPv4' )
if ipv == 6 and self . _databaseType != const . COUNTRY_EDITION_V6 :
    raise GeoIPError ( 'Invalid database type; this database supports IPv4 addresses, not IPv6' )
ipnum = util . ip2long ( addr )
return self . _seek_country ( ipnum ) - const . COUNTRY_BEGIN


signature:
def country_code_by_addr ( self , addr ) :
    """Returns 2 - letter country code ( e . g . US ) from IP address .
 : arg addr : IP address ( e . g . 203.0.113.30)"""

implementation:
VALID_EDITIONS = ( const . COUNTRY_EDITION , const . COUNTRY_EDITION_V6 )
if self . _databaseType in VALID_EDITIONS :
    country_id = self . id_by_addr ( addr )
    return const . COUNTRY_CODES [ country_id ]
elif self . _databaseType in const . REGION_CITY_EDITIONS :
    return self . region_by_addr ( addr ) . get ( 'country_code' )
raise GeoIPError ( 'Invalid database type, expected Country, City or Region' )


signature:
def country_code_by_name ( self , hostname ) :
    """Returns 2 - letter country code ( e . g . US ) from hostname .
 : arg hostname : Hostname ( e . g . example . com )"""

implementation:
addr = self . _gethostbyname ( hostname )
return self . country_code_by_addr ( addr )


signature:
def netspeed_by_addr ( self , addr ) :
    """Returns NetSpeed name from address .
 : arg addr : IP address ( e . g . 203.0.113.30)"""

implementation:
if self . _databaseType == const . NETSPEED_EDITION :
    return const . NETSPEED_NAMES [ self . id_by_addr ( addr ) ]
elif self . _databaseType in ( const . NETSPEED_EDITION_REV1 , const . NETSPEED_EDITION_REV1_V6 ) :
    ipnum = util . ip2long ( addr )
    return self . _get_org ( ipnum )
raise GeoIPError ( 'Invalid database type, expected NetSpeed or NetSpeedCell' )


signature:
def netspeed_by_name ( self , hostname ) :
    """Returns NetSpeed name from hostname . Can be Unknown , Dial - up ,
 Cable , or Corporate .
 : arg hostname : Hostname ( e . g . example . com )"""

implementation:
addr = self . _gethostbyname ( hostname )
return self . netspeed_by_addr ( addr )


signature:
def country_name_by_addr ( self , addr ) :
    """Returns full country name for specified IP address .
 : arg addr : IP address ( e . g . 203.0.113.30)"""

implementation:
VALID_EDITIONS = ( const . COUNTRY_EDITION , const . COUNTRY_EDITION_V6 )
if self . _databaseType in VALID_EDITIONS :
    country_id = self . id_by_addr ( addr )
    return const . COUNTRY_NAMES [ country_id ]
elif self . _databaseType in const . CITY_EDITIONS :
    return self . record_by_addr ( addr ) . get ( 'country_name' )
else :
    message = 'Invalid database type, expected Country or City'
    raise GeoIPError ( message )


signature:
def country_name_by_name ( self , hostname ) :
    """Returns full country name for specified hostname .
 : arg hostname : Hostname ( e . g . example . com )"""

implementation:
addr = self . _gethostbyname ( hostname )
return self . country_name_by_addr ( addr )


signature:
def org_by_addr ( self , addr ) :
    """Returns Organization , ISP , or ASNum name for given IP address .
 : arg addr : IP address ( e . g . 203.0.113.30)"""

implementation:
valid = ( const . ORG_EDITION , const . ISP_EDITION , const . ASNUM_EDITION , const . ASNUM_EDITION_V6 )
if self . _databaseType not in valid :
    message = 'Invalid database type, expected Org, ISP or ASNum'
    raise GeoIPError ( message )
ipnum = util . ip2long ( addr )
return self . _get_org ( ipnum )


signature:
def org_by_name ( self , hostname ) :
    """Returns Organization , ISP , or ASNum name for given hostname .
 : arg hostname : Hostname ( e . g . example . com )"""

implementation:
addr = self . _gethostbyname ( hostname )
return self . org_by_addr ( addr )


signature:
def record_by_addr ( self , addr ) :
    """Returns dictionary with city data containing ` country _ code ` , ` country _ name ` ,
 ` region ` , ` city ` , ` postal _ code ` , ` latitude ` , ` longitude ` , ` dma _ code ` ,
 ` metro _ code ` , ` area _ code ` , ` region _ code ` and ` time _ zone ` .
 : arg addr : IP address ( e . g . 203.0.113.30)"""

implementation:
if self . _databaseType not in const . CITY_EDITIONS :
    message = 'Invalid database type, expected City'
    raise GeoIPError ( message )
ipnum = util . ip2long ( addr )
rec = self . _get_record ( ipnum )
if not rec :
    return None
return rec


signature:
def record_by_name ( self , hostname ) :
    """Returns dictionary with city data containing ` country _ code ` , ` country _ name ` ,
 ` region ` , ` city ` , ` postal _ code ` , ` latitude ` , ` longitude ` , ` dma _ code ` ,
 ` metro _ code ` , ` area _ code ` , ` region _ code ` and ` time _ zone ` .
 : arg hostname : Hostname ( e . g . example . com )"""

implementation:
addr = self . _gethostbyname ( hostname )
return self . record_by_addr ( addr )


signature:
def region_by_addr ( self , addr ) :
    """Returns dictionary containing ` country _ code ` and ` region _ code ` .
 : arg addr : IP address ( e . g . 203.0.113.30)"""

implementation:
if self . _databaseType not in const . REGION_CITY_EDITIONS :
    message = 'Invalid database type, expected Region or City'
    raise GeoIPError ( message )
ipnum = util . ip2long ( addr )
return self . _get_region ( ipnum )


signature:
def region_by_name ( self , hostname ) :
    """Returns dictionary containing ` country _ code ` and ` region _ code ` .
 : arg hostname : Hostname ( e . g . example . com )"""

implementation:
addr = self . _gethostbyname ( hostname )
return self . region_by_addr ( addr )


signature:
def time_zone_by_addr ( self , addr ) :
    """Returns time zone in tzdata format ( e . g . America / New _ York or Europe / Paris )
 : arg addr : IP address ( e . g . 203.0.113.30)"""

implementation:
if self . _databaseType not in const . CITY_EDITIONS :
    message = 'Invalid database type, expected City'
    raise GeoIPError ( message )
ipnum = util . ip2long ( addr )
return self . _get_record ( ipnum ) . get ( 'time_zone' )


signature:
def time_zone_by_name ( self , hostname ) :
    """Returns time zone in tzdata format ( e . g . America / New _ York or Europe / Paris )
 : arg hostname : Hostname ( e . g . example . com )"""

implementation:
addr = self . _gethostbyname ( hostname )
return self . time_zone_by_addr ( addr )


signature:
def time_zone_by_country_and_region ( country_code , region_code = None ) :
    """Returns time zone from country and region code .
 : arg country _ code : Country code
 : arg region _ code : Region code"""

implementation:
timezone = country_dict . get ( country_code )
if not timezone :
    return None
if isinstance ( timezone , str ) :
    return timezone
return timezone . get ( region_code )


signature:
def compress ( self , filename ) :
    """Compress a file , only if needed ."""

implementation:
compressed_filename = self . get_compressed_filename ( filename )
if not compressed_filename :
    return
self . do_compress ( filename , compressed_filename )


signature:
def get_compressed_filename ( self , filename ) :
    """If the given filename should be compressed , returns the
 compressed filename .
 A file can be compressed if :
 - It is a whitelisted extension
 - The compressed file does not exist
 - The compressed file exists by is older than the file itself
 Otherwise , it returns False ."""

implementation:
if not os . path . splitext ( filename ) [ 1 ] [ 1 : ] in self . suffixes_to_compress :
    return False
file_stats = None
compressed_stats = None
compressed_filename = '{}.{}' . format ( filename , self . suffix )
try :
    file_stats = os . stat ( filename )
    compressed_stats = os . stat ( compressed_filename )
except OSError : # FileNotFoundError is for Python3 only
    pass
if file_stats and compressed_stats :
    return ( compressed_filename if file_stats . st_mtime > compressed_stats . st_mtime else False )
else :
    return compressed_filename


signature:
def copy ( src , dst , symlink = False , rellink = False ) :
    """Copy or symlink the file ."""

implementation:
func = os . symlink if symlink else shutil . copy2
if symlink and os . path . lexists ( dst ) :
    os . remove ( dst )
if rellink : # relative symlink from dst
    func ( os . path . relpath ( src , os . path . dirname ( dst ) ) , dst )
else :
    func ( src , dst )


signature:
def url_from_path ( path ) :
    """Transform path to url , converting backslashes to slashes if needed ."""

implementation:
if os . sep != '/' :
    path = '/' . join ( path . split ( os . sep ) )
return quote ( path )


signature:
def read_markdown ( filename ) :
    """Reads markdown file , converts output and fetches title and meta - data for
 further processing ."""

implementation:
global MD
# Use utf - 8 - sig codec to remove BOM if it is present . This is only possible
# this way prior to feeding the text to the markdown parser ( which would
# also default to pure utf - 8)
with open ( filename , 'r' , encoding = 'utf-8-sig' ) as f :
    text = f . read ( )
if MD is None :
    MD = Markdown ( extensions = [ 'markdown.extensions.meta' , 'markdown.extensions.tables' ] , output_format = 'html5' )
else :
    MD . reset ( )
    # When https : / / github . com / Python - Markdown / markdown / pull / 672
    # will be available , this can be removed .
    MD . Meta = { }
# Mark HTML with Markup to prevent jinja2 autoescaping
output = { 'description' : Markup ( MD . convert ( text ) ) }
try :
    meta = MD . Meta . copy ( )
except AttributeError :
    pass
else :
    output [ 'meta' ] = meta
    try :
        output [ 'title' ] = MD . Meta [ 'title' ] [ 0 ]
    except KeyError :
        pass
return output


signature:
def load_exif ( album ) :
    """Loads the exif data of all images in an album from cache"""

implementation:
if not hasattr ( album . gallery , "exifCache" ) :
    _restore_cache ( album . gallery )
cache = album . gallery . exifCache
for media in album . medias :
    if media . type == "image" :
        key = os . path . join ( media . path , media . filename )
        if key in cache :
            media . exif = cache [ key ]


signature:
def _restore_cache ( gallery ) :
    """Restores the exif data cache from the cache file"""

implementation:
cachePath = os . path . join ( gallery . settings [ "destination" ] , ".exif_cache" )
try :
    if os . path . exists ( cachePath ) :
        with open ( cachePath , "rb" ) as cacheFile :
            gallery . exifCache = pickle . load ( cacheFile )
            logger . debug ( "Loaded cache with %d entries" , len ( gallery . exifCache ) )
    else :
        gallery . exifCache = { }
except Exception as e :
    logger . warn ( "Could not load cache: %s" , e )
    gallery . exifCache = { }


signature:
def save_cache ( gallery ) :
    """Stores the exif data of all images in the gallery"""

implementation:
if hasattr ( gallery , "exifCache" ) :
    cache = gallery . exifCache
else :
    cache = gallery . exifCache = { }
for album in gallery . albums . values ( ) :
    for image in album . images :
        cache [ os . path . join ( image . path , image . filename ) ] = image . exif
cachePath = os . path . join ( gallery . settings [ "destination" ] , ".exif_cache" )
if len ( cache ) == 0 :
    if os . path . exists ( cachePath ) :
        os . remove ( cachePath )
    return
try :
    with open ( cachePath , "wb" ) as cacheFile :
        pickle . dump ( cache , cacheFile )
        logger . debug ( "Stored cache with %d entries" , len ( gallery . exifCache ) )
except Exception as e :
    logger . warn ( "Could not store cache: %s" , e )
    os . remove ( cachePath )


signature:
def filter_nomedia ( album , settings = None ) :
    """Removes all filtered Media and subdirs from an Album"""

implementation:
nomediapath = os . path . join ( album . src_path , ".nomedia" )
if os . path . isfile ( nomediapath ) :
    if os . path . getsize ( nomediapath ) == 0 :
        logger . info ( "Ignoring album '%s' because of present 0-byte " ".nomedia file" , album . name )
        # subdirs have been added to the gallery already , remove them
        # there , too
        _remove_albums_with_subdirs ( album . gallery . albums , [ album . path ] )
        try :
            os . rmdir ( album . dst_path )
        except OSError as e : # directory was created and populated with images in a
        # previous run = > keep it
            pass
        # cannot set albums = > empty subdirs so that no albums are
        # generated
        album . subdirs = [ ]
        album . medias = [ ]
    else :
        with open ( nomediapath , "r" ) as nomediaFile :
            logger . info ( "Found a .nomedia file in %s, ignoring its " "entries" , album . name )
            ignored = nomediaFile . read ( ) . split ( "\n" )
            album . medias = [ media for media in album . medias if media . src_filename not in ignored ]
            album . subdirs = [ dirname for dirname in album . subdirs if dirname not in ignored ]
            # subdirs have been added to the gallery already , remove
            # them there , too
            _remove_albums_with_subdirs ( album . gallery . albums , ignored , album . path + os . path . sep )


signature:
def init ( path ) :
    """Copy a sample config file in the current directory ( default to
 ' sigal . conf . py ' ) , or use the provided ' path ' ."""

implementation:
if os . path . isfile ( path ) :
    print ( "Found an existing config file, will abort to keep it safe." )
    sys . exit ( 1 )
from pkg_resources import resource_string
conf = resource_string ( __name__ , 'templates/sigal.conf.py' )
with open ( path , 'w' , encoding = 'utf-8' ) as f :
    f . write ( conf . decode ( 'utf8' ) )
print ( "Sample config file created: {}" . format ( path ) )


signature:
def build ( source , destination , debug , verbose , force , config , theme , title , ncpu ) :
    """Run sigal to process a directory .
 If provided , ' source ' , ' destination ' and ' theme ' will override the
 corresponding values from the settings file ."""

implementation:
level = ( ( debug and logging . DEBUG ) or ( verbose and logging . INFO ) or logging . WARNING )
init_logging ( __name__ , level = level )
logger = logging . getLogger ( __name__ )
if not os . path . isfile ( config ) :
    logger . error ( "Settings file not found: %s" , config )
    sys . exit ( 1 )
start_time = time . time ( )
settings = read_settings ( config )
for key in ( 'source' , 'destination' , 'theme' ) :
    arg = locals ( ) [ key ]
    if arg is not None :
        settings [ key ] = os . path . abspath ( arg )
    logger . info ( "%12s : %s" , key . capitalize ( ) , settings [ key ] )
if not settings [ 'source' ] or not os . path . isdir ( settings [ 'source' ] ) :
    logger . error ( "Input directory not found: %s" , settings [ 'source' ] )
    sys . exit ( 1 )
# on windows os . path . relpath raises a ValueError if the two paths are on
# different drives , in that case we just ignore the exception as the two
# paths are anyway not relative
relative_check = True
try :
    relative_check = os . path . relpath ( settings [ 'destination' ] , settings [ 'source' ] ) . startswith ( '..' )
except ValueError :
    pass
if not relative_check :
    logger . error ( "Output directory should be outside of the input " "directory." )
    sys . exit ( 1 )
if title :
    settings [ 'title' ] = title
locale . setlocale ( locale . LC_ALL , settings [ 'locale' ] )
init_plugins ( settings )
gal = Gallery ( settings , ncpu = ncpu )
gal . build ( force = force )
# copy extra files
for src , dst in settings [ 'files_to_copy' ] :
    src = os . path . join ( settings [ 'source' ] , src )
    dst = os . path . join ( settings [ 'destination' ] , dst )
    logger . debug ( 'Copy %s to %s' , src , dst )
    copy ( src , dst , symlink = settings [ 'orig_link' ] , rellink = settings [ 'rel_link' ] )
stats = gal . stats
def format_stats ( _type ) :
    opt = [ "{} {}" . format ( stats [ _type + '_' + subtype ] , subtype ) for subtype in ( 'skipped' , 'failed' ) if stats [ _type + '_' + subtype ] > 0 ]
    opt = ' ({})' . format ( ', ' . join ( opt ) ) if opt else ''
    return '{} {}s{}' . format ( stats [ _type ] , _type , opt )
print ( 'Done.\nProcessed {} and {} in {:.2f} seconds.' . format ( format_stats ( 'image' ) , format_stats ( 'video' ) , time . time ( ) - start_time ) )


signature:
def init_plugins ( settings ) :
    """Load plugins and call register ( ) ."""

implementation:
logger = logging . getLogger ( __name__ )
logger . debug ( 'Plugin paths: %s' , settings [ 'plugin_paths' ] )
for path in settings [ 'plugin_paths' ] :
    sys . path . insert ( 0 , path )
for plugin in settings [ 'plugins' ] :
    try :
        if isinstance ( plugin , str ) :
            mod = importlib . import_module ( plugin )
            mod . register ( settings )
        else :
            plugin . register ( settings )
        logger . debug ( 'Registered plugin %s' , plugin )
    except Exception as e :
        logger . error ( 'Failed to load plugin %s: %r' , plugin , e )
for path in settings [ 'plugin_paths' ] :
    sys . path . remove ( path )


signature:
def serve ( destination , port , config ) :
    """Run a simple web server ."""

implementation:
if os . path . exists ( destination ) :
    pass
elif os . path . exists ( config ) :
    settings = read_settings ( config )
    destination = settings . get ( 'destination' )
    if not os . path . exists ( destination ) :
        sys . stderr . write ( "The '{}' directory doesn't exist, maybe try " "building first?\n" . format ( destination ) )
        sys . exit ( 1 )
else :
    sys . stderr . write ( "The {destination} directory doesn't exist " "and the config file ({config}) could not be read.\n" . format ( destination = destination , config = config ) )
    sys . exit ( 2 )
print ( 'DESTINATION : {}' . format ( destination ) )
os . chdir ( destination )
Handler = server . SimpleHTTPRequestHandler
httpd = socketserver . TCPServer ( ( "" , port ) , Handler , False )
print ( " * Running on http://127.0.0.1:{}/" . format ( port ) )
try :
    httpd . allow_reuse_address = True
    httpd . server_bind ( )
    httpd . server_activate ( )
    httpd . serve_forever ( )
except KeyboardInterrupt :
    print ( '\nAll done!' )


signature:
def set_meta ( target , keys , overwrite = False ) :
    """Write metadata keys to . md file .
 TARGET can be a media file or an album directory . KEYS are key / value pairs .
 Ex , to set the title of test . jpg to " My test image " :
 sigal set _ meta test . jpg title " My test image " """

implementation:
if not os . path . exists ( target ) :
    sys . stderr . write ( "The target {} does not exist.\n" . format ( target ) )
    sys . exit ( 1 )
if len ( keys ) < 2 or len ( keys ) % 2 > 0 :
    sys . stderr . write ( "Need an even number of arguments.\n" )
    sys . exit ( 1 )
if os . path . isdir ( target ) :
    descfile = os . path . join ( target , 'index.md' )
else :
    descfile = os . path . splitext ( target ) [ 0 ] + '.md'
if os . path . exists ( descfile ) and not overwrite :
    sys . stderr . write ( "Description file '{}' already exists. " "Use --overwrite to overwrite it.\n" . format ( descfile ) )
    sys . exit ( 2 )
with open ( descfile , "w" ) as fp :
    for i in range ( len ( keys ) // 2 ) :
        k , v = keys [ i * 2 : ( i + 1 ) * 2 ]
        fp . write ( "{}: {}\n" . format ( k . capitalize ( ) , v ) )
print ( "{} metadata key(s) written to {}" . format ( len ( keys ) // 2 , descfile ) )


signature:
def generate_image ( source , outname , settings , options = None ) :
    """Image processor , rotate and resize the image .
 : param source : path to an image
 : param outname : output filename
 : param settings : settings dict
 : param options : dict with PIL options ( quality , optimize , progressive )"""

implementation:
logger = logging . getLogger ( __name__ )
if settings [ 'use_orig' ] or source . endswith ( '.gif' ) :
    utils . copy ( source , outname , symlink = settings [ 'orig_link' ] )
    return
img = _read_image ( source )
original_format = img . format
if settings [ 'copy_exif_data' ] and settings [ 'autorotate_images' ] :
    logger . warning ( "The 'autorotate_images' and 'copy_exif_data' settings " "are not compatible because Sigal can't save the " "modified Orientation tag." )
# Preserve EXIF data
if settings [ 'copy_exif_data' ] and _has_exif_tags ( img ) :
    if options is not None :
        options = deepcopy ( options )
    else :
        options = { }
    options [ 'exif' ] = img . info [ 'exif' ]
# Rotate the img , and catch IOError when PIL fails to read EXIF
if settings [ 'autorotate_images' ] :
    try :
        img = Transpose ( ) . process ( img )
    except ( IOError , IndexError ) :
        pass
# Resize the image
if settings [ 'img_processor' ] :
    try :
        logger . debug ( 'Processor: %s' , settings [ 'img_processor' ] )
        processor_cls = getattr ( pilkit . processors , settings [ 'img_processor' ] )
    except AttributeError :
        logger . error ( 'Wrong processor name: %s' , settings [ 'img_processor' ] )
        sys . exit ( )
    width , height = settings [ 'img_size' ]
    if img . size [ 0 ] < img . size [ 1 ] : # swap target size if image is in portrait mode
        height , width = width , height
    processor = processor_cls ( width , height , upscale = False )
    img = processor . process ( img )
# signal . send ( ) does not work here as plugins can modify the image , so we
# iterate other the receivers to call them with the image .
for receiver in signals . img_resized . receivers_for ( img ) :
    img = receiver ( img , settings = settings )
outformat = img . format or original_format or 'JPEG'
logger . debug ( 'Save resized image to %s (%s)' , outname , outformat )
save_image ( img , outname , outformat , options = options , autoconvert = True )


signature:
def generate_thumbnail ( source , outname , box , fit = True , options = None , thumb_fit_centering = ( 0.5 , 0.5 ) ) :
    """Create a thumbnail image ."""

implementation:
logger = logging . getLogger ( __name__ )
img = _read_image ( source )
original_format = img . format
if fit :
    img = ImageOps . fit ( img , box , PILImage . ANTIALIAS , centering = thumb_fit_centering )
else :
    img . thumbnail ( box , PILImage . ANTIALIAS )
outformat = img . format or original_format or 'JPEG'
logger . debug ( 'Save thumnail image: %s (%s)' , outname , outformat )
save_image ( img , outname , outformat , options = options , autoconvert = True )


signature:
def process_image ( filepath , outpath , settings ) :
    """Process one image : resize , create thumbnail ."""

implementation:
logger = logging . getLogger ( __name__ )
logger . info ( 'Processing %s' , filepath )
filename = os . path . split ( filepath ) [ 1 ]
outname = os . path . join ( outpath , filename )
ext = os . path . splitext ( filename ) [ 1 ]
if ext in ( '.jpg' , '.jpeg' , '.JPG' , '.JPEG' ) :
    options = settings [ 'jpg_options' ]
elif ext == '.png' :
    options = { 'optimize' : True }
else :
    options = { }
try :
    generate_image ( filepath , outname , settings , options = options )
    if settings [ 'make_thumbs' ] :
        thumb_name = os . path . join ( outpath , get_thumb ( settings , filename ) )
        generate_thumbnail ( outname , thumb_name , settings [ 'thumb_size' ] , fit = settings [ 'thumb_fit' ] , options = options , thumb_fit_centering = settings [ "thumb_fit_centering" ] )
except Exception as e :
    logger . info ( 'Failed to process: %r' , e )
    if logger . getEffectiveLevel ( ) == logging . DEBUG :
        raise
    else :
        return Status . FAILURE
return Status . SUCCESS


signature:
def get_size ( file_path ) :
    """Return image size ( width and height ) ."""

implementation:
try :
    im = _read_image ( file_path )
except ( IOError , IndexError , TypeError , AttributeError ) as e :
    logger = logging . getLogger ( __name__ )
    logger . error ( "Could not read size of %s due to %r" , file_path , e )
else :
    width , height = im . size
    return { 'width' : width , 'height' : height }


signature:
def get_exif_data ( filename ) :
    """Return a dict with the raw EXIF data ."""

implementation:
logger = logging . getLogger ( __name__ )
img = _read_image ( filename )
try :
    exif = img . _getexif ( ) or { }
except ZeroDivisionError :
    logger . warning ( 'Failed to read EXIF data.' )
    return None
data = { TAGS . get ( tag , tag ) : value for tag , value in exif . items ( ) }
if 'GPSInfo' in data :
    try :
        data [ 'GPSInfo' ] = { GPSTAGS . get ( tag , tag ) : value for tag , value in data [ 'GPSInfo' ] . items ( ) }
    except AttributeError :
        logger = logging . getLogger ( __name__ )
        logger . info ( 'Failed to get GPS Info' )
        del data [ 'GPSInfo' ]
return data


signature:
def get_iptc_data ( filename ) :
    """Return a dict with the raw IPTC data ."""

implementation:
logger = logging . getLogger ( __name__ )
iptc_data = { }
raw_iptc = { }
# PILs IptcImagePlugin issues a SyntaxError in certain circumstances
# with malformed metadata , see PIL / IptcImagePlugin . py " , line 71.
# ( https : / / github . com / python - pillow / Pillow / blob / 9dd0348be2751beb2c617e32ff9985aa2f92ae5f / src / PIL / IptcImagePlugin . py # L71 )
try :
    img = _read_image ( filename )
    raw_iptc = IptcImagePlugin . getiptcinfo ( img )
except SyntaxError :
    logger . info ( 'IPTC Error in %s' , filename )
# IPTC fields are catalogued in :
# https : / / www . iptc . org / std / photometadata / specification / IPTC - PhotoMetadata
# 2:05 is the IPTC title property
if raw_iptc and ( 2 , 5 ) in raw_iptc :
    iptc_data [ "title" ] = raw_iptc [ ( 2 , 5 ) ] . decode ( 'utf-8' , errors = 'replace' )
# 2:120 is the IPTC description property
if raw_iptc and ( 2 , 120 ) in raw_iptc :
    iptc_data [ "description" ] = raw_iptc [ ( 2 , 120 ) ] . decode ( 'utf-8' , errors = 'replace' )
# 2:105 is the IPTC headline property
if raw_iptc and ( 2 , 105 ) in raw_iptc :
    iptc_data [ "headline" ] = raw_iptc [ ( 2 , 105 ) ] . decode ( 'utf-8' , errors = 'replace' )
return iptc_data


signature:
def dms_to_degrees ( v ) :
    """Convert degree / minute / second to decimal degrees ."""


signature:
def get_exif_tags ( data , datetime_format = '%c' ) :
    """Make a simplified version with common tags from raw EXIF data ."""

implementation:
logger = logging . getLogger ( __name__ )
simple = { }
for tag in ( 'Model' , 'Make' , 'LensModel' ) :
    if tag in data :
        if isinstance ( data [ tag ] , tuple ) :
            simple [ tag ] = data [ tag ] [ 0 ] . strip ( )
        else :
            simple [ tag ] = data [ tag ] . strip ( )
if 'FNumber' in data :
    fnumber = data [ 'FNumber' ]
    try :
        simple [ 'fstop' ] = float ( fnumber [ 0 ] ) / fnumber [ 1 ]
    except Exception :
        logger . debug ( 'Skipped invalid FNumber: %r' , fnumber , exc_info = True )
if 'FocalLength' in data :
    focal = data [ 'FocalLength' ]
    try :
        simple [ 'focal' ] = round ( float ( focal [ 0 ] ) / focal [ 1 ] )
    except Exception :
        logger . debug ( 'Skipped invalid FocalLength: %r' , focal , exc_info = True )
if 'ExposureTime' in data :
    exptime = data [ 'ExposureTime' ]
    if isinstance ( exptime , tuple ) :
        try :
            simple [ 'exposure' ] = str ( fractions . Fraction ( exptime [ 0 ] , exptime [ 1 ] ) )
        except ZeroDivisionError :
            logger . info ( 'Invalid ExposureTime: %r' , exptime )
    elif isinstance ( exptime , int ) :
        simple [ 'exposure' ] = str ( exptime )
    else :
        logger . info ( 'Unknown format for ExposureTime: %r' , exptime )
if data . get ( 'ISOSpeedRatings' ) :
    simple [ 'iso' ] = data [ 'ISOSpeedRatings' ]
if 'DateTimeOriginal' in data : # Remove null bytes at the end if necessary
    date = data [ 'DateTimeOriginal' ] . rsplit ( '\x00' ) [ 0 ]
    try :
        simple [ 'dateobj' ] = datetime . strptime ( date , '%Y:%m:%d %H:%M:%S' )
        simple [ 'datetime' ] = simple [ 'dateobj' ] . strftime ( datetime_format )
    except ( ValueError , TypeError ) as e :
        logger . info ( 'Could not parse DateTimeOriginal: %s' , e )
if 'GPSInfo' in data :
    info = data [ 'GPSInfo' ]
    lat_info = info . get ( 'GPSLatitude' )
    lon_info = info . get ( 'GPSLongitude' )
    lat_ref_info = info . get ( 'GPSLatitudeRef' )
    lon_ref_info = info . get ( 'GPSLongitudeRef' )
    if lat_info and lon_info and lat_ref_info and lon_ref_info :
        try :
            lat = dms_to_degrees ( lat_info )
            lon = dms_to_degrees ( lon_info )
        except ( ZeroDivisionError , ValueError , TypeError ) :
            logger . info ( 'Failed to read GPS info' )
        else :
            simple [ 'gps' ] = { 'lat' : - lat if lat_ref_info != 'N' else lat , 'lon' : - lon if lon_ref_info != 'E' else lon , }
return simple


signature:
def big ( self ) :
    """Path to the original image , if ` ` keep _ orig ` ` is set ( relative to the
 album directory ) . Copy the file if needed ."""

implementation:
if self . settings [ 'keep_orig' ] :
    s = self . settings
    if s [ 'use_orig' ] : # The image * is * the original , just use it
        return self . filename
    orig_path = join ( s [ 'destination' ] , self . path , s [ 'orig_dir' ] )
    check_or_create_dir ( orig_path )
    big_path = join ( orig_path , self . src_filename )
    if not isfile ( big_path ) :
        copy ( self . src_path , big_path , symlink = s [ 'orig_link' ] , rellink = self . settings [ 'rel_link' ] )
    return join ( s [ 'orig_dir' ] , self . src_filename )


signature:
def thumbnail ( self ) :
    """Path to the thumbnail image ( relative to the album directory ) ."""

implementation:
if not isfile ( self . thumb_path ) :
    self . logger . debug ( 'Generating thumbnail for %r' , self )
    path = ( self . dst_path if os . path . exists ( self . dst_path ) else self . src_path )
    try : # if thumbnail is missing ( if settings [ ' make _ thumbs ' ] is False )
        s = self . settings
        if self . type == 'image' :
            image . generate_thumbnail ( path , self . thumb_path , s [ 'thumb_size' ] , fit = s [ 'thumb_fit' ] )
        elif self . type == 'video' :
            video . generate_thumbnail ( path , self . thumb_path , s [ 'thumb_size' ] , s [ 'thumb_video_delay' ] , fit = s [ 'thumb_fit' ] , converter = s [ 'video_converter' ] )
    except Exception as e :
        self . logger . error ( 'Failed to generate thumbnail: %s' , e )
        return
return url_from_path ( self . thumb_name )


signature:
def _get_metadata ( self ) :
    """Get image metadata from filename . md : title , description , meta ."""

implementation:
self . description = ''
self . meta = { }
self . title = ''
descfile = splitext ( self . src_path ) [ 0 ] + '.md'
if isfile ( descfile ) :
    meta = read_markdown ( descfile )
    for key , val in meta . items ( ) :
        setattr ( self , key , val )


signature:
def _get_metadata ( self ) :
    """Get album metadata from ` description _ file ` ( ` index . md ` ) :
 - > title , thumbnail image , description"""

implementation:
descfile = join ( self . src_path , self . description_file )
self . description = ''
self . meta = { }
# default : get title from directory name
self . title = os . path . basename ( self . path if self . path != '.' else self . src_path )
if isfile ( descfile ) :
    meta = read_markdown ( descfile )
    for key , val in meta . items ( ) :
        setattr ( self , key , val )
try :
    self . author = self . meta [ 'author' ] [ 0 ]
except KeyError :
    self . author = self . settings . get ( 'author' )


signature:
def create_output_directories ( self ) :
    """Create output directories for thumbnails and original images ."""

implementation:
check_or_create_dir ( self . dst_path )
if self . medias :
    check_or_create_dir ( join ( self . dst_path , self . settings [ 'thumb_dir' ] ) )
if self . medias and self . settings [ 'keep_orig' ] :
    self . orig_path = join ( self . dst_path , self . settings [ 'orig_dir' ] )
    check_or_create_dir ( self . orig_path )


signature:
def albums ( self ) :
    """List of : class : ` ~ sigal . gallery . Album ` objects for each
 sub - directory ."""

implementation:
root_path = self . path if self . path != '.' else ''
return [ self . gallery . albums [ join ( root_path , path ) ] for path in self . subdirs ]


signature:
def url ( self ) :
    """URL of the album , relative to its parent ."""

implementation:
url = self . name . encode ( 'utf-8' )
return url_quote ( url ) + '/' + self . url_ext


signature:
def thumbnail ( self ) :
    """Path to the thumbnail of the album ."""

implementation:
if self . _thumbnail : # stop if it is already set
    return self . _thumbnail
# Test the thumbnail from the Markdown file .
thumbnail = self . meta . get ( 'thumbnail' , [ '' ] ) [ 0 ]
if thumbnail and isfile ( join ( self . src_path , thumbnail ) ) :
    self . _thumbnail = url_from_path ( join ( self . name , get_thumb ( self . settings , thumbnail ) ) )
    self . logger . debug ( "Thumbnail for %r : %s" , self , self . _thumbnail )
    return self . _thumbnail
else : # find and return the first landscape image
    for f in self . medias :
        ext = splitext ( f . filename ) [ 1 ]
        if ext . lower ( ) in self . settings [ 'img_extensions' ] : # Use f . size if available as it is quicker ( in cache ) , but
        # fallback to the size of src _ path if dst _ path is missing
            size = f . size
            if size is None :
                size = get_size ( f . src_path )
            if size [ 'width' ] > size [ 'height' ] :
                self . _thumbnail = ( url_quote ( self . name ) + '/' + f . thumbnail )
                self . logger . debug ( "Use 1st landscape image as thumbnail for %r : %s" , self , self . _thumbnail )
                return self . _thumbnail
    # else simply return the 1st media file
    if not self . _thumbnail and self . medias :
        for media in self . medias :
            if media . thumbnail is not None :
                self . _thumbnail = ( url_quote ( self . name ) + '/' + media . thumbnail )
                break
        else :
            self . logger . warning ( "No thumbnail found for %r" , self )
            return None
        self . logger . debug ( "Use the 1st image as thumbnail for %r : %s" , self , self . _thumbnail )
        return self . _thumbnail
    # use the thumbnail of their sub - directories
    if not self . _thumbnail :
        for path , album in self . gallery . get_albums ( self . path ) :
            if album . thumbnail :
                self . _thumbnail = ( url_quote ( self . name ) + '/' + album . thumbnail )
                self . logger . debug ( "Using thumbnail from sub-directory for %r : %s" , self , self . _thumbnail )
                return self . _thumbnail
self . logger . error ( 'Thumbnail not found for %r' , self )
return None


signature:
def breadcrumb ( self ) :
    """List of ` ` ( url , title ) ` ` tuples defining the current breadcrumb
 path ."""

implementation:
if self . path == '.' :
    return [ ]
path = self . path
breadcrumb = [ ( ( self . url_ext or '.' ) , self . title ) ]
while True :
    path = os . path . normpath ( os . path . join ( path , '..' ) )
    if path == '.' :
        break
    url = ( url_from_path ( os . path . relpath ( path , self . path ) ) + '/' + self . url_ext )
    breadcrumb . append ( ( url , self . gallery . albums [ path ] . title ) )
breadcrumb . reverse ( )
return breadcrumb


signature:
def zip ( self ) :
    """Make a ZIP archive with all media files and return its path .
 If the ` ` zip _ gallery ` ` setting is set , it contains the location of a zip
 archive with all original images of the corresponding directory ."""

implementation:
zip_gallery = self . settings [ 'zip_gallery' ]
if zip_gallery and len ( self ) > 0 :
    zip_gallery = zip_gallery . format ( album = self )
    archive_path = join ( self . dst_path , zip_gallery )
    if ( self . settings . get ( 'zip_skip_if_exists' , False ) and isfile ( archive_path ) ) :
        self . logger . debug ( "Archive %s already created, passing" , archive_path )
        return zip_gallery
    archive = zipfile . ZipFile ( archive_path , 'w' , allowZip64 = True )
    attr = ( 'src_path' if self . settings [ 'zip_media_format' ] == 'orig' else 'dst_path' )
    for p in self :
        path = getattr ( p , attr )
        try :
            archive . write ( path , os . path . split ( path ) [ 1 ] )
        except OSError as e :
            self . logger . warn ( 'Failed to add %s to the ZIP: %s' , p , e )
    archive . close ( )
    self . logger . debug ( 'Created ZIP archive %s' , archive_path )
    return zip_gallery


signature:
def get_albums ( self , path ) :
    """Return the list of all sub - directories of path ."""

implementation:
for name in self . albums [ path ] . subdirs :
    subdir = os . path . normpath ( join ( path , name ) )
    yield subdir , self . albums [ subdir ]
    for subname , album in self . get_albums ( subdir ) :
        yield subname , self . albums [ subdir ]


signature:
def build ( self , force = False ) :
    "Create the image gallery"

implementation:
if not self . albums :
    self . logger . warning ( "No albums found." )
    return
def log_func ( x ) : # 63 is the total length of progressbar , label , percentage , etc
    available_length = get_terminal_size ( ) [ 0 ] - 64
    if x and available_length > 10 :
        return x . name [ : available_length ]
    else :
        return ""
try :
    with progressbar ( self . albums . values ( ) , label = "Collecting files" , item_show_func = log_func , show_eta = False , file = self . progressbar_target ) as albums :
        media_list = [ f for album in albums for f in self . process_dir ( album , force = force ) ]
except KeyboardInterrupt :
    sys . exit ( 'Interrupted' )
bar_opt = { 'label' : "Processing files" , 'show_pos' : True , 'file' : self . progressbar_target }
failed_files = [ ]
if self . pool :
    try :
        with progressbar ( length = len ( media_list ) , ** bar_opt ) as bar :
            for res in self . pool . imap_unordered ( worker , media_list ) :
                if res :
                    failed_files . append ( res )
                bar . update ( 1 )
        self . pool . close ( )
        self . pool . join ( )
    except KeyboardInterrupt :
        self . pool . terminate ( )
        sys . exit ( 'Interrupted' )
    except pickle . PicklingError :
        self . logger . critical ( "Failed to process files with the multiprocessing feature." " This can be caused by some module import or object " "defined in the settings file, which can't be serialized." , exc_info = True )
        sys . exit ( 'Abort' )
else :
    with progressbar ( media_list , ** bar_opt ) as medias :
        for media_item in medias :
            res = process_file ( media_item )
            if res :
                failed_files . append ( res )
if failed_files :
    self . remove_files ( failed_files )
if self . settings [ 'write_html' ] :
    album_writer = AlbumPageWriter ( self . settings , index_title = self . title )
    album_list_writer = AlbumListPageWriter ( self . settings , index_title = self . title )
    with progressbar ( self . albums . values ( ) , label = "%16s" % "Writing files" , item_show_func = log_func , show_eta = False , file = self . progressbar_target ) as albums :
        for album in albums :
            if album . albums :
                if album . medias :
                    self . logger . warning ( "Album %s contains sub-albums and images. " "Please move images to their own sub-album. " "Images in album %s will not be visible." , album . title , album . title )
                album_list_writer . write ( album )
            else :
                album_writer . write ( album )
print ( '' )
signals . gallery_build . send ( self )


signature:
def process_dir ( self , album , force = False ) :
    """Process a list of images in a directory ."""

implementation:
for f in album :
    if isfile ( f . dst_path ) and not force :
        self . logger . info ( "%s exists - skipping" , f . filename )
        self . stats [ f . type + '_skipped' ] += 1
    else :
        self . stats [ f . type ] += 1
        yield ( f . type , f . path , f . filename , f . src_path , album . dst_path , self . settings )


signature:
def reduce_opacity ( im , opacity ) :
    """Returns an image with reduced opacity ."""

implementation:
assert opacity >= 0 and opacity <= 1
if im . mode != 'RGBA' :
    im = im . convert ( 'RGBA' )
else :
    im = im . copy ( )
alpha = im . split ( ) [ 3 ]
alpha = ImageEnhance . Brightness ( alpha ) . enhance ( opacity )
im . putalpha ( alpha )
return im


signature:
def watermark ( im , mark , position , opacity = 1 ) :
    """Adds a watermark to an image ."""

implementation:
if opacity < 1 :
    mark = reduce_opacity ( mark , opacity )
if im . mode != 'RGBA' :
    im = im . convert ( 'RGBA' )
# create a transparent layer the size of the image and draw the
# watermark in that layer .
layer = Image . new ( 'RGBA' , im . size , ( 0 , 0 , 0 , 0 ) )
if position == 'tile' :
    for y in range ( 0 , im . size [ 1 ] , mark . size [ 1 ] ) :
        for x in range ( 0 , im . size [ 0 ] , mark . size [ 0 ] ) :
            layer . paste ( mark , ( x , y ) )
elif position == 'scale' : # scale , but preserve the aspect ratio
    ratio = min ( float ( im . size [ 0 ] ) / mark . size [ 0 ] , float ( im . size [ 1 ] ) / mark . size [ 1 ] )
    w = int ( mark . size [ 0 ] * ratio )
    h = int ( mark . size [ 1 ] * ratio )
    mark = mark . resize ( ( w , h ) )
    layer . paste ( mark , ( int ( ( im . size [ 0 ] - w ) / 2 ) , int ( ( im . size [ 1 ] - h ) / 2 ) ) )
else :
    layer . paste ( mark , position )
# composite the watermark with the layer
return Image . composite ( layer , im , layer )


signature:
def check_subprocess ( cmd , source , outname ) :
    """Run the command to resize the video and remove the output file if the
 processing fails ."""

implementation:
logger = logging . getLogger ( __name__ )
try :
    res = subprocess . run ( cmd , stdout = subprocess . PIPE , stderr = subprocess . PIPE )
except KeyboardInterrupt :
    logger . debug ( 'Process terminated, removing file %s' , outname )
    if os . path . isfile ( outname ) :
        os . remove ( outname )
    raise
if res . returncode :
    logger . debug ( 'STDOUT:\n %s' , res . stdout . decode ( 'utf8' ) )
    logger . debug ( 'STDERR:\n %s' , res . stderr . decode ( 'utf8' ) )
    if os . path . isfile ( outname ) :
        logger . debug ( 'Removing file %s' , outname )
        os . remove ( outname )
    raise SubprocessException ( 'Failed to process ' + source )


signature:
def video_size ( source , converter = 'ffmpeg' ) :
    """Returns the dimensions of the video ."""

implementation:
res = subprocess . run ( [ converter , '-i' , source ] , stderr = subprocess . PIPE )
stderr = res . stderr . decode ( 'utf8' )
pattern = re . compile ( r'Stream.*Video.* ([0-9]+)x([0-9]+)' )
match = pattern . search ( stderr )
rot_pattern = re . compile ( r'rotate\s*:\s*-?(90|270)' )
rot_match = rot_pattern . search ( stderr )
if match :
    x , y = int ( match . groups ( ) [ 0 ] ) , int ( match . groups ( ) [ 1 ] )
else :
    x = y = 0
if rot_match :
    x , y = y , x
return x , y


signature:
def generate_video ( source , outname , settings , options = None ) :
    """Video processor .
 : param source : path to a video
 : param outname : path to the generated video
 : param settings : settings dict
 : param options : array of options passed to ffmpeg"""

implementation:
logger = logging . getLogger ( __name__ )
# Don ' t transcode if source is in the required format and
# has fitting datedimensions , copy instead .
converter = settings [ 'video_converter' ]
w_src , h_src = video_size ( source , converter = converter )
w_dst , h_dst = settings [ 'video_size' ]
logger . debug ( 'Video size: %i, %i -> %i, %i' , w_src , h_src , w_dst , h_dst )
base , src_ext = splitext ( source )
base , dst_ext = splitext ( outname )
if dst_ext == src_ext and w_src <= w_dst and h_src <= h_dst :
    logger . debug ( 'Video is smaller than the max size, copying it instead' )
    shutil . copy ( source , outname )
    return
# http : / / stackoverflow . com / questions / 8218363 / maintaining - ffmpeg - aspect - ratio
# + I made a drawing on paper to figure this out
if h_dst * w_src < h_src * w_dst : # biggest fitting dimension is height
    resize_opt = [ '-vf' , "scale=trunc(oh*a/2)*2:%i" % h_dst ]
else : # biggest fitting dimension is width
    resize_opt = [ '-vf' , "scale=%i:trunc(ow/a/2)*2" % w_dst ]
# do not resize if input dimensions are smaller than output dimensions
if w_src <= w_dst and h_src <= h_dst :
    resize_opt = [ ]
# Encoding options improved , thanks to
# http : / / ffmpeg . org / trac / ffmpeg / wiki / vpxEncodingGuide
cmd = [ converter , '-i' , source , '-y' ]
# - y to overwrite output files
if options is not None :
    cmd += options
cmd += resize_opt + [ outname ]
logger . debug ( 'Processing video: %s' , ' ' . join ( cmd ) )
check_subprocess ( cmd , source , outname )


signature:
def generate_thumbnail ( source , outname , box , delay , fit = True , options = None , converter = 'ffmpeg' ) :
    """Create a thumbnail image for the video source , based on ffmpeg ."""

implementation:
logger = logging . getLogger ( __name__ )
tmpfile = outname + ".tmp.jpg"
# dump an image of the video
cmd = [ converter , '-i' , source , '-an' , '-r' , '1' , '-ss' , delay , '-vframes' , '1' , '-y' , tmpfile ]
logger . debug ( 'Create thumbnail for video: %s' , ' ' . join ( cmd ) )
check_subprocess ( cmd , source , outname )
# use the generate _ thumbnail function from sigal . image
image . generate_thumbnail ( tmpfile , outname , box , fit = fit , options = options )
# remove the image
os . unlink ( tmpfile )


signature:
def process_video ( filepath , outpath , settings ) :
    """Process a video : resize , create thumbnail ."""

implementation:
logger = logging . getLogger ( __name__ )
filename = os . path . split ( filepath ) [ 1 ]
basename , ext = splitext ( filename )
try :
    if settings [ 'use_orig' ] and is_valid_html5_video ( ext ) :
        outname = os . path . join ( outpath , filename )
        utils . copy ( filepath , outname , symlink = settings [ 'orig_link' ] )
    else :
        valid_formats = [ 'mp4' , 'webm' ]
        video_format = settings [ 'video_format' ]
        if video_format not in valid_formats :
            logger . error ( 'Invalid video_format. Please choose one of: %s' , valid_formats )
            raise ValueError
        outname = os . path . join ( outpath , basename + '.' + video_format )
        generate_video ( filepath , outname , settings , options = settings . get ( video_format + '_options' ) )
except Exception :
    if logger . getEffectiveLevel ( ) == logging . DEBUG :
        raise
    else :
        return Status . FAILURE
if settings [ 'make_thumbs' ] :
    thumb_name = os . path . join ( outpath , get_thumb ( settings , filename ) )
    try :
        generate_thumbnail ( outname , thumb_name , settings [ 'thumb_size' ] , settings [ 'thumb_video_delay' ] , fit = settings [ 'thumb_fit' ] , options = settings [ 'jpg_options' ] , converter = settings [ 'video_converter' ] )
    except Exception :
        if logger . getEffectiveLevel ( ) == logging . DEBUG :
            raise
        else :
            return Status . FAILURE
return Status . SUCCESS


signature:
def init_logging ( name , level = logging . INFO ) :
    """Logging config
 Set the level and create a more detailed formatter for debug mode ."""

implementation:
logger = logging . getLogger ( name )
logger . setLevel ( level )
try :
    if os . isatty ( sys . stdout . fileno ( ) ) and not sys . platform . startswith ( 'win' ) :
        formatter = ColoredFormatter ( )
    elif level == logging . DEBUG :
        formatter = Formatter ( '%(levelname)s - %(message)s' )
    else :
        formatter = Formatter ( '%(message)s' )
except Exception : # This fails when running tests with click ( test _ build )
    formatter = Formatter ( '%(message)s' )
handler = logging . StreamHandler ( )
handler . setFormatter ( formatter )
logger . addHandler ( handler )


signature:
def generate_context ( self , album ) :
    """Generate the context dict for the given path ."""

implementation:
from . import __url__ as sigal_link
self . logger . info ( "Output album : %r" , album )
return { 'album' : album , 'index_title' : self . index_title , 'settings' : self . settings , 'sigal_link' : sigal_link , 'theme' : { 'name' : os . path . basename ( self . theme ) , 'url' : url_from_path ( os . path . relpath ( self . theme_path , album . dst_path ) ) } , }


signature:
def write ( self , album ) :
    """Generate the HTML page and save it ."""

implementation:
page = self . template . render ( ** self . generate_context ( album ) )
output_file = os . path . join ( album . dst_path , album . output_file )
with open ( output_file , 'w' , encoding = 'utf-8' ) as f :
    f . write ( page )


signature:
def get_thumb ( settings , filename ) :
    """Return the path to the thumb .
 examples :
 > > > default _ settings = create _ settings ( )
 > > > get _ thumb ( default _ settings , " bar / foo . jpg " )
 " bar / thumbnails / foo . jpg "
 > > > get _ thumb ( default _ settings , " bar / foo . png " )
 " bar / thumbnails / foo . png "
 for videos , it returns a jpg file :
 > > > get _ thumb ( default _ settings , " bar / foo . webm " )
 " bar / thumbnails / foo . jpg " """

implementation:
path , filen = os . path . split ( filename )
name , ext = os . path . splitext ( filen )
if ext . lower ( ) in settings [ 'video_extensions' ] :
    ext = '.jpg'
return join ( path , settings [ 'thumb_dir' ] , settings [ 'thumb_prefix' ] + name + settings [ 'thumb_suffix' ] + ext )


signature:
def read_settings ( filename = None ) :
    """Read settings from a config file in the source _ dir root ."""

implementation:
logger = logging . getLogger ( __name__ )
logger . info ( "Reading settings ..." )
settings = _DEFAULT_CONFIG . copy ( )
if filename :
    logger . debug ( "Settings file: %s" , filename )
    settings_path = os . path . dirname ( filename )
    tempdict = { }
    with open ( filename ) as f :
        code = compile ( f . read ( ) , filename , 'exec' )
        exec ( code , tempdict )
    settings . update ( ( k , v ) for k , v in tempdict . items ( ) if k not in [ '__builtins__' ] )
    # Make the paths relative to the settings file
    paths = [ 'source' , 'destination' , 'watermark' ]
    if os . path . isdir ( join ( settings_path , settings [ 'theme' ] ) ) and os . path . isdir ( join ( settings_path , settings [ 'theme' ] , 'templates' ) ) :
        paths . append ( 'theme' )
    for p in paths :
        path = settings [ p ]
        if path and not isabs ( path ) :
            settings [ p ] = abspath ( normpath ( join ( settings_path , path ) ) )
            logger . debug ( "Rewrite %s : %s -> %s" , p , path , settings [ p ] )
for key in ( 'img_size' , 'thumb_size' , 'video_size' ) :
    w , h = settings [ key ]
    if h > w :
        settings [ key ] = ( h , w )
        logger . warning ( "The %s setting should be specified with the " "largest value first." , key )
if not settings [ 'img_processor' ] :
    logger . info ( 'No Processor, images will not be resized' )
logger . debug ( 'Settings:\n%s' , pformat ( settings , width = 120 ) )
return settings


signature:
def generate_media_pages ( gallery ) :
    '''Generates and writes the media pages for all media in the gallery'''

implementation:
writer = PageWriter ( gallery . settings , index_title = gallery . title )
for album in gallery . albums . values ( ) :
    medias = album . medias
    next_medias = medias [ 1 : ] + [ None ]
    previous_medias = [ None ] + medias [ : - 1 ]
    # The media group allows us to easily get next and previous links
    media_groups = zip ( medias , next_medias , previous_medias )
    for media_group in media_groups :
        writer . write ( album , media_group )


signature:
def write ( self , album , media_group ) :
    '''Generate the media page and save it'''

implementation:
from sigal import __url__ as sigal_link
file_path = os . path . join ( album . dst_path , media_group [ 0 ] . filename )
page = self . template . render ( { 'album' : album , 'media' : media_group [ 0 ] , 'previous_media' : media_group [ - 1 ] , 'next_media' : media_group [ 1 ] , 'index_title' : self . index_title , 'settings' : self . settings , 'sigal_link' : sigal_link , 'theme' : { 'name' : os . path . basename ( self . theme ) , 'url' : url_from_path ( os . path . relpath ( self . theme_path , album . dst_path ) ) } , } )
output_file = "%s.html" % file_path
with open ( output_file , 'w' , encoding = 'utf-8' ) as f :
    f . write ( page )


signature:
def check_install ( config_data ) :
    """Here we do some * * really * * basic environment sanity checks .
 Basically we test for the more delicate and failing - prone dependencies :
 * database driver
 * Pillow image format support
 Many other errors will go undetected"""

implementation:
errors = [ ]
# PIL tests
try :
    from PIL import Image
    try :
        im = Image . open ( os . path . join ( os . path . dirname ( __file__ ) , '../share/test_image.png' ) )
        im . load ( )
    except IOError : # pragma : no cover
        errors . append ( 'Pillow is not compiled with PNG support, see "Libraries installation issues" ' 'documentation section: https://djangocms-installer.readthedocs.io/en/latest/' 'libraries.html.' )
    try :
        im = Image . open ( os . path . join ( os . path . dirname ( __file__ ) , '../share/test_image.jpg' ) )
        im . load ( )
    except IOError : # pragma : no cover
        errors . append ( 'Pillow is not compiled with JPEG support, see "Libraries installation issues" ' 'documentation section: https://djangocms-installer.readthedocs.io/en/latest/' 'libraries.html' )
except ImportError : # pragma : no cover
    errors . append ( 'Pillow is not installed check for installation errors and see "Libraries installation' ' issues" documentation section: https://djangocms-installer.readthedocs.io/en/latest/' 'libraries.html' )
# PostgreSQL test
if config_data . db_driver == 'psycopg2' and not config_data . no_db_driver : # pragma : no cover
    try :
        import psycopg2
        # NOQA
    except ImportError :
        errors . append ( 'PostgreSQL driver is not installed, but you configured a PostgreSQL database, ' 'please check your installation and see "Libraries installation issues" ' 'documentation section: https://djangocms-installer.readthedocs.io/en/latest/' 'libraries.html' )
# MySQL test
if config_data . db_driver == 'mysqlclient' and not config_data . no_db_driver : # pragma : no cover # NOQA
    try :
        import MySQLdb
        # NOQA
    except ImportError :
        errors . append ( 'MySQL driver is not installed, but you configured a MySQL database, please check ' 'your installation and see "Libraries installation issues" documentation section: ' 'https://djangocms-installer.readthedocs.io/en/latest/libraries.html' )
if errors : # pragma : no cover
    raise EnvironmentError ( '\n' . join ( errors ) )


signature:
def cleanup_directory ( config_data ) :
    """Asks user for removal of project directory and eventually removes it"""

implementation:
if os . path . exists ( config_data . project_directory ) :
    choice = False
    if config_data . noinput is False and not config_data . verbose :
        choice = query_yes_no ( 'The installation failed.\n' 'Do you want to clean up by removing {0}?\n' '\tWarning: this will delete all files in:\n' '\t\t{0}\n' 'Do you want to cleanup?' . format ( os . path . abspath ( config_data . project_directory ) ) , 'no' )
    else :
        sys . stdout . write ( 'The installation has failed.\n' )
    if config_data . skip_project_dir_check is False and ( choice or ( config_data . noinput and config_data . delete_project_dir ) ) :
        sys . stdout . write ( 'Removing everything under {0}\n' . format ( os . path . abspath ( config_data . project_directory ) ) )
        shutil . rmtree ( config_data . project_directory , True )


signature:
def validate_project ( project_name ) :
    """Check the defined project name against keywords , builtins and existing
 modules to avoid name clashing"""

implementation:
if '-' in project_name :
    return None
if keyword . iskeyword ( project_name ) :
    return None
if project_name in dir ( __builtins__ ) :
    return None
try :
    __import__ ( project_name )
    return None
except ImportError :
    return project_name