signature:
def index_of_first_consumed_mesh_in_row ( self ) :
    """The index of the first consumed mesh of this instruction in its row .
 Same as : attr : ` index _ of _ first _ produced _ mesh _ in _ row `
 but for consumed meshes ."""

implementation:
index = 0
for instruction in self . row_instructions :
    if instruction is self :
        break
    index += instruction . number_of_consumed_meshes
else :
    self . _raise_not_found_error ( )
return index


signature:
def convert_color_to_rrggbb ( color ) :
    """The color in " # RRGGBB " format .
 : return : the : attr : ` color ` in " # RRGGBB " format"""

implementation:
if not color . startswith ( "#" ) :
    rgb = webcolors . html5_parse_legacy_color ( color )
    hex_color = webcolors . html5_serialize_simple_color ( rgb )
else :
    hex_color = color
return webcolors . normalize_hex ( hex_color )


signature:
def _start ( self ) :
    """Initialize the parsing process ."""

implementation:
self . _instruction_library = self . _spec . new_default_instructions ( )
self . _as_instruction = self . _instruction_library . as_instruction
self . _id_cache = { }
self . _pattern_set = None
self . _inheritance_todos = [ ]
self . _instruction_todos = [ ]


signature:
def knitting_pattern_set ( self , values ) :
    """Parse a knitting pattern set .
 : param dict value : the specification of the knitting pattern set
 : rtype : knittingpattern . KnittingPatternSet . KnittingPatternSet
 : raises knittingpattern . KnittingPatternSet . ParsingError : if
 : paramref : ` value ` does not fulfill the : ref : ` specification
 < FileFormatSpecification > ` ."""

implementation:
self . _start ( )
pattern_collection = self . _new_pattern_collection ( )
self . _fill_pattern_collection ( pattern_collection , values )
self . _create_pattern_set ( pattern_collection , values )
return self . _pattern_set


signature:
def _finish_inheritance ( self ) :
    """Finish those who still need to inherit ."""

implementation:
while self . _inheritance_todos :
    prototype , parent_id = self . _inheritance_todos . pop ( )
    parent = self . _id_cache [ parent_id ]
    prototype . inherit_from ( parent )


signature:
def _finish_instructions ( self ) :
    """Finish those who still need to inherit ."""

implementation:
while self . _instruction_todos :
    row = self . _instruction_todos . pop ( )
    instructions = row . get ( INSTRUCTIONS , [ ] )
    row . instructions . extend ( instructions )


signature:
def _fill_pattern_collection ( self , pattern_collection , values ) :
    """Fill a pattern collection ."""

implementation:
pattern = values . get ( PATTERNS , [ ] )
for pattern_to_parse in pattern :
    parsed_pattern = self . _pattern ( pattern_to_parse )
    pattern_collection . append ( parsed_pattern )


signature:
def _row ( self , values ) :
    """Parse a row ."""

implementation:
row_id = self . _to_id ( values [ ID ] )
row = self . _spec . new_row ( row_id , values , self )
if SAME_AS in values :
    self . _delay_inheritance ( row , self . _to_id ( values [ SAME_AS ] ) )
self . _delay_instructions ( row )
self . _id_cache [ row_id ] = row
return row


signature:
def instruction_in_row ( self , row , specification ) :
    """Parse an instruction .
 : param row : the row of the instruction
 : param specification : the specification of the instruction
 : return : the instruction in the row"""

implementation:
whole_instruction_ = self . _as_instruction ( specification )
return self . _spec . new_instruction_in_row ( row , whole_instruction_ )


signature:
def _pattern ( self , base ) :
    """Parse a pattern ."""

implementation:
rows = self . _rows ( base . get ( ROWS , [ ] ) )
self . _finish_inheritance ( )
self . _finish_instructions ( )
self . _connect_rows ( base . get ( CONNECTIONS , [ ] ) )
id_ = self . _to_id ( base [ ID ] )
name = base [ NAME ]
return self . new_pattern ( id_ , name , rows )


signature:
def new_pattern ( self , id_ , name , rows = None ) :
    """Create a new knitting pattern .
 If rows is : obj : ` None ` it is replaced with the
 : meth : ` new _ row _ collection ` ."""

implementation:
if rows is None :
    rows = self . new_row_collection ( )
return self . _spec . new_pattern ( id_ , name , rows , self )


signature:
def _rows ( self , spec ) :
    """Parse a collection of rows ."""

implementation:
rows = self . new_row_collection ( )
for row in spec :
    rows . append ( self . _row ( row ) )
return rows


signature:
def _connect_rows ( self , connections ) :
    """Connect the parsed rows ."""

implementation:
for connection in connections :
    from_row_id = self . _to_id ( connection [ FROM ] [ ID ] )
    from_row = self . _id_cache [ from_row_id ]
    from_row_start_index = connection [ FROM ] . get ( START , DEFAULT_START )
    from_row_number_of_possible_meshes = from_row . number_of_produced_meshes - from_row_start_index
    to_row_id = self . _to_id ( connection [ TO ] [ ID ] )
    to_row = self . _id_cache [ to_row_id ]
    to_row_start_index = connection [ TO ] . get ( START , DEFAULT_START )
    to_row_number_of_possible_meshes = to_row . number_of_consumed_meshes - to_row_start_index
    meshes = min ( from_row_number_of_possible_meshes , to_row_number_of_possible_meshes )
    # TODO : test all kinds of connections
    number_of_meshes = connection . get ( MESHES , meshes )
    from_row_stop_index = from_row_start_index + number_of_meshes
    to_row_stop_index = to_row_start_index + number_of_meshes
    assert 0 <= from_row_start_index <= from_row_stop_index
    produced_meshes = from_row . produced_meshes [ from_row_start_index : from_row_stop_index ]
    assert 0 <= to_row_start_index <= to_row_stop_index
    consumed_meshes = to_row . consumed_meshes [ to_row_start_index : to_row_stop_index ]
    assert len ( produced_meshes ) == len ( consumed_meshes )
    mesh_pairs = zip ( produced_meshes , consumed_meshes )
    for produced_mesh , consumed_mesh in mesh_pairs :
        produced_mesh . connect_to ( consumed_mesh )


signature:
def _get_type ( self , values ) :
    """: return : the type of a knitting pattern set ."""

implementation:
if TYPE not in values :
    self . _error ( "No pattern type given but should be " "\"{}\"" . format ( KNITTING_PATTERN_TYPE ) )
type_ = values [ TYPE ]
if type_ != KNITTING_PATTERN_TYPE :
    self . _error ( "Wrong pattern type. Type is \"{}\" " "but should be \"{}\"" "" . format ( type_ , KNITTING_PATTERN_TYPE ) )
return type_


signature:
def _create_pattern_set ( self , pattern , values ) :
    """Create a new pattern set ."""

implementation:
type_ = self . _get_type ( values )
version = self . _get_version ( values )
comment = values . get ( COMMENT )
self . _pattern_set = self . _spec . new_pattern_set ( type_ , version , pattern , self , comment )


signature:
def add_row ( self , id_ ) :
    """Add a new row to the pattern .
 : param id _ : the id of the row"""

implementation:
row = self . _parser . new_row ( id_ )
self . _rows . append ( row )
return row


signature:
def write ( self , bytes_ ) :
    """Write bytes to the file ."""

implementation:
string = bytes_ . decode ( self . _encoding )
self . _file . write ( string )


signature:
def write ( self , string ) :
    """Write a string to the file ."""

implementation:
bytes_ = string . encode ( self . _encoding )
self . _file . write ( bytes_ )


signature:
def kivy_svg ( self ) :
    """An SVG object .
 : return : an SVG object
 : rtype : kivy . graphics . svg . Svg
 : raises ImportError : if the module was not found"""

implementation:
from kivy . graphics . svg import Svg
path = self . temporary_path ( ".svg" )
try :
    return Svg ( path )
finally :
    remove_file ( path )


signature:
def bounding_box ( self ) :
    """the bounding box of this SVG
 ` ` ( min _ x , min _ y , max _ x , max _ y ) ` ` .
 . . code : : python
 svg _ builder10x10 . bounding _ box = ( 0 , 0 , 10 , 10)
 assert svg _ builder10x10 . bounding _ box = = ( 0 , 0 , 10 , 10)
 ` ` viewBox ` ` , ` ` width ` ` and ` ` height ` ` are computed from this .
 If the bounding box was never set , the result is a tuple of four
 : obj : ` None ` ."""

implementation: return ( self . _min_x , self . _min_y , self . _max_x , self . _max_y )


signature:
def place ( self , x , y , svg , layer_id ) :
    """Place the : paramref : ` svg ` content at ` ` ( x , y ) ` ` position
 in the SVG , in a layer with the id : paramref : ` layer _ id ` .
 : param float x : the x position of the svg
 : param float y : the y position of the svg
 : param str svg : the SVG to place at ` ` ( x , y ) ` `
 : param str layer _ id : the id of the layer that this
 : paramref : ` svg ` should be placed inside"""

implementation:
content = xmltodict . parse ( svg )
self . place_svg_dict ( x , y , content , layer_id )


signature:
def place_svg_dict ( self , x , y , svg_dict , layer_id , group = None ) :
    """Same as : meth : ` place ` but with a dictionary as : paramref : ` svg _ dict ` .
 : param dict svg _ dict : a dictionary returned by ` xmltodict . parse ( )
 < https : / / github . com / martinblech / xmltodict > ` _ _
 : param dict group : a dictionary of values to add to the group the
 : paramref : ` svg _ dict ` will be added to or : obj : ` None ` if nothing
 should be added"""

implementation:
if group is None :
    group = { }
group_ = { "@transform" : "translate({},{})" . format ( x , y ) , "g" : list ( svg_dict . values ( ) ) }
group_ . update ( group )
layer = self . _get_layer ( layer_id )
layer [ "g" ] . append ( group_ )


signature:
def place_svg_use_coords ( self , x , y , symbol_id , layer_id , group = None ) :
    """Similar to : meth : ` place ` but with an id as : paramref : ` symbol _ id ` .
 : param str symbol _ id : an id which identifies an svg object defined in
 the defs
 : param dict group : a dictionary of values to add to the group the
 use statement will be added to or : obj : ` None ` if nothing
 should be added"""

implementation:
if group is None :
    group = { }
use = { "@x" : x , "@y" : y , "@xlink:href" : "#{}" . format ( symbol_id ) }
group_ = { "use" : use }
group_ . update ( group )
layer = self . _get_layer ( layer_id )
layer [ "g" ] . append ( group_ )


signature:
def place_svg_use ( self , symbol_id , layer_id , group = None ) :
    """Same as : meth : ` place _ svg _ use _ coords ` .
 With implicit ` x ` and ` y ` which are set to ` 0 ` in this method and then
 : meth : ` place _ svg _ use _ coords ` is called ."""

implementation: self . place_svg_use_coords ( 0 , 0 , symbol_id , layer_id , group )


signature:
def _get_layer ( self , layer_id ) :
    """: return : the layer with the : paramref : ` layer _ id ` . If the layer
 does not exist , it is created .
 : param str layer _ id : the id of the layer"""

implementation:
if layer_id not in self . _layer_id_to_layer :
    self . _svg . setdefault ( "g" , [ ] )
    layer = { "g" : [ ] , "@inkscape:label" : layer_id , "@id" : layer_id , "@inkscape:groupmode" : "layer" , "@class" : "row" }
    self . _layer_id_to_layer [ layer_id ] = layer
    self . _svg [ "g" ] . append ( layer )
return self . _layer_id_to_layer [ layer_id ]


signature:
def insert_defs ( self , defs ) :
    """Adds the defs to the SVG structure .
 : param defs : a list of SVG dictionaries , which contain the defs ,
 which should be added to the SVG structure ."""

implementation:
if self . _svg [ "defs" ] is None :
    self . _svg [ "defs" ] = { }
for def_ in defs :
    for key , value in def_ . items ( ) :
        if key . startswith ( "@" ) :
            continue
        if key not in self . _svg [ "defs" ] :
            self . _svg [ "defs" ] [ key ] = [ ]
        if not isinstance ( value , list ) :
            value = [ value ]
        self . _svg [ "defs" ] [ key ] . extend ( value )


signature:
def write_to_file ( self , file ) :
    """Writes the current SVG to the : paramref : ` file ` .
 : param file : a file - like object"""

implementation: xmltodict . unparse ( self . _structure , file , pretty = True )


signature:
def _width ( self ) :
    """For ` ` self . width ` ` ."""

implementation:
layout = self . _instruction . get ( GRID_LAYOUT )
if layout is not None :
    width = layout . get ( WIDTH )
    if width is not None :
        return width
return self . _instruction . number_of_consumed_meshes


signature:
def instructions ( self ) :
    """The instructions in a grid .
 : return : the : class : ` instructions in a grid < InstructionInGrid > ` of
 this row
 : rtype : list"""

implementation:
x = self . x
y = self . y
result = [ ]
for instruction in self . _row . instructions :
    instruction_in_grid = InstructionInGrid ( instruction , Point ( x , y ) )
    x += instruction_in_grid . width
    result . append ( instruction_in_grid )
return result


signature:
def _expand ( self , row , consumed_position , passed ) :
    """Add the arguments ` ( args , kw ) ` to ` _ walk ` to the todo list ."""

implementation: self . _todo . append ( ( row , consumed_position , passed ) )


signature:
def _step ( self , row , position , passed ) :
    """Walk through the knitting pattern by expanding an row ."""

implementation:
if row in passed or not self . _row_should_be_placed ( row , position ) :
    return
self . _place_row ( row , position )
passed = [ row ] + passed
# print ( " { } { } at \ t { } { } " . format ( " " * len ( passed ) , row , position ,
# passed ) )
for i , produced_mesh in enumerate ( row . produced_meshes ) :
    self . _expand_produced_mesh ( produced_mesh , i , position , passed )
for i , consumed_mesh in enumerate ( row . consumed_meshes ) :
    self . _expand_consumed_mesh ( consumed_mesh , i , position , passed )


signature:
def _expand_consumed_mesh ( self , mesh , mesh_index , row_position , passed ) :
    """expand the consumed meshes"""

implementation:
if not mesh . is_produced ( ) :
    return
row = mesh . producing_row
position = Point ( row_position . x + mesh . index_in_producing_row - mesh_index , row_position . y - INSTRUCTION_HEIGHT )
self . _expand ( row , position , passed )


signature:
def _expand_produced_mesh ( self , mesh , mesh_index , row_position , passed ) :
    """expand the produced meshes"""

implementation:
if not mesh . is_consumed ( ) :
    return
row = mesh . consuming_row
position = Point ( row_position . x - mesh . index_in_consuming_row + mesh_index , row_position . y + INSTRUCTION_HEIGHT )
self . _expand ( row , position , passed )


signature:
def _row_should_be_placed ( self , row , position ) :
    """: return : whether to place this instruction"""

implementation:
placed_row = self . _rows_in_grid . get ( row )
return placed_row is None or placed_row . y < position . y


signature:
def _place_row ( self , row , position ) :
    """place the instruction on a grid"""

implementation: self . _rows_in_grid [ row ] = RowInGrid ( row , position )


signature:
def _walk ( self ) :
    """Loop through all the instructions that are ` _ todo ` ."""

implementation:
while self . _todo :
    args = self . _todo . pop ( 0 )
    self . _step ( * args )


signature:
def instruction_in_grid ( self , instruction ) :
    """Returns an ` InstructionInGrid ` object for the ` instruction `"""

implementation:
row_position = self . _rows_in_grid [ instruction . row ] . xy
x = instruction . index_of_first_consumed_mesh_in_row
position = Point ( row_position . x + x , row_position . y )
return InstructionInGrid ( instruction , position )


signature:
def is_visible ( self ) :
    """: return : is this connection is visible
 : rtype : bool
 A connection is visible if it is longer that 0."""

implementation:
if self . _start . y + 1 < self . _stop . y :
    return True
return False


signature:
def walk_instructions ( self , mapping = identity ) :
    """Iterate over instructions .
 : return : an iterator over : class : ` instructions in grid
 < InstructionInGrid > `
 : param mapping : funcion to map the result
 . . code : : python
 for pos , c in layout . walk _ instructions ( lambda i : ( i . xy , i . color ) ) :
 print ( " color { } at { } " . format ( c , pos ) )"""

implementation:
instructions = chain ( * self . walk_rows ( lambda row : row . instructions ) )
return map ( mapping , instructions )


signature:
def walk_rows ( self , mapping = identity ) :
    """Iterate over rows .
 : return : an iterator over : class : ` rows < RowsInGrid > `
 : param mapping : funcion to map the result , see
 : meth : ` walk _ instructions ` for an example usage"""

implementation:
row_in_grid = self . _walk . row_in_grid
return map ( lambda row : mapping ( row_in_grid ( row ) ) , self . _rows )


signature:
def walk_connections ( self , mapping = identity ) :
    """Iterate over connections between instructions .
 : return : an iterator over : class : ` connections < Connection > ` between
 : class : ` instructions in grid < InstructionInGrid > `
 : param mapping : funcion to map the result , see
 : meth : ` walk _ instructions ` for an example usage"""

implementation:
for start in self . walk_instructions ( ) :
    for stop_instruction in start . instruction . consuming_instructions :
        if stop_instruction is None :
            continue
        stop = self . _walk . instruction_in_grid ( stop_instruction )
        connection = Connection ( start , stop )
        if connection . is_visible ( ) : # print ( " connection : " ,
        # connection . start . instruction ,
        # connection . stop . instruction )
            yield mapping ( connection )


signature:
def bounding_box ( self ) :
    """The minimum and maximum bounds of this layout .
 : return : ` ` ( min _ x , min _ y , max _ x , max _ y ) ` ` the bounding box
 of this layout
 : rtype : tuple"""

implementation:
min_x , min_y , max_x , max_y = zip ( * list ( self . walk_rows ( lambda row : row . bounding_box ) ) )
return min ( min_x ) , min ( min_y ) , max ( max_x ) , max ( max_y )


signature:
def _process_loaded_object ( self , path ) :
    """process the : paramref : ` path ` .
 : param str path : the path to load an svg from"""

implementation:
file_name = os . path . basename ( path )
name = os . path . splitext ( file_name ) [ 0 ]
with open ( path ) as file :
    string = file . read ( )
    self . _instruction_type_to_file_content [ name ] = string


signature:
def instruction_to_svg_dict ( self , instruction ) :
    """: return : an xml - dictionary with the same content as
 : meth : ` instruction _ to _ svg ` ."""

implementation:
instruction_type = instruction . type
if instruction_type in self . _instruction_type_to_file_content :
    svg = self . _instruction_type_to_file_content [ instruction_type ]
    return self . _set_fills_in_color_layer ( svg , instruction . hex_color )
return self . default_instruction_to_svg_dict ( instruction )


signature:
def _set_fills_in_color_layer ( self , svg_string , color ) :
    """replaces fill colors in ` ` < g inkscape : label = " color "
 inkscape : groupmode = " layer " > ` ` with : paramref : ` color `
 : param color : a color fill the objects in the layer with"""

implementation:
structure = xmltodict . parse ( svg_string )
if color is None :
    return structure
layers = structure [ "svg" ] [ "g" ]
if not isinstance ( layers , list ) :
    layers = [ layers ]
for layer in layers :
    if not isinstance ( layer , dict ) :
        continue
    if layer . get ( "@inkscape:label" ) == "color" and layer . get ( "@inkscape:groupmode" ) == "layer" :
        for key , elements in layer . items ( ) :
            if key . startswith ( "@" ) or key . startswith ( "#" ) :
                continue
            if not isinstance ( elements , list ) :
                elements = [ elements ]
            for element in elements :
                style = element . get ( "@style" , None )
                if style :
                    style = style . split ( ";" )
                    processed_style = [ ]
                    for style_element in style :
                        if style_element . startswith ( "fill:" ) :
                            style_element = "fill:" + color
                        processed_style . append ( style_element )
                    style = ";" . join ( processed_style )
                    element [ "@style" ] = style
return structure


signature:
def default_instruction_to_svg ( self , instruction ) :
    """As : meth : ` instruction _ to _ svg ` but it only takes the ` ` default . svg ` `
 file into account .
 In case no file is found for an instruction in
 : meth : ` instruction _ to _ svg ` ,
 this method is used to determine the default svg for it .
 The content is created by replacing the text ` ` { instruction . type } ` ` in
 the whole svg file named ` ` default . svg ` ` .
 If no file ` ` default . svg ` ` was loaded , an empty string is returned ."""

implementation:
svg_dict = self . default_instruction_to_svg_dict ( instruction )
return xmltodict . unparse ( svg_dict )


signature:
def default_instruction_to_svg_dict ( self , instruction ) :
    """Returns an xml - dictionary with the same content as
 : meth : ` default _ instruction _ to _ svg `
 If no file ` ` default . svg ` ` was loaded , an empty svg - dict is returned ."""

implementation:
instruction_type = instruction . type
default_type = "default"
rep_str = "{instruction.type}"
if default_type not in self . _instruction_type_to_file_content :
    return { "svg" : "" }
default_svg = self . _instruction_type_to_file_content [ default_type ]
default_svg = default_svg . replace ( rep_str , instruction_type )
colored_svg = self . _set_fills_in_color_layer ( default_svg , instruction . hex_color )
return colored_svg


signature:
def _dump_to_file ( self , file ) :
    """dump to the file"""

implementation: xmltodict . unparse ( self . object ( ) , file , pretty = True )


signature:
def add_instruction ( self , specification ) :
    """Add an instruction specification
 : param specification : a specification with a key
 : data : ` knittingpattern . Instruction . TYPE `
 . . seealso : : : meth : ` as _ instruction `"""

implementation:
instruction = self . as_instruction ( specification )
self . _type_to_instruction [ instruction . type ] = instruction


signature:
def as_instruction ( self , specification ) :
    """Convert the specification into an instruction
 : param specification : a specification with a key
 : data : ` knittingpattern . Instruction . TYPE `
 The instruction is not added .
 . . seealso : : : meth : ` add _ instruction `"""

implementation:
instruction = self . _instruction_class ( specification )
type_ = instruction . type
if type_ in self . _type_to_instruction :
    instruction . inherit_from ( self . _type_to_instruction [ type_ ] )
return instruction


signature:
def eigh ( self ) :
    """Eigen decomposition of K .
 Returns
 S : ndarray
 The eigenvalues in ascending order , each repeated according to its
 multiplicity .
 U : ndarray
 Normalized eigenvectors ."""

implementation:
from numpy . linalg import svd
if self . _cache [ "eig" ] is not None :
    return self . _cache [ "eig" ]
U , S = svd ( self . L ) [ : 2 ]
S *= S
S += self . _epsilon
self . _cache [ "eig" ] = S , U
return self . _cache [ "eig" ]


signature:
def L ( self ) :
    """Lower - triangular matrix L such that K = LLT + εI .
 Returns
 L : ( d , d ) ndarray
 Lower - triangular matrix ."""

implementation:
m = len ( self . _tril1 [ 0 ] )
self . _L [ self . _tril1 ] = self . _Lu . value [ : m ]
self . _L [ self . _diag ] = exp ( self . _Lu . value [ m : ] )
return self . _L


signature:
def logdet ( self ) :
    """Log of | K | .
 Returns
 float
 Log - determinant of K ."""

implementation:
from numpy . linalg import slogdet
K = self . value ( )
sign , logdet = slogdet ( K )
if sign != 1.0 :
    msg = "The estimated determinant of K is not positive: "
    msg += f" ({sign}, {logdet})."
    raise RuntimeError ( msg )
return logdet


signature:
def value ( self ) :
    """Covariance matrix .
 Returns
 K : ndarray
 Matrix K = LLT + εI , for a very small positive number ε ."""

implementation:
K = dot ( self . L , self . L . T )
return K + self . _epsilon * eye ( K . shape [ 0 ] )


signature:
def gradient ( self ) :
    """Derivative of the covariance matrix over the parameters of L .
 Returns
 Lu : ndarray
 Derivative of K over the lower triangular part of L ."""

implementation:
L = self . L
self . _grad_Lu [ : ] = 0
for i in range ( len ( self . _tril1 [ 0 ] ) ) :
    row = self . _tril1 [ 0 ] [ i ]
    col = self . _tril1 [ 1 ] [ i ]
    self . _grad_Lu [ row , : , i ] = L [ : , col ]
    self . _grad_Lu [ : , row , i ] += L [ : , col ]
m = len ( self . _tril1 [ 0 ] )
for i in range ( len ( self . _diag [ 0 ] ) ) :
    row = self . _diag [ 0 ] [ i ]
    col = self . _diag [ 1 ] [ i ]
    self . _grad_Lu [ row , : , m + i ] = L [ row , col ] * L [ : , col ]
    self . _grad_Lu [ : , row , m + i ] += L [ row , col ] * L [ : , col ]
return { "Lu" : self . _grad_Lu }


signature:
def Ge ( self ) :
    """Result of US from the SVD decomposition G = USVT ."""

implementation:
from scipy . linalg import svd
from numpy_sugar . linalg import ddot
U , S , _ = svd ( self . _G , full_matrices = False , check_finite = False )
if U . shape [ 1 ] < self . _G . shape [ 1 ] :
    return ddot ( U , S )
return self . _G


signature:
def listen ( self , func ) :
    """Listen to parameters change .
 Parameters
 func : callable
 Function to be called when a parameter changes ."""

implementation:
self . _C0 . listen ( func )
self . _C1 . listen ( func )


signature:
def _LhD ( self ) :
    """Implements Lh and D .
 Returns
 Lh : ndarray
 UhT S1 − 1 ⁄ 2 U1T .
 D : ndarray
 ( Sh ⊗ Sx + Ihx ) − 1."""

implementation:
from numpy_sugar . linalg import ddot
self . _init_svd ( )
if self . _cache [ "LhD" ] is not None :
    return self . _cache [ "LhD" ]
S1 , U1 = self . C1 . eigh ( )
U1S1 = ddot ( U1 , 1 / sqrt ( S1 ) )
Sh , Uh = eigh ( U1S1 . T @ self . C0 . value ( ) @ U1S1 )
self . _cache [ "LhD" ] = { "Lh" : ( U1S1 @ Uh ) . T , "D" : 1 / ( kron ( Sh , self . _Sx ) + 1 ) , "De" : 1 / ( kron ( Sh , self . _Sxe ) + 1 ) , }
return self . _cache [ "LhD" ]


signature:
def value ( self ) :
    """Covariance matrix K = C0 ⊗ GGT + C1 ⊗ I .
 Returns
 K : ndarray
 C0 ⊗ GGT + C1 ⊗ I ."""

implementation:
C0 = self . _C0 . value ( )
C1 = self . _C1 . value ( )
return kron ( C0 , self . _GG ) + kron ( C1 , self . _I )


signature:
def gradient ( self ) :
    """Gradient of K .
 Returns
 C0 : ndarray
 Derivative of C0 over its parameters .
 C1 : ndarray
 Derivative of C1 over its parameters ."""

implementation:
self . _init_svd ( )
C0 = self . _C0 . gradient ( ) [ "Lu" ] . T
C1 = self . _C1 . gradient ( ) [ "Lu" ] . T
grad = { "C0.Lu" : kron ( C0 , self . _X ) . T , "C1.Lu" : kron ( C1 , self . _I ) . T }
return grad


signature:
def gradient_dot ( self , v ) :
    """Implements ∂ K ⋅ v .
 Parameters
 v : array _ like
 Vector from ∂ K ⋅ v .
 Returns
 C0 . Lu : ndarray
 ∂ K ⋅ v , where the gradient is taken over the C0 parameters .
 C1 . Lu : ndarray
 ∂ K ⋅ v , where the gradient is taken over the C1 parameters ."""

implementation:
self . _init_svd ( )
V = unvec ( v , ( self . G . shape [ 0 ] , - 1 ) + v . shape [ 1 : ] )
r = { }
C = self . _C0 . gradient ( ) [ "Lu" ]
r [ "C0.Lu" ] = tensordot ( V . T @ self . G @ self . G . T , C , axes = ( [ - 2 ] , [ 0 ] ) )
r [ "C0.Lu" ] = r [ "C0.Lu" ] . reshape ( V . shape [ 2 : ] + ( - 1 , ) + ( C . shape [ - 1 ] , ) , order = "F" )
C = self . _C1 . gradient ( ) [ "Lu" ]
r [ "C1.Lu" ] = tensordot ( V . T , C , axes = ( [ - 2 ] , [ 0 ] ) )
r [ "C1.Lu" ] = r [ "C1.Lu" ] . reshape ( V . shape [ 2 : ] + ( - 1 , ) + ( C . shape [ - 1 ] , ) , order = "F" )
return r


signature:
def solve ( self , v ) :
    """Implements the product K − 1 ⋅ v .
 Parameters
 v : array _ like
 Array to be multiplied .
 Returns
 x : ndarray
 Solution x to the equation K ⋅ x = y ."""

implementation:
from numpy_sugar . linalg import ddot
self . _init_svd ( )
L = kron ( self . Lh , self . Lx )
return L . T @ ddot ( self . D , L @ v , left = True )


signature:
def logdet ( self ) :
    """Implements log | K | = - log | D | + n ⋅ log | C1 | .
 Returns
 logdet : float
 Log - determinant of K ."""

implementation:
self . _init_svd ( )
return - log ( self . _De ) . sum ( ) + self . G . shape [ 0 ] * self . C1 . logdet ( )


signature:
def logdet_gradient ( self ) :
    """Implements ∂ log | K | = Tr [ K − 1 ∂ K ] .
 It can be shown that : :
 ∂ log | K | = diag ( D ) Tdiag ( L ( ∂ K ) LT ) = diag ( D ) T ( diag ( Lh ∂ C0LhT ) ⊗ diag ( LxGGTLxT ) ) ,
 when the derivative is over the parameters of C0 . Similarly ,
 ∂ log | K | = diag ( D ) Tdiag ( L ( ∂ K ) LT ) = diag ( D ) T ( diag ( Lh ∂ C1LhT ) ⊗ diag ( I ) ) ,
 over the parameters of C1.
 Returns
 C0 : ndarray
 Derivative of C0 over its parameters .
 C1 : ndarray
 Derivative of C1 over its parameters ."""

implementation:
from numpy_sugar . linalg import dotd
self . _init_svd ( )
dC0 = self . _C0 . gradient ( ) [ "Lu" ]
grad_C0 = zeros_like ( self . _C0 . Lu )
for i in range ( self . _C0 . Lu . shape [ 0 ] ) :
    t = kron ( dotd ( self . Lh , dC0 [ ... , i ] @ self . Lh . T ) , self . _diag_LxGGLxe )
    grad_C0 [ i ] = ( self . _De * t ) . sum ( )
dC1 = self . _C1 . gradient ( ) [ "Lu" ]
grad_C1 = zeros_like ( self . _C1 . Lu )
p = self . _Sxe . shape [ 0 ]
np = self . _G . shape [ 0 ] - p
for i in range ( self . _C1 . Lu . shape [ 0 ] ) :
    t = ( dotd ( self . Lh , dC1 [ ... , i ] @ self . Lh . T ) * np ) . sum ( )
    t1 = kron ( dotd ( self . Lh , dC1 [ ... , i ] @ self . Lh . T ) , eye ( p ) )
    t += ( self . _De * t1 ) . sum ( )
    grad_C1 [ i ] = t
return { "C0.Lu" : grad_C0 , "C1.Lu" : grad_C1 }


signature:
def LdKL_dot ( self , v , v1 = None ) :
    """Implements L ( ∂ K ) LTv .
 The array v can have one or two dimensions and the first dimension has to have
 size n ⋅ p .
 Let vec ( V ) = v . We have
 L ( ∂ K ) LT ⋅ v = ( ( Lh ∂ C0LhT ) ⊗ ( LxGGTLxT ) ) vec ( V ) = vec ( LxGGTLxTVLh ∂ C0LhT ) ,
 when the derivative is over the parameters of C0 . Similarly ,
 L ( ∂ K ) LTv = ( ( Lh ∂ C1LhT ) ⊗ ( LxLxT ) ) vec ( V ) = vec ( LxLxTVLh ∂ C1LhT ) ,
 over the parameters of C1."""

implementation:
self . _init_svd ( )
def dot ( a , b ) :
    r = tensordot ( a , b , axes = ( [ 1 ] , [ 0 ] ) )
    if a . ndim > b . ndim :
        return r . transpose ( [ 0 , 2 , 1 ] )
    return r
Lh = self . Lh
V = unvec ( v , ( self . Lx . shape [ 0 ] , - 1 ) + v . shape [ 1 : ] )
LdKL_dot = { "C0.Lu" : empty ( ( v . shape [ 0 ] , ) + v . shape [ 1 : ] + ( self . _C0 . Lu . shape [ 0 ] , ) ) , "C1.Lu" : empty ( ( v . shape [ 0 ] , ) + v . shape [ 1 : ] + ( self . _C1 . Lu . shape [ 0 ] , ) ) , }
dC0 = self . _C0 . gradient ( ) [ "Lu" ]
for i in range ( self . _C0 . Lu . shape [ 0 ] ) :
    t = dot ( self . _LxG , dot ( self . _LxG . T , dot ( V , Lh @ dC0 [ ... , i ] @ Lh . T ) ) )
    LdKL_dot [ "C0.Lu" ] [ ... , i ] = t . reshape ( ( - 1 , ) + t . shape [ 2 : ] , order = "F" )
dC1 = self . _C1 . gradient ( ) [ "Lu" ]
for i in range ( self . _C1 . Lu . shape [ 0 ] ) :
    t = dot ( V , Lh @ dC1 [ ... , i ] @ Lh . T )
    LdKL_dot [ "C1.Lu" ] [ ... , i ] = t . reshape ( ( - 1 , ) + t . shape [ 2 : ] , order = "F" )
return LdKL_dot


signature:
def rsolve ( A , y ) :
    """Robust solve Ax = y ."""

implementation:
from numpy_sugar . linalg import rsolve as _rsolve
try :
    beta = _rsolve ( A , y )
except LinAlgError :
    msg = "Could not converge to solve Ax=y."
    msg += " Setting x to zero."
    warnings . warn ( msg , RuntimeWarning )
    beta = zeros ( A . shape [ 0 ] )
return beta


signature:
def multivariate_normal ( random , mean , cov ) :
    """Draw random samples from a multivariate normal distribution .
 Parameters
 random : np . random . RandomState instance
 Random state .
 mean : array _ like
 Mean of the n - dimensional distribution .
 cov : array _ like
 Covariance matrix of the distribution . It must be symmetric and
 positive - definite for proper sampling .
 Returns
 out : ndarray
 The drawn sample ."""

implementation:
from numpy . linalg import cholesky
L = cholesky ( cov )
return L @ random . randn ( L . shape [ 0 ] ) + mean


signature:
def gradient ( self ) :
    """Sum of covariance function derivatives .
 Returns
 dict
 ∂ K0 + ∂ K1 + ⋯"""

implementation:
grad = { }
for i , f in enumerate ( self . _covariances ) :
    for varname , g in f . gradient ( ) . items ( ) :
        grad [ f"{self._name}[{i}].{varname}" ] = g
return grad


signature:
def value ( self ) :
    """Covariance matrix .
 Returns
 K : ndarray
 s ⋅ XXT ."""

implementation:
X = self . X
return self . scale * ( X @ X . T )


signature:
def B ( self ) :
    """Effect - sizes parameter , B ."""

implementation: return unvec ( self . _vecB . value , ( self . X . shape [ 1 ] , self . A . shape [ 0 ] ) )


signature:
def bernoulli_sample ( offset , G , heritability = 0.5 , causal_variants = None , causal_variance = 0 , random_state = None , ) :
    r"""Bernoulli likelihood sampling .
 Sample according to
 . . math : :
 \ mathbf y \ sim \ prod _ { i = 1 } ^ n
 \ text { Bernoulli } ( \ mu _ i = \ text { logit } ( z _ i ) )
 \ mathcal N ( ~ o \ mathbf 1 + \ mathbf a ^ \ intercal \ boldsymbol \ alpha ;
 ~ ( h ^ 2 - v _ c ) \ mathrm G ^ \ intercal \ mathrm G +
 (1 - h ^ 2 - v _ c ) \ mathrm I ~ )
 using the canonical Logit link function to define the conditional Bernoulli
 mean : math : ` \ mu _ i ` .
 The causal : math : ` \ mathbf a ` covariates and the corresponding effect - sizes
 are randomly draw according to the following idea . The ` ` causal _ variants ` ` ,
 if given , are first mean - zero and std - one normalized and then having
 its elements divided by the squared - root the the number of variances : :
 causal _ variants = _ stdnorm ( causal _ variants , axis = 0)
 causal _ variants / = sqrt ( causal _ variants . shape [ 1 ] )
 The causal effect - sizes : math : ` \ boldsymbol \ alpha ` are draw from
 : math : ` \ { - 1 , + 1 \ } ` and subsequently normalized for mean - zero and std - one " "
 Parameters
 random _ state : random _ state
 Set the initial random state .
 Example
 . . doctest : :
 > > > from glimix _ core . random import bernoulli _ sample
 > > > from numpy . random import RandomState
 > > > offset = 5
 > > > G = [ [ 1 , - 1 ] , [ 2 , 1 ] ]
 > > > bernoulli _ sample ( offset , G , random _ state = RandomState ( 0 ) )
 array ( [ 1 . , 1 . ] )"""

implementation:
link = LogitLink ( )
mean , cov = _mean_cov ( offset , G , heritability , causal_variants , causal_variance , random_state )
lik = BernoulliProdLik ( link )
sampler = GGPSampler ( lik , mean , cov )
return sampler . sample ( random_state )


signature:
def poisson_sample ( offset , G , heritability = 0.5 , causal_variants = None , causal_variance = 0 , random_state = None , ) :
    """Poisson likelihood sampling .
 Parameters
 random _ state : random _ state
 Set the initial random state .
 Example
 . . doctest : :
 > > > from glimix _ core . random import poisson _ sample
 > > > from numpy . random import RandomState
 > > > offset = - 0.5
 > > > G = [ [ 0.5 , - 1 ] , [ 2 , 1 ] ]
 > > > poisson _ sample ( offset , G , random _ state = RandomState ( 0 ) )
 array ( [ 0 , 6 ] )"""

implementation:
mean , cov = _mean_cov ( offset , G , heritability , causal_variants , causal_variance , random_state )
link = LogLink ( )
lik = PoissonProdLik ( link )
sampler = GGPSampler ( lik , mean , cov )
return sampler . sample ( random_state )


signature:
def L ( self ) :
    r"""Cholesky decomposition of : math : ` \ mathrm B ` .
 . . math : :
 \ mathrm B = \ mathrm Q ^ { \ intercal } \ tilde { \ mathrm { T } } \ mathrm Q
 + \ mathrm { S } ^ { - 1}"""

implementation:
from numpy_sugar . linalg import ddot , sum2diag
if self . _L_cache is not None :
    return self . _L_cache
s = self . _cov [ "scale" ]
d = self . _cov [ "delta" ]
Q = self . _cov [ "QS" ] [ 0 ] [ 0 ]
S = self . _cov [ "QS" ] [ 1 ]
ddot ( self . A * self . _site . tau , Q , left = True , out = self . _NxR )
B = dot ( Q . T , self . _NxR , out = self . _RxR )
B *= 1 - d
sum2diag ( B , 1.0 / S / s , out = B )
self . _L_cache = _cho_factor ( B )
return self . _L_cache


signature:
def fit ( self , verbose = True , factr = 1e5 , pgtol = 1e-7 ) :
    r"""Maximise the marginal likelihood .
 Parameters
 verbose : bool
 ` ` True ` ` for progress output ; ` ` False ` ` otherwise .
 Defaults to ` ` True ` ` .
 factr : float , optional
 The iteration stops when
 ` ` ( f ^ k - f ^ { k + 1 } ) / max { | f ^ k | , | f ^ { k + 1 } | , 1 } < = factr * eps ` ` , where ` ` eps ` ` is
 the machine precision .
 pgtol : float , optional
 The iteration will stop when ` ` max { | proj g _ i | i = 1 , . . . , n } < = pgtol ` `
 where ` ` pg _ i ` ` is the i - th component of the projected gradient .
 Notes
 Please , refer to : func : ` scipy . optimize . fmin _ l _ bfgs _ b ` for further information
 about ` ` factr ` ` and ` ` pgtol ` ` ."""

implementation: self . _maximize ( verbose = verbose , factr = factr , pgtol = pgtol )


signature:
def covariance ( self ) :
    r"""Covariance of the prior .
 Returns
 : class : ` numpy . ndarray `
 : math : ` v _ 0 \ mathrm K + v _ 1 \ mathrm I ` ."""

implementation:
from numpy_sugar . linalg import ddot , sum2diag
Q0 = self . _QS [ 0 ] [ 0 ]
S0 = self . _QS [ 1 ]
return sum2diag ( dot ( ddot ( Q0 , self . v0 * S0 ) , Q0 . T ) , self . v1 )


implementation:
self . _verbose = verbose
self . _maximize ( verbose = verbose , factr = factr , pgtol = pgtol )
self . _verbose = False


signature:
def posteriori_mean ( self ) :
    r"""Mean of the estimated posteriori .
 This is also the maximum a posteriori estimation of the latent variable ."""

implementation:
from numpy_sugar . linalg import rsolve
Sigma = self . posteriori_covariance ( )
eta = self . _ep . _posterior . eta
return dot ( Sigma , eta + rsolve ( GLMM . covariance ( self ) , self . mean ( ) ) )


signature:
def posteriori_covariance ( self ) :
    r"""Covariance of the estimated posteriori ."""

implementation:
K = GLMM . covariance ( self )
tau = self . _ep . _posterior . tau
return pinv ( pinv ( K ) + diag ( 1 / tau ) )


signature:
def _bstar_1effect ( beta , alpha , yTBy , yTBX , yTBM , XTBX , XTBM , MTBM ) :
    """Same as : func : ` _ bstar _ set ` but for single - effect ."""

implementation:
from numpy_sugar import epsilon
from numpy_sugar . linalg import dotd
from numpy import sum
r = full ( MTBM [ 0 ] . shape [ 0 ] , yTBy )
r -= 2 * add . reduce ( [ dot ( i , beta ) for i in yTBX ] )
r -= 2 * add . reduce ( [ i * alpha for i in yTBM ] )
r += add . reduce ( [ dotd ( beta . T , dot ( i , beta ) ) for i in XTBX ] )
r += add . reduce ( [ dotd ( beta . T , i * alpha ) for i in XTBM ] )
r += add . reduce ( [ sum ( alpha * i * beta , axis = 0 ) for i in XTBM ] )
r += add . reduce ( [ alpha * i . ravel ( ) * alpha for i in MTBM ] )
return clip ( r , epsilon . tiny , inf )


signature:
def _bstar_set ( beta , alpha , yTBy , yTBX , yTBM , XTBX , XTBM , MTBM ) :
    """Compute - 2yTBEjbj + ( bjEj ) TBEjbj .
 For bj = [ βjT αjT ] T ."""

implementation:
from numpy_sugar import epsilon
r = yTBy
r -= 2 * add . reduce ( [ i @ beta for i in yTBX ] )
r -= 2 * add . reduce ( [ i @ alpha for i in yTBM ] )
r += add . reduce ( [ beta . T @ i @ beta for i in XTBX ] )
r += 2 * add . reduce ( [ beta . T @ i @ alpha for i in XTBM ] )
r += add . reduce ( [ alpha . T @ i @ alpha for i in MTBM ] )
return clip ( r , epsilon . tiny , inf )


signature:
def null_lml ( self ) :
    """Log of the marginal likelihood for the null hypothesis .
 It is implemented as : :
 2 · log ( p ( Y ) ) = - n · log ( 2πs ) - log | D | - n ,
 Returns
 lml : float
 Log of the marginal likelihood ."""

implementation:
n = self . _nsamples
scale = self . null_scale
return ( self . _static_lml ( ) - n * log ( scale ) ) / 2


signature:
def null_beta ( self ) :
    """Optimal β according to the marginal likelihood .
 It is compute by solving the equation : :
 ( XTBX ) β = XTBy .
 Returns
 beta : ndarray
 Optimal β ."""

implementation:
ETBE = self . _ETBE
yTBX = self . _yTBX
A = sum ( i . XTBX for i in ETBE )
b = sum ( yTBX )
return rsolve ( A , b )


signature:
def null_beta_covariance ( self ) :
    """Covariance of the optimal β according to the marginal likelihood .
 Returns
 beta _ covariance : ndarray
 ( XT ( s ( K + vI ) ) − 1X ) − 1."""

implementation:
A = sum ( i @ j . T for ( i , j ) in zip ( self . _XTQDi , self . _XTQ ) )
return self . null_scale * pinv ( A )


signature:
def null_scale ( self ) :
    """Optimal s according to the marginal likelihood .
 The optimal s is given by : :
 s = n − 1yTB ( y - Xβ ) ,
 where β is optimal .
 Returns
 scale : float
 Optimal scale ."""

implementation:
n = self . _nsamples
beta = self . null_beta
sqrdot = self . _yTBy - dot ( sum ( self . _yTBX ) , beta )
return sqrdot / n


signature:
def fast_scan ( self , M , verbose = True ) :
    """LMLs , fixed - effect sizes , and scales for single - marker scan .
 Parameters
 M : array _ like
 Matrix of fixed - effects across columns .
 verbose : bool , optional
 ` ` True ` ` for progress information ; ` ` False ` ` otherwise .
 Defaults to ` ` True ` ` .
 Returns
 lmls : ndarray
 Log of the marginal likelihoods .
 effsizes0 : ndarray
 Covariate fixed - effect sizes .
 effsizes1 : ndarray
 Candidate set fixed - effect sizes .
 scales : ndarray
 Scales ."""

implementation:
from tqdm import tqdm
if M . ndim != 2 :
    raise ValueError ( "`M` array must be bidimensional." )
p = M . shape [ 1 ]
lmls = empty ( p )
effsizes0 = empty ( ( p , self . _XTQ [ 0 ] . shape [ 0 ] ) )
effsizes0_se = empty ( ( p , self . _XTQ [ 0 ] . shape [ 0 ] ) )
effsizes1 = empty ( p )
effsizes1_se = empty ( p )
scales = empty ( p )
if verbose :
    nchunks = min ( p , 30 )
else :
    nchunks = min ( p , 1 )
chunk_size = ( p + nchunks - 1 ) // nchunks
for i in tqdm ( range ( nchunks ) , desc = "Scanning" , disable = not verbose ) :
    start = i * chunk_size
    stop = min ( start + chunk_size , M . shape [ 1 ] )
    r = self . _fast_scan_chunk ( M [ : , start : stop ] )
    lmls [ start : stop ] = r [ "lml" ]
    effsizes0 [ start : stop , : ] = r [ "effsizes0" ]
    effsizes0_se [ start : stop , : ] = r [ "effsizes0_se" ]
    effsizes1 [ start : stop ] = r [ "effsizes1" ]
    effsizes1_se [ start : stop ] = r [ "effsizes1_se" ]
    scales [ start : stop ] = r [ "scale" ]
return { "lml" : lmls , "effsizes0" : effsizes0 , "effsizes0_se" : effsizes0_se , "effsizes1" : effsizes1 , "effsizes1_se" : effsizes1_se , "scale" : scales , }


signature:
def scan ( self , M ) :
    """LML , fixed - effect sizes , and scale of the candidate set .
 Parameters
 M : array _ like
 Fixed - effects set .
 Returns
 lml : float
 Log of the marginal likelihood .
 effsizes0 : ndarray
 Covariates fixed - effect sizes .
 effsizes0 _ se : ndarray
 Covariates fixed - effect size standard errors .
 effsizes1 : ndarray
 Candidate set fixed - effect sizes .
 effsizes1 _ se : ndarray
 Candidate fixed - effect size standard errors .
 scale : ndarray
 Optimal scale ."""

implementation:
from numpy_sugar . linalg import ddot
from numpy_sugar import is_all_finite
M = asarray ( M , float )
if M . shape [ 1 ] == 0 :
    return { "lml" : self . null_lml ( ) , "effsizes0" : self . null_beta , "effsizes0_se" : self . null_beta_se , "effsizes1" : empty ( ( 0 ) ) , "effsizes1_se" : empty ( ( 0 ) ) , "scale" : self . null_scale , }
if not is_all_finite ( M ) :
    raise ValueError ( "M parameter has non-finite elements." )
MTQ = [ dot ( M . T , Q ) for Q in self . _QS [ 0 ] if Q . size > 0 ]
yTBM = [ dot ( i , j . T ) for ( i , j ) in zip ( self . _yTQDi , MTQ ) ]
XTBM = [ dot ( i , j . T ) for ( i , j ) in zip ( self . _XTQDi , MTQ ) ]
D = self . _D
MTBM = [ ddot ( i , 1 / j ) @ i . T for i , j in zip ( MTQ , D ) if j . min ( ) > 0 ]
return self . _multicovariate_set ( yTBM , XTBM , MTBM )


signature:
def null_lml ( self ) :
    """Log of the marginal likelihood for the null hypothesis .
 It is implemented as : :
 2 · log ( p ( Y ) ) = - n · p · log ( 2πs ) - log | K | - n · p ,
 for which s and Β are optimal .
 Returns
 lml : float
 Log of the marginal likelihood ."""

implementation:
np = self . _nsamples * self . _ntraits
scale = self . null_scale
return self . _static_lml ( ) / 2 - np * safe_log ( scale ) / 2 - np / 2


signature:
def null_scale ( self ) :
    """Optimal s according to the marginal likelihood .
 The optimal s is given by
 s = ( n · p ) − 1yTK − 1 ( y - m ) ,
 where m = ( A ⊗ X ) vec ( Β ) and Β is optimal .
 Returns
 scale : float
 Optimal scale ."""

implementation:
np = self . _nsamples * self . _ntraits
b = vec ( self . null_beta )
mKiy = b . T @ self . _MKiy
sqrtdot = self . _yKiy - mKiy
scale = sqrtdot / np
return scale


signature:
def scan ( self , A1 , X1 ) :
    """LML , fixed - effect sizes , and scale of the candidate set .
 Parameters
 A1 : ( p , e ) array _ like
 Trait - by - environments design matrix .
 X1 : ( n , m ) array _ like
 Variants set matrix .
 Returns
 lml : float
 Log of the marginal likelihood for the set .
 effsizes0 : ( c , p ) ndarray
 Fixed - effect sizes for the covariates .
 effsizes0 _ se : ( c , p ) ndarray
 Fixed - effect size standard errors for the covariates .
 effsizes1 : ( m , e ) ndarray
 Fixed - effect sizes for the candidates .
 effsizes1 _ se : ( m , e ) ndarray
 Fixed - effect size standard errors for the candidates .
 scale : float
 Optimal scale ."""

implementation:
from numpy import empty
from numpy . linalg import multi_dot
from numpy_sugar import epsilon , is_all_finite
from scipy . linalg import cho_solve
A1 = asarray ( A1 , float )
X1 = asarray ( X1 , float )
if not is_all_finite ( A1 ) :
    raise ValueError ( "A1 parameter has non-finite elements." )
if not is_all_finite ( X1 ) :
    raise ValueError ( "X1 parameter has non-finite elements." )
if A1 . shape [ 1 ] == 0 :
    beta_se = sqrt ( self . null_beta_covariance . diagonal ( ) )
    return { "lml" : self . null_lml ( ) , "effsizes0" : unvec ( self . null_beta , ( self . _ncovariates , - 1 ) ) , "effsizes0_se" : unvec ( beta_se , ( self . _ncovariates , - 1 ) ) , "effsizes1" : empty ( ( 0 , ) ) , "effsizes1_se" : empty ( ( 0 , ) ) , "scale" : self . null_scale , }
X1X1 = X1 . T @ X1
XX1 = self . _X . T @ X1
AWA1 = self . _WA . T @ A1
A1W = A1 . T @ self . _W
GX1 = self . _G . T @ X1
MRiM1 = kron ( AWA1 , XX1 )
M1RiM1 = kron ( A1W @ A1 , X1X1 )
M1Riy = vec ( multi_dot ( [ X1 . T , self . _Y , A1W . T ] ) )
XRiM1 = kron ( self . _WL0 . T @ A1 , GX1 )
ZiXRiM1 = cho_solve ( self . _Lz , XRiM1 )
MRiXZiXRiM1 = self . _XRiM . T @ ZiXRiM1
M1RiXZiXRiM1 = XRiM1 . T @ ZiXRiM1
M1RiXZiXRiy = XRiM1 . T @ self . _ZiXRiy
T0 = [ [ self . _MRiM , MRiM1 ] , [ MRiM1 . T , M1RiM1 ] ]
T1 = [ [ self . _MRiXZiXRiM , MRiXZiXRiM1 ] , [ MRiXZiXRiM1 . T , M1RiXZiXRiM1 ] ]
T2 = [ self . _MRiy , M1Riy ]
T3 = [ self . _MRiXZiXRiy , M1RiXZiXRiy ]
MKiM = block ( T0 ) - block ( T1 )
MKiy = block ( T2 ) - block ( T3 )
beta = rsolve ( MKiM , MKiy )
mKiy = beta . T @ MKiy
cp = self . _ntraits * self . _ncovariates
effsizes0 = unvec ( beta [ : cp ] , ( self . _ncovariates , self . _ntraits ) )
effsizes1 = unvec ( beta [ cp : ] , ( X1 . shape [ 1 ] , A1 . shape [ 1 ] ) )
np = self . _nsamples * self . _ntraits
sqrtdot = self . _yKiy - mKiy
scale = clip ( sqrtdot / np , epsilon . tiny , inf )
lml = self . _static_lml ( ) / 2 - np * safe_log ( scale ) / 2 - np / 2
effsizes_se = sqrt ( clip ( scale * pinv ( MKiM ) . diagonal ( ) , epsilon . tiny , inf ) )
effsizes0_se = unvec ( effsizes_se [ : cp ] , ( self . _ncovariates , self . _ntraits ) )
effsizes1_se = unvec ( effsizes_se [ cp : ] , ( X1 . shape [ 1 ] , A1 . shape [ 1 ] ) )
return { "lml" : lml , "effsizes0" : effsizes0 , "effsizes1" : effsizes1 , "scale" : scale , "effsizes0_se" : effsizes0_se , "effsizes1_se" : effsizes1_se , }


signature:
def sample ( self , random_state = None ) :
    r"""Sample from the specified distribution .
 Parameters
 random _ state : random _ state
 Set the initial random state .
 Returns
 numpy . ndarray
 Sample ."""

implementation:
from numpy_sugar import epsilon
from numpy_sugar . linalg import sum2diag
from numpy_sugar . random import multivariate_normal
if random_state is None :
    random_state = RandomState ( )
m = self . _mean . value ( )
K = self . _cov . value ( ) . copy ( )
sum2diag ( K , + epsilon . small , out = K )
return self . _lik . sample ( multivariate_normal ( m , K , random_state ) , random_state )


signature:
def economic_qs_zeros ( n ) :
    """Eigen decomposition of a zero matrix ."""

implementation:
Q0 = empty ( ( n , 0 ) )
Q1 = eye ( n )
S0 = empty ( 0 )
return ( ( Q0 , Q1 ) , S0 )


signature:
def get_fast_scanner ( self ) :
    """Return : class : ` . FastScanner ` for association scan .
 Returns
 : class : ` . FastScanner `
 Instance of a class designed to perform very fast association scan ."""

implementation:
terms = self . _terms
return KronFastScanner ( self . _Y , self . _mean . A , self . _mean . X , self . _cov . Ge , terms )


signature:
def lml ( self ) :
    """Log of the marginal likelihood .
 Let y = vec ( Y ) , M = A ⊗ X , and H = MTK − 1M . The restricted log of the marginal
 likelihood is given by [ R07 ] _ : :
 2 ⋅ log ( p ( y ) ) = - ( n ⋅ p - c ⋅ p ) log ( 2π ) + log ( | MTM | ) - log ( | K | ) - log ( | H | )
 - ( y - m ) T K − 1 ( y - m ) ,
 where m = Mβ for β = H − 1MTK − 1y .
 For implementation purpose , let X = ( L0 ⊗ G ) and R = ( L1 ⊗ I ) ( L1 ⊗ I ) T .
 The covariance can be written as : :
 K = XXT + R .
 From the Woodbury matrix identity , we have
 yTK − 1y = yTR − 1y - yTR − 1XZ − 1XTR − 1y ,
 where Z = I + XTR − 1X . Note that R − 1 = ( U1S1 − 1U1T ) ⊗ I and : :
 XTR − 1y = ( L0TW ⊗ GT ) y = vec ( GTYWL0 ) ,
 where W = U1S1 − 1U1T . The term GTY can be calculated only once and it will form a
 r × p matrix . We similarly have : :
 XTR − 1M = ( L0TWA ) ⊗ ( GTX ) ,
 for which GTX is pre - computed .
 The log - determinant of the covariance matrix is given by
 log ( | K | ) = log ( | Z | ) - log ( | R − 1 | ) = log ( | Z | ) - 2 · n · log ( | U1S1 − 1 ⁄ 2 | ) .
 The log of the marginal likelihood can be rewritten as : :
 2 ⋅ log ( p ( y ) ) = - ( n ⋅ p - c ⋅ p ) log ( 2π ) + log ( | MTM | )
 - log ( | Z | ) + 2 · n · log ( | U1S1 − 1 ⁄ 2 | )
 - log ( | MTR − 1M - MTR − 1XZ − 1XTR − 1M | )
 - yTR − 1y + ( yTR − 1X ) Z − 1 ( XTR − 1y )
 - mTR − 1m + ( mTR − 1X ) Z − 1 ( XTR − 1m )
 + 2yTR − 1m - 2 ( yTR − 1X ) Z − 1 ( XTR − 1m ) .
 Returns
 lml : float
 Log of the marginal likelihood .
 References
 . . [ R07 ] LaMotte , L . R . ( 2007 ) . A direct derivation of the REML likelihood
 function . Statistical Papers , 48(2 ) , 321-327."""

implementation:
terms = self . _terms
yKiy = terms [ "yKiy" ]
mKiy = terms [ "mKiy" ]
mKim = terms [ "mKim" ]
lml = - self . _df * log2pi + self . _logdet_MM - self . _logdetK
lml -= self . _logdetH
lml += - yKiy - mKim + 2 * mKiy
return lml / 2


signature:
def _lml_gradient ( self ) :
    """Gradient of the log of the marginal likelihood .
 Let y = vec ( Y ) , K = K − 1 ∂ ( K ) K − 1 , and H = MTK − 1M . The gradient is given by : :
 2 ⋅ ∂ log ( p ( y ) ) = - tr ( K − 1 ∂ K ) - tr ( H − 1 ∂ H ) + yTKy - mTK ( 2 ⋅ y - m )
 - 2 ⋅ ( m - y ) TK − 1 ∂ ( m ) .
 Observe that
 ∂ β = - H − 1 ( ∂ H ) β - H − 1MTKy and ∂ H = - MTKM .
 Let Z = I + XTR − 1X and R = R − 1 ( ∂ K ) R − 1 . We use Woodbury matrix identity to
 write : :
 yTKy = yTRy - 2 ( yTRX ) Z − 1 ( XTR − 1y ) + ( yTR − 1X ) Z − 1 ( XTRX ) Z − 1 ( XTR − 1y )
 MTKM = MTRM - 2 ( MTRX ) Z − 1 ( XTR − 1M ) + ( MTR − 1X ) Z − 1 ( XTRX ) Z − 1 ( XTR − 1M )
 MTKy = MTRy - ( MTR − 1X ) Z − 1 ( XTRy ) - ( MTRX ) Z − 1 ( XTR − 1y )
 + ( MTR − 1X ) Z − 1 ( XTRX ) Z − 1 ( XTR − 1y )
 H − 1 = MTR − 1M - ( MTR − 1X ) Z − 1 ( XTR − 1M ) ,
 where we have used parentheses to separate expressions
 that we will compute separately . For example , we have : :
 yTRy = yT ( U1S1 − 1U1T ⊗ I ) ( ∂ C0 ⊗ GGT ) ( U1S1 − 1U1T ⊗ I ) y
 = yT ( U1S1 − 1U1T ∂ C0 ⊗ G ) ( U1S1 − 1U1T ⊗ GT ) y
 = vec ( GTYU1S1 − 1U1T ∂ C0 ) Tvec ( GTYU1S1 − 1U1T ) ,
 when the derivative is over the parameters of C0 . Otherwise , we have
 yTRy = vec ( YU1S1 − 1U1T ∂ C1 ) Tvec ( YU1S1 − 1U1T ) .
 The above equations can be more compactly written as
 yTRy = vec ( EiTYW ∂ Ci ) Tvec ( EiTYW ) ,
 where W = U1S1 − 1U1T , E0 = G , and E1 = I . We will now just state the results for
 the other instances of the aBc form , which follow similar derivations : :
 XTRX = ( L0TW ∂ CiWL0 ) ⊗ ( GTEiEiTG )
 MTRy = ( ATW ∂ Ci ⊗ XTEi ) vec ( EiTYW ) = vec ( XTEiEiTYW ∂ CiWA )
 MTRX = ATW ∂ CiWL0 ⊗ XTEiEiTG
 MTRM = ATW ∂ CiWA ⊗ XTEiEiTX
 XTRy = GTEiEiTYW ∂ CiWL0
 From Woodbury matrix identity and Kronecker product properties we have : :
 tr ( K − 1 ∂ K ) = tr [ W ∂ Ci ] tr [ EiEiT ] - tr [ Z − 1 ( XTRX ) ]
 tr ( H − 1 ∂ H ) = - tr [ ( MTR − 1M ) ( MTKM ) ] + tr [ ( MTR − 1X ) Z − 1 ( XTR − 1M ) ( MTKM ) ]
 Note also that : :
 ∂ β = H − 1MTKMβ - H − 1MTKy .
 Returns
 C0 . Lu : ndarray
 Gradient of the log of the marginal likelihood over C0 parameters .
 C1 . Lu : ndarray
 Gradient of the log of the marginal likelihood over C1 parameters ."""

implementation:
from scipy . linalg import cho_solve
terms = self . _terms
dC0 = self . _cov . C0 . gradient ( ) [ "Lu" ]
dC1 = self . _cov . C1 . gradient ( ) [ "Lu" ]
b = terms [ "b" ]
W = terms [ "W" ]
Lh = terms [ "Lh" ]
Lz = terms [ "Lz" ]
WA = terms [ "WA" ]
WL0 = terms [ "WL0" ]
YW = terms [ "YW" ]
MRiM = terms [ "MRiM" ]
MRiy = terms [ "MRiy" ]
XRiM = terms [ "XRiM" ]
XRiy = terms [ "XRiy" ]
ZiXRiM = terms [ "ZiXRiM" ]
ZiXRiy = terms [ "ZiXRiy" ]
WdC0 = _mdot ( W , dC0 )
WdC1 = _mdot ( W , dC1 )
AWdC0 = _mdot ( WA . T , dC0 )
AWdC1 = _mdot ( WA . T , dC1 )
# MTRM
MR0M = _mkron ( _mdot ( AWdC0 , WA ) , self . _XGGX )
MR1M = _mkron ( _mdot ( AWdC1 , WA ) , self . _XX )
# MTRX
MR0X = _mkron ( _mdot ( AWdC0 , WL0 ) , self . _XGGG )
MR1X = _mkron ( _mdot ( AWdC1 , WL0 ) , self . _GX . T )
# MTRy = ( ATW ∂ Ci ⊗ XTEi ) vec ( EiTYW ) = vec ( XTEiEiTYW ∂ CiWA )
MR0y = vec ( _mdot ( self . _XGGY , _mdot ( WdC0 , WA ) ) )
MR1y = vec ( _mdot ( self . _XY , WdC1 , WA ) )
# XTRX
XR0X = _mkron ( _mdot ( WL0 . T , dC0 , WL0 ) , self . _GGGG )
XR1X = _mkron ( _mdot ( WL0 . T , dC1 , WL0 ) , self . _GG )
# XTRy
XR0y = vec ( _mdot ( self . _GGGY , WdC0 , WL0 ) )
XR1y = vec ( _mdot ( self . _GY , WdC1 , WL0 ) )
# yTRy = vec ( EiTYW ∂ Ci ) Tvec ( EiTYW )
yR0y = vec ( _mdot ( self . _GY , WdC0 ) ) . T @ vec ( self . _GY @ W )
yR1y = ( YW . T * _mdot ( self . _Y , WdC1 ) . T ) . T . sum ( axis = ( 0 , 1 ) )
ZiXR0X = cho_solve ( Lz , XR0X )
ZiXR1X = cho_solve ( Lz , XR1X )
ZiXR0y = cho_solve ( Lz , XR0y )
ZiXR1y = cho_solve ( Lz , XR1y )
# MTKy = MTRy - ( MTR − 1X ) Z − 1 ( XTRy ) - ( MTRX ) Z − 1 ( XTR − 1y )
# + ( MTR − 1X ) Z − 1 ( XTRX ) Z − 1 ( XTR − 1y )
MK0y = MR0y - _mdot ( XRiM . T , ZiXR0y ) - _mdot ( MR0X , ZiXRiy )
MK0y += _mdot ( XRiM . T , ZiXR0X , ZiXRiy )
MK1y = MR1y - _mdot ( XRiM . T , ZiXR1y ) - _mdot ( MR1X , ZiXRiy )
MK1y += _mdot ( XRiM . T , ZiXR1X , ZiXRiy )
# yTKy = yTRy - 2 ( yTRX ) Z − 1 ( XTR − 1y ) + ( yTR − 1X ) Z − 1 ( XTRX ) Z − 1 ( XTR − 1y )
yK0y = yR0y - 2 * XR0y . T @ ZiXRiy + ZiXRiy . T @ _mdot ( XR0X , ZiXRiy )
yK1y = yR1y - 2 * XR1y . T @ ZiXRiy + ZiXRiy . T @ _mdot ( XR1X , ZiXRiy )
# MTKM = MTRM - ( MTRX ) Z − 1 ( XTR − 1M ) - ( MTR − 1X ) Z − 1 ( XTRM )
# + ( MTR − 1X ) Z − 1 ( XTRX ) Z − 1 ( XTR − 1M )
MR0XZiXRiM = _mdot ( MR0X , ZiXRiM )
MK0M = MR0M - MR0XZiXRiM - MR0XZiXRiM . transpose ( [ 1 , 0 , 2 ] )
MK0M += _mdot ( ZiXRiM . T , XR0X , ZiXRiM )
MR1XZiXRiM = _mdot ( MR1X , ZiXRiM )
MK1M = MR1M - MR1XZiXRiM - MR1XZiXRiM . transpose ( [ 1 , 0 , 2 ] )
MK1M += _mdot ( ZiXRiM . T , XR1X , ZiXRiM )
MK0m = _mdot ( MK0M , b )
mK0y = b . T @ MK0y
mK0m = b . T @ MK0m
MK1m = _mdot ( MK1M , b )
mK1y = b . T @ MK1y
mK1m = b . T @ MK1m
XRim = XRiM @ b
MRim = MRiM @ b
db = { "C0.Lu" : cho_solve ( Lh , MK0m - MK0y ) , "C1.Lu" : cho_solve ( Lh , MK1m - MK1y ) }
grad = { "C0.Lu" : - trace ( WdC0 ) * self . _trGG + trace ( ZiXR0X ) , "C1.Lu" : - trace ( WdC1 ) * self . nsamples + trace ( ZiXR1X ) , }
if self . _restricted :
    grad [ "C0.Lu" ] += cho_solve ( Lh , MK0M ) . diagonal ( ) . sum ( 1 )
    grad [ "C1.Lu" ] += cho_solve ( Lh , MK1M ) . diagonal ( ) . sum ( 1 )
mKiM = MRim . T - XRim . T @ ZiXRiM
yKiM = MRiy . T - XRiy . T @ ZiXRiM
grad [ "C0.Lu" ] += yK0y - 2 * mK0y + mK0m - 2 * _mdot ( mKiM , db [ "C0.Lu" ] )
grad [ "C0.Lu" ] += 2 * _mdot ( yKiM , db [ "C0.Lu" ] )
grad [ "C1.Lu" ] += yK1y - 2 * mK1y + mK1m - 2 * _mdot ( mKiM , db [ "C1.Lu" ] )
grad [ "C1.Lu" ] += 2 * _mdot ( yKiM , db [ "C1.Lu" ] )
grad [ "C0.Lu" ] /= 2
grad [ "C1.Lu" ] /= 2
return grad


signature:
def gradient ( self ) :
    r"""Gradient of the log of the marginal likelihood .
 Returns
 dict
 Map between variables to their gradient values ."""

implementation:
self . _update_approx ( )
g = self . _ep . lml_derivatives ( self . _X )
ed = exp ( - self . logitdelta )
es = exp ( self . logscale )
grad = dict ( )
grad [ "logitdelta" ] = g [ "delta" ] * ( ed / ( 1 + ed ) ) / ( 1 + ed )
grad [ "logscale" ] = g [ "scale" ] * es
grad [ "beta" ] = g [ "mean" ]
return grad


signature:
def gradient ( self ) :
    """Derivative of the covariance matrix over the lower triangular , flat part of L .
 It is equal to
 ∂ K / ∂ Lij = ALT + LAT ,
 where Aij is an n × m matrix of zeros except at [ Aij ] ij = 1.
 Returns
 Lu : ndarray
 Derivative of K over the lower - triangular , flat part of L ."""

implementation:
L = self . L
n = self . L . shape [ 0 ]
grad = { "Lu" : zeros ( ( n , n , n * self . _L . shape [ 1 ] ) ) }
for ii in range ( self . _L . shape [ 0 ] * self . _L . shape [ 1 ] ) :
    row = ii // self . _L . shape [ 1 ]
    col = ii % self . _L . shape [ 1 ]
    grad [ "Lu" ] [ row , : , ii ] = L [ : , col ]
    grad [ "Lu" ] [ : , row , ii ] += L [ : , col ]
return grad


signature:
def beta ( self ) :
    """Fixed - effect sizes .
 Returns
 effect - sizes : numpy . ndarray
 Optimal fixed - effect sizes .
 Notes
 Setting the derivative of log ( p ( y ) ) over effect sizes equal
 to zero leads to solutions β from equation : :
 ( QTX ) TD − 1 ( QTX ) β = ( QTX ) TD − 1 ( QTy ) ."""

implementation:
from numpy_sugar . linalg import rsolve
return rsolve ( self . _X [ "VT" ] , rsolve ( self . _X [ "tX" ] , self . mean ( ) ) )


signature:
def beta_covariance ( self ) :
    """Estimates the covariance - matrix of the optimal beta .
 Returns
 beta - covariance : ndarray
 ( XT ( s ( ( 1 - δ ) K + δI ) ) − 1X ) − 1.
 References
 . . Rencher , A . C . , & Schaalje , G . B . ( 2008 ) . Linear models in statistics . John
 Wiley & Sons ."""

implementation:
from numpy_sugar . linalg import ddot
tX = self . _X [ "tX" ]
Q = concatenate ( self . _QS [ 0 ] , axis = 1 )
S0 = self . _QS [ 1 ]
D = self . v0 * S0 + self . v1
D = D . tolist ( ) + [ self . v1 ] * ( len ( self . _y ) - len ( D ) )
D = asarray ( D )
A = inv ( tX . T @ ( Q @ ddot ( 1 / D , Q . T @ tX ) ) )
VT = self . _X [ "VT" ]
H = lstsq ( VT , A , rcond = None ) [ 0 ]
return lstsq ( VT , H . T , rcond = None ) [ 0 ]


signature:
def fix ( self , param ) :
    """Disable parameter optimization .
 Parameters
 param : str
 Possible values are ` ` " delta " ` ` , ` ` " beta " ` ` , and ` ` " scale " ` ` ."""

implementation:
if param == "delta" :
    super ( ) . _fix ( "logistic" )
else :
    self . _fix [ param ] = True


signature:
def unfix ( self , param ) :
    """Enable parameter optimization .
 Parameters
 param : str
 Possible values are ` ` " delta " ` ` , ` ` " beta " ` ` , and ` ` " scale " ` ` ."""

implementation:
if param == "delta" :
    self . _unfix ( "logistic" )
else :
    self . _fix [ param ] = False


signature:
def fit ( self , verbose = True ) :
    """Maximise the marginal likelihood .
 Parameters
 verbose : bool , optional
 ` ` True ` ` for progress output ; ` ` False ` ` otherwise .
 Defaults to ` ` True ` ` ."""

implementation:
if not self . _isfixed ( "logistic" ) :
    self . _maximize_scalar ( desc = "LMM" , rtol = 1e-6 , atol = 1e-6 , verbose = verbose )
if not self . _fix [ "beta" ] :
    self . _update_beta ( )
if not self . _fix [ "scale" ] :
    self . _update_scale ( )


signature:
def get_fast_scanner ( self ) :
    """Return : class : ` . FastScanner ` for association scan .
 Returns
 fast - scanner : : class : ` . FastScanner `
 Instance of a class designed to perform very fast association scan ."""

implementation:
v0 = self . v0
v1 = self . v1
QS = ( self . _QS [ 0 ] , v0 * self . _QS [ 1 ] )
return FastScanner ( self . _y , self . X , QS , v1 )


signature:
def value ( self ) :
    """Internal use only ."""

implementation:
if not self . _fix [ "beta" ] :
    self . _update_beta ( )
if not self . _fix [ "scale" ] :
    self . _update_scale ( )
return self . lml ( )


signature:
def lml ( self ) :
    """Log of the marginal likelihood .
 Returns
 lml : float
 Log of the marginal likelihood .
 Notes
 The log of the marginal likelihood is given by : :
 2 ⋅ log ( p ( y ) ) = - n ⋅ log ( 2π ) - n ⋅ log ( s ) - log | D | - ( QTy ) Ts − 1D − 1 ( QTy )
 + ( QTy ) Ts − 1D − 1 ( QTXβ ) / 2 - ( QTXβ ) Ts − 1D − 1 ( QTXβ ) .
 By using the optimal β , the log of the marginal likelihood can be rewritten
 as : :
 2 ⋅ log ( p ( y ) ) = - n ⋅ log ( 2π ) - n ⋅ log ( s ) - log | D | + ( QTy ) Ts − 1D − 1QT ( Xβ - y ) .
 In the extreme case where β is such that y = Xβ , the maximum is attained as
 s → 0.
 For optimals β and s , the log of the marginal likelihood can be further
 simplified to : :
 2 ⋅ log ( p ( y ; β , s ) ) = - n ⋅ log ( 2π ) - n ⋅ log s - log | D | - n ."""

implementation:
reml = ( self . _logdetXX ( ) - self . _logdetH ( ) ) / 2
if self . _optimal [ "scale" ] :
    lml = self . _lml_optimal_scale ( )
else :
    lml = self . _lml_arbitrary_scale ( )
return lml + reml


signature:
def delta ( self ) :
    """Variance ratio between ` ` K ` ` and ` ` I ` ` ."""

implementation:
v = float ( self . _logistic . value )
if v > 0.0 :
    v = 1 / ( 1 + exp ( - v ) )
else :
    v = exp ( v )
    v = v / ( v + 1.0 )
return min ( max ( v , epsilon . tiny ) , 1 - epsilon . tiny )


signature:
def _logdetXX ( self ) :
    """log ( | XTX | ) ."""

implementation:
if not self . _restricted :
    return 0.0
ldet = slogdet ( self . _X [ "tX" ] . T @ self . _X [ "tX" ] )
if ldet [ 0 ] != 1.0 :
    raise ValueError ( "The determinant of XᵀX should be positive." )
return ldet [ 1 ]


signature:
def _logdetH ( self ) :
    """log ( | H | ) for H = s − 1XTQD − 1QTX ."""

implementation:
if not self . _restricted :
    return 0.0
ldet = slogdet ( sum ( self . _XTQDiQTX ) / self . scale )
if ldet [ 0 ] != 1.0 :
    raise ValueError ( "The determinant of H should be positive." )
return ldet [ 1 ]


signature:
def _lml_optimal_scale ( self ) :
    """Log of the marginal likelihood for optimal scale .
 Implementation for unrestricted LML : :
 Returns
 lml : float
 Log of the marginal likelihood ."""

implementation:
assert self . _optimal [ "scale" ]
n = len ( self . _y )
lml = - self . _df * log2pi - self . _df - n * log ( self . scale )
lml -= sum ( npsum ( log ( D ) ) for D in self . _D )
return lml / 2


signature:
def _lml_arbitrary_scale ( self ) :
    """Log of the marginal likelihood for arbitrary scale .
 Returns
 lml : float
 Log of the marginal likelihood ."""

implementation:
s = self . scale
D = self . _D
n = len ( self . _y )
lml = - self . _df * log2pi - n * log ( s )
lml -= sum ( npsum ( log ( d ) ) for d in D )
d = ( mTQ - yTQ for ( mTQ , yTQ ) in zip ( self . _mTQ , self . _yTQ ) )
lml -= sum ( ( i / j ) @ i for ( i , j ) in zip ( d , D ) ) / s
return lml / 2


signature:
def _df ( self ) :
    """Degrees of freedom ."""

implementation:
if not self . _restricted :
    return self . nsamples
return self . nsamples - self . _X [ "tX" ] . shape [ 1 ]


signature:
def get_fast_scanner ( self ) :
    r"""Return : class : ` glimix _ core . lmm . FastScanner ` for the current
 delta ."""

implementation:
from numpy_sugar . linalg import ddot , economic_qs , sum2diag
y = self . eta / self . tau
if self . _QS is None :
    K = eye ( y . shape [ 0 ] ) / self . tau
else :
    Q0 = self . _QS [ 0 ] [ 0 ]
    S0 = self . _QS [ 1 ]
    K = dot ( ddot ( Q0 , self . v0 * S0 ) , Q0 . T )
    K = sum2diag ( K , 1 / self . tau )
return FastScanner ( y , self . _X , economic_qs ( K ) , self . v1 )


signature:
def value ( self ) :
    r"""Log of the marginal likelihood .
 Formally ,
 . . math : :
 - \ frac { n } { 2 } \ log { 2 \ pi } - \ frac { 1 } { 2 } \ log { \ left |
 v _ 0 \ mathrm K + v _ 1 \ mathrm I + \ tilde { \ Sigma } \ right | }
 - \ frac { 1 } { 2}
 \ left ( \ tilde { \ boldsymbol \ mu } -
 \ mathrm X \ boldsymbol \ beta \ right ) ^ { \ intercal }
 \ left ( v _ 0 \ mathrm K + v _ 1 \ mathrm I +
 \ tilde { \ Sigma } \ right ) ^ { - 1}
 \ left ( \ tilde { \ boldsymbol \ mu } -
 \ mathrm X \ boldsymbol \ beta \ right )
 Returns
 float
 : math : ` \ log { p ( \ tilde { \ boldsymbol \ mu } ) } `"""

implementation:
from numpy_sugar . linalg import ddot , sum2diag
if self . _cache [ "value" ] is not None :
    return self . _cache [ "value" ]
scale = exp ( self . logscale )
delta = 1 / ( 1 + exp ( - self . logitdelta ) )
v0 = scale * ( 1 - delta )
v1 = scale * delta
mu = self . eta / self . tau
n = len ( mu )
if self . _QS is None :
    K = zeros ( ( n , n ) )
else :
    Q0 = self . _QS [ 0 ] [ 0 ]
    S0 = self . _QS [ 1 ]
    K = dot ( ddot ( Q0 , S0 ) , Q0 . T )
A = sum2diag ( sum2diag ( v0 * K , v1 ) , 1 / self . tau )
m = mu - self . mean ( )
v = - n * log ( 2 * pi )
v -= slogdet ( A ) [ 1 ]
v -= dot ( m , solve ( A , m ) )
self . _cache [ "value" ] = v / 2
return self . _cache [ "value" ]


signature:
def _initialize ( self ) :
    r"""Initialize the mean and covariance of the posterior .
 Given that : math : ` \ tilde { \ mathrm T } ` is a matrix of zeros right before
 the first EP iteration , we have
 . . math : :
 \ boldsymbol \ mu = \ mathrm K ^ { - 1 } \ mathbf m ~ \ text { and } ~
 \ Sigma = \ mathrm K
 as the initial posterior mean and covariance ."""

implementation:
if self . _mean is None or self . _cov is None :
    return
Q = self . _cov [ "QS" ] [ 0 ] [ 0 ]
S = self . _cov [ "QS" ] [ 1 ]
if S . size > 0 :
    self . tau [ : ] = 1 / npsum ( ( Q * sqrt ( S ) ) ** 2 , axis = 1 )
else :
    self . tau [ : ] = 0.0
self . eta [ : ] = self . _mean
self . eta [ : ] *= self . tau


signature:
def L ( self ) :
    r"""Cholesky decomposition of : math : ` \ mathrm B ` .
 . . math : :
 \ mathrm B = \ mathrm Q ^ { \ intercal } \ tilde { \ mathrm { T } } \ mathrm Q
 + \ mathrm { S } ^ { - 1}"""

implementation:
from scipy . linalg import cho_factor
from numpy_sugar . linalg import ddot , sum2diag
if self . _L_cache is not None :
    return self . _L_cache
Q = self . _cov [ "QS" ] [ 0 ] [ 0 ]
S = self . _cov [ "QS" ] [ 1 ]
B = dot ( Q . T , ddot ( self . _site . tau , Q , left = True ) )
sum2diag ( B , 1.0 / S , out = B )
self . _L_cache = cho_factor ( B , lower = True ) [ 0 ]
return self . _L_cache


signature:
def build_engine_session ( connection , echo = False , autoflush = None , autocommit = None , expire_on_commit = None , scopefunc = None ) :
    """Build an engine and a session .
 : param str connection : An RFC - 1738 database connection string
 : param bool echo : Turn on echoing SQL
 : param Optional [ bool ] autoflush : Defaults to True if not specified in kwargs or configuration .
 : param Optional [ bool ] autocommit : Defaults to False if not specified in kwargs or configuration .
 : param Optional [ bool ] expire _ on _ commit : Defaults to False if not specified in kwargs or configuration .
 : param scopefunc : Scoped function to pass to : func : ` sqlalchemy . orm . scoped _ session `
 : rtype : tuple [ Engine , Session ]
 From the Flask - SQLAlchemy documentation :
 An extra key ` ` ' scopefunc ' ` ` can be set on the ` ` options ` ` dict to
 specify a custom scope function . If it ' s not provided , Flask ' s app
 context stack identity is used . This will ensure that sessions are
 created and removed with the request / response cycle , and should be fine
 in most cases ."""

implementation:
if connection is None :
    raise ValueError ( 'can not build engine when connection is None' )
engine = create_engine ( connection , echo = echo )
autoflush = autoflush if autoflush is not None else False
autocommit = autocommit if autocommit is not None else False
expire_on_commit = expire_on_commit if expire_on_commit is not None else True
log . debug ( 'auto flush: %s, auto commit: %s, expire on commmit: %s' , autoflush , autocommit , expire_on_commit )
# : A SQLAlchemy session maker
session_maker = sessionmaker ( bind = engine , autoflush = autoflush , autocommit = autocommit , expire_on_commit = expire_on_commit , )
# : A SQLAlchemy session object
session = scoped_session ( session_maker , scopefunc = scopefunc )
return engine , session


signature:
def _get_connection ( cls , connection : Optional [ str ] = None ) -> str :
    """Get a default connection string .
 Wraps : func : ` bio2bel . utils . get _ connection ` and passing this class ' s : data : ` module _ name ` to it ."""

implementation: return get_connection ( cls . module_name , connection = connection )


signature:
def setup_smtp_factory ( ** settings ) :
    """expects a dictionary with ' mail . ' keys to create an appropriate smtplib . SMTP instance"""

implementation: return CustomSMTP ( host = settings . get ( 'mail.host' , 'localhost' ) , port = int ( settings . get ( 'mail.port' , 25 ) ) , user = settings . get ( 'mail.user' ) , password = settings . get ( 'mail.password' ) , timeout = float ( settings . get ( 'mail.timeout' , 60 ) ) , )


signature:
def sendMultiPart ( smtp , gpg_context , sender , recipients , subject , text , attachments ) :
    """a helper method that composes and sends an email with attachments
 requires a pre - configured smtplib . SMTP instance"""

implementation:
sent = 0
for to in recipients :
    if not to . startswith ( '<' ) :
        uid = '<%s>' % to
    else :
        uid = to
    if not checkRecipient ( gpg_context , uid ) :
        continue
    msg = MIMEMultipart ( )
    msg [ 'From' ] = sender
    msg [ 'To' ] = to
    msg [ 'Subject' ] = subject
    msg [ "Date" ] = formatdate ( localtime = True )
    msg . preamble = u'This is an email in encrypted multipart format.'
    attach = MIMEText ( str ( gpg_context . encrypt ( text . encode ( 'utf-8' ) , uid , always_trust = True ) ) )
    attach . set_charset ( 'UTF-8' )
    msg . attach ( attach )
    for attachment in attachments :
        with open ( attachment , 'rb' ) as fp :
            attach = MIMEBase ( 'application' , 'octet-stream' )
            attach . set_payload ( str ( gpg_context . encrypt_file ( fp , uid , always_trust = True ) ) )
        attach . add_header ( 'Content-Disposition' , 'attachment' , filename = basename ( '%s.pgp' % attachment ) )
        msg . attach ( attach )
    # TODO : need to catch exception ?
    # yes : - ) we need to adjust the status accordingly ( > 500 so it will be destroyed )
    smtp . begin ( )
    smtp . sendmail ( sender , to , msg . as_string ( ) )
    smtp . quit ( )
    sent += 1
return sent


signature:
def begin ( self ) :
    """connects and optionally authenticates a connection ."""

implementation:
self . connect ( self . host , self . port )
if self . user :
    self . starttls ( )
    self . login ( self . user , self . password )


signature:
def generate ( self , ** kwargs ) :
    '''Generate a : term : ` URI ` based on parameters passed .
 : param id : The id of the concept or collection .
 : param type : What we ' re generating a : term : ` URI ` for : ` concept `
 or ` collection ` .
 : rtype : string'''

implementation:
if kwargs [ 'type' ] not in [ 'concept' , 'collection' ] :
    raise ValueError ( 'Type %s is invalid' % kwargs [ 'type' ] )
return ( self . pattern % ( self . vocabulary_id , kwargs [ 'type' ] , kwargs [ 'id' ] ) ) . lower ( )


signature:
def has_address ( start : int , data_length : int ) -> bool :
    """Determine whether the packet has an " address " encoded into it .
 There exists an undocumented bug / edge case in the spec - some packets
 with 0x82 as _ start _ , still encode the address into the packet , and thus
 throws off decoding . This edge case is handled explicitly ."""

implementation: return bool ( 0x01 & start ) or ( start == 0x82 and data_length == 16 )


signature:
def decode_timestamp ( data : str ) -> datetime . datetime :
    """Decode timestamp using bespoke decoder .
 Cannot use simple strptime since the ness panel contains a bug
 that P199E zone and state updates emitted on the hour cause a minute
 value of ` 60 ` to be sent , causing strptime to fail . This decoder handles
 this edge case ."""

implementation:
year = 2000 + int ( data [ 0 : 2 ] )
month = int ( data [ 2 : 4 ] )
day = int ( data [ 4 : 6 ] )
hour = int ( data [ 6 : 8 ] )
minute = int ( data [ 8 : 10 ] )
second = int ( data [ 10 : 12 ] )
if minute == 60 :
    minute = 0
    hour += 1
return datetime . datetime ( year = year , month = month , day = day , hour = hour , minute = minute , second = second )


signature:
def create_application ( connection : Optional [ str ] = None ) -> Flask :
    """Create a Flask application ."""

implementation:
app = Flask ( __name__ )
flask_bootstrap . Bootstrap ( app )
Admin ( app )
connection = connection or DEFAULT_CACHE_CONNECTION
engine , session = build_engine_session ( connection )
for name , add_admin in add_admins . items ( ) :
    url = '/{}' . format ( name )
    add_admin ( app , session , url = url , endpoint = name , name = name )
    log . debug ( 'added %s - %s to %s' , name , add_admin , url )
app . register_blueprint ( ui )
return app


signature:
def register_provider ( self , provider ) :
    '''Register a : class : ` skosprovider . providers . VocabularyProvider ` .
 : param skosprovider . providers . VocabularyProvider provider : The provider
 to register .
 : raises RegistryException : A provider with this id or uri has already
 been registered .'''

implementation:
if provider . get_vocabulary_id ( ) in self . providers :
    raise RegistryException ( 'A provider with this id has already been registered.' )
self . providers [ provider . get_vocabulary_id ( ) ] = provider
if provider . concept_scheme . uri in self . concept_scheme_uri_map :
    raise RegistryException ( 'A provider with URI %s has already been registered.' % provider . concept_scheme . uri )
self . concept_scheme_uri_map [ provider . concept_scheme . uri ] = provider . get_vocabulary_id ( )


signature:
def remove_provider ( self , id ) :
    '''Remove the provider with the given id or : term : ` URI ` .
 : param str id : The identifier for the provider .
 : returns : A : class : ` skosprovider . providers . VocabularyProvider ` or
 ` False ` if the id is unknown .'''

implementation:
if id in self . providers :
    p = self . providers . get ( id , False )
    del self . providers [ id ]
    del self . concept_scheme_uri_map [ p . concept_scheme . uri ]
    return p
elif id in self . concept_scheme_uri_map :
    id = self . concept_scheme_uri_map [ id ]
    return self . remove_provider ( id )
else :
    return False


signature:
def get_provider ( self , id ) :
    '''Get a provider by id or : term : ` uri ` .
 : param str id : The identifier for the provider . This can either be the
 id with which it was registered or the : term : ` uri ` of the conceptscheme
 that the provider services .
 : returns : A : class : ` skosprovider . providers . VocabularyProvider `
 or ` False ` if the id or uri is unknown .'''

implementation:
if id in self . providers :
    return self . providers . get ( id , False )
elif is_uri ( id ) and id in self . concept_scheme_uri_map :
    return self . providers . get ( self . concept_scheme_uri_map [ id ] , False )
return False


signature:
def get_providers ( self , ** kwargs ) :
    '''Get all providers registered .
 If keyword ` ids ` is present , get only the providers with these ids .
 If keys ` subject ` is present , get only the providers that have this subject .
 . . code - block : : python
 # Get all providers with subject ' biology '
 registry . get _ providers ( subject = ' biology ' )
 # Get all providers with id 1 or 2
 registry . get _ providers ( ids = [ 1,2 ] )
 # Get all providers with id 1 or 2 and subject ' biology '
 registry . get _ providers ( ids = [ 1,2 ] , subject = ' biology ' ]
 : param list ids : Only return providers with one of the Ids or : term : ` URIs < uri > ` .
 : param str subject : Only return providers with this subject .
 : returns : A list of : class : ` providers < skosprovider . providers . VocabularyProvider > `'''

implementation:
if 'ids' in kwargs :
    ids = [ self . concept_scheme_uri_map . get ( id , id ) for id in kwargs [ 'ids' ] ]
    providers = [ self . providers [ k ] for k in self . providers . keys ( ) if k in ids ]
else :
    providers = list ( self . providers . values ( ) )
if 'subject' in kwargs :
    providers = [ p for p in providers if kwargs [ 'subject' ] in p . metadata [ 'subject' ] ]
return providers


signature:
def find ( self , query , ** kwargs ) :
    '''Launch a query across all or a selection of providers .
 . . code - block : : python
 # Find anything that has a label of church in any provider .
 registry . find ( { ' label ' : ' church ' } )
 # Find anything that has a label of church with the BUILDINGS provider .
 # Attention , this syntax was deprecated in version 0.3.0
 registry . find ( { ' label ' : ' church ' } , providers = [ ' BUILDINGS ' ] )
 # Find anything that has a label of church with the BUILDINGS provider .
 registry . find ( { ' label ' : ' church ' } , providers = { ' ids ' : [ ' BUILDINGS ' ] } )
 # Find anything that has a label of church with a provider
 # marked with the subject ' architecture ' .
 registry . find ( { ' label ' : ' church ' } , providers = { ' subject ' : ' architecture ' } )
 # Find anything that has a label of church in any provider .
 # If possible , display the results with a Dutch label .
 registry . find ( { ' label ' : ' church ' } , language = ' nl ' )
 : param dict query : The query parameters that will be passed on to each
 : meth : ` ~ skosprovider . providers . VocabularyProvider . find ` method of
 the selected .
 : class : ` providers < skosprovider . providers . VocabularyProvider > ` .
 : param dict providers : Optional . If present , it should be a dictionary .
 This dictionary can contain any of the keyword arguments available
 to the : meth : ` get _ providers ` method . The query will then only
 be passed to the providers confirming to these arguments .
 : param string language : Optional . If present , it should be a
 : term : ` language - tag ` . This language - tag is passed on to the
 underlying providers and used when selecting the label to display
 for each concept .
 : returns : a list of : class : ` dict ` .
 Each dict has two keys : id and concepts .'''

implementation:
if 'providers' not in kwargs :
    providers = self . get_providers ( )
else :
    pargs = kwargs [ 'providers' ]
    if isinstance ( pargs , list ) :
        providers = self . get_providers ( ids = pargs )
    else :
        providers = self . get_providers ( ** pargs )
kwarguments = { }
if 'language' in kwargs :
    kwarguments [ 'language' ] = kwargs [ 'language' ]
return [ { 'id' : p . get_vocabulary_id ( ) , 'concepts' : p . find ( query , ** kwarguments ) } for p in providers ]


signature:
def get_all ( self , ** kwargs ) :
    '''Get all concepts from all providers .
 . . code - block : : python
 # get all concepts in all providers .
 registry . get _ all ( )
 # get all concepts in all providers .
 # If possible , display the results with a Dutch label .
 registry . get _ all ( language = ' nl ' )
 : param string language : Optional . If present , it should be a
 : term : ` language - tag ` . This language - tag is passed on to the
 underlying providers and used when selecting the label to display
 for each concept .
 : returns : a list of : class : ` dict ` .
 Each dict has two keys : id and concepts .'''

implementation:
kwarguments = { }
if 'language' in kwargs :
    kwarguments [ 'language' ] = kwargs [ 'language' ]
return [ { 'id' : p . get_vocabulary_id ( ) , 'concepts' : p . get_all ( ** kwarguments ) } for p in self . providers . values ( ) ]


signature:
def get_by_uri ( self , uri ) :
    '''Get a concept or collection by its uri .
 Returns a single concept or collection if one exists with this uri .
 Returns False otherwise .
 : param string uri : The uri to find a concept or collection for .
 : raises ValueError : The uri is invalid .
 : rtype : : class : ` skosprovider . skos . Concept ` or
 : class : ` skosprovider . skos . Collection `'''

implementation:
if not is_uri ( uri ) :
    raise ValueError ( '%s is not a valid URI.' % uri )
# Check if there ' s a provider that ' s more likely to have the URI
csuris = [ csuri for csuri in self . concept_scheme_uri_map . keys ( ) if uri . startswith ( csuri ) ]
for csuri in csuris :
    c = self . get_provider ( csuri ) . get_by_uri ( uri )
    if c :
        return c
# Check all providers
for p in self . providers . values ( ) :
    c = p . get_by_uri ( uri )
    if c :
        return c
return False


signature:
def find_module ( self , fullname , path = None ) :
    """Find a module if its name starts with : code : ` self . group ` and is registered ."""

implementation:
if not fullname . startswith ( self . _group_with_dot ) :
    return
end_name = fullname [ len ( self . _group_with_dot ) : ]
for entry_point in iter_entry_points ( group = self . group , name = None ) :
    if entry_point . name == end_name :
        return self


signature:
def load_module ( self , fullname ) :
    """Load a module if its name starts with : code : ` self . group ` and is registered ."""

implementation:
if fullname in sys . modules :
    return sys . modules [ fullname ]
end_name = fullname [ len ( self . _group_with_dot ) : ]
for entry_point in iter_entry_points ( group = self . group , name = end_name ) :
    mod = entry_point . load ( )
    sys . modules [ fullname ] = mod
    return mod


signature:
def upload_theme ( ) :
    """upload and / or update the theme with the current git state"""

implementation:
get_vars ( )
with fab . settings ( ) :
    local_theme_path = path . abspath ( path . join ( fab . env [ 'config_base' ] , fab . env . instance . config [ 'local_theme_path' ] ) )
    rsync ( '-av' , '--delete' , '%s/' % local_theme_path , '{{host_string}}:{themes_dir}/{ploy_theme_name}' . format ( ** AV ) )
    briefkasten_ctl ( 'restart' )


signature:
def upload_pgp_keys ( ) :
    """upload and / or update the PGP keys for editors , import them into PGP"""

implementation:
get_vars ( )
upload_target = '/tmp/pgp_pubkeys.tmp'
with fab . settings ( fab . hide ( 'running' ) ) :
    fab . run ( 'rm -rf %s' % upload_target )
    fab . run ( 'mkdir %s' % upload_target )
    local_key_path = path . join ( fab . env [ 'config_base' ] , fab . env . instance . config [ 'local_pgpkey_path' ] )
    remote_key_path = '/var/briefkasten/pgp_pubkeys/' . format ( ** AV )
    rsync ( '-av' , local_key_path , '{host_string}:%s' % upload_target )
    fab . run ( 'chown -R %s %s' % ( AV [ 'appuser' ] , remote_key_path ) )
    fab . run ( 'chmod 700 %s' % remote_key_path )
    with fab . shell_env ( GNUPGHOME = remote_key_path ) :
        fab . sudo ( '''gpg --import %s/*.*''' % upload_target , user = AV [ 'appuser' ] , shell_escape = False )
    fab . run ( 'rm -rf %s' % upload_target )


signature:
def upload_backend ( index = 'dev' , user = None ) :
    """Build the backend and upload it to the remote server at the given index"""

implementation:
get_vars ( )
use_devpi ( index = index )
with fab . lcd ( '../application' ) :
    fab . local ( 'make upload' )


signature:
def update_backend ( use_pypi = False , index = 'dev' , build = True , user = None , version = None ) :
    """Install the backend from the given devpi index at the given version on the target host and restart the service .
 If version is None , it defaults to the latest version
 Optionally , build and upload the application first from local sources . This requires a
 full backend development environment on the machine running this command ( pyramid etc . )"""

implementation:
get_vars ( )
if value_asbool ( build ) :
    upload_backend ( index = index , user = user )
with fab . cd ( '{apphome}' . format ( ** AV ) ) :
    if value_asbool ( use_pypi ) :
        command = 'bin/pip install --upgrade briefkasten'
    else :
        command = 'bin/pip install --upgrade --pre -i {ploy_default_publish_devpi}/briefkasten/{index}/+simple/ briefkasten' . format ( index = index , user = user , ** AV )
    if version :
        command = '%s==%s' % ( command , version )
    fab . sudo ( command )
briefkasten_ctl ( 'restart' )


signature:
def _sort ( self , concepts , sort = None , language = 'any' , reverse = False ) :
    '''Returns a sorted version of a list of concepts . Will leave the original
 list unsorted .
 : param list concepts : A list of concepts and collections .
 : param string sort : What to sort on : ` id ` , ` label ` or ` sortlabel `
 : param string language : Language to use when sorting on ` label ` or
 ` sortlabel ` .
 : param boolean reverse : Reverse the sort order ?
 : rtype : list'''

implementation:
sorted = copy . copy ( concepts )
if sort :
    sorted . sort ( key = methodcaller ( '_sortkey' , sort , language ) , reverse = reverse )
return sorted


signature:
def _include_in_find ( self , c , query ) :
    ''': param c : A : class : ` skosprovider . skos . Concept ` or
 : class : ` skosprovider . skos . Collection ` .
 : param query : A dict that can be used to express a query .
 : rtype : boolean'''

implementation:
include = True
if include and 'type' in query :
    include = query [ 'type' ] == c . type
if include and 'label' in query :
    def finder ( l , query ) :
        if not self . case_insensitive :
            return l . label . find ( query [ 'label' ] )
        else :
            return l . label . upper ( ) . find ( query [ 'label' ] . upper ( ) )
    include = any ( [ finder ( l , query ) >= 0 for l in c . labels ] )
if include and 'collection' in query :
    coll = self . get_by_id ( query [ 'collection' ] [ 'id' ] )
    if not coll or not isinstance ( coll , Collection ) :
        raise ValueError ( 'You are searching for items in an unexisting collection.' )
    if 'depth' in query [ 'collection' ] and query [ 'collection' ] [ 'depth' ] == 'all' :
        members = self . expand ( coll . id )
    else :
        members = coll . members
    include = any ( [ True for id in members if str ( id ) == str ( c . id ) ] )
return include


signature:
def _get_find_dict ( self , c , ** kwargs ) :
    '''Return a dict that can be used in the return list of the : meth : ` find `
 method .
 : param c : A : class : ` skosprovider . skos . Concept ` or
 : class : ` skosprovider . skos . Collection ` .
 : rtype : dict'''

implementation:
language = self . _get_language ( ** kwargs )
return { 'id' : c . id , 'uri' : c . uri , 'type' : c . type , 'label' : None if c . label ( ) is None else c . label ( language ) . label }


signature:
async def update ( self ) -> None :
    """Force update of alarm status and zones"""

implementation:
_LOGGER . debug ( "Requesting state update from server (S00, S14)" )
await asyncio . gather ( # List unsealed Zones
self . send_command ( 'S00' ) , # Arming status update
self . send_command ( 'S14' ) , )


signature:
async def _update_loop ( self ) -> None :
    """Schedule a state update to keep the connection alive"""

implementation:
await asyncio . sleep ( self . _update_interval )
while not self . _closed :
    await self . update ( )
    await asyncio . sleep ( self . _update_interval )


signature:
def _iterate_namespace_models ( self , ** kwargs ) -> Iterable :
    """Return an iterator over the models to be converted to the namespace ."""

implementation: return tqdm ( self . _get_query ( self . namespace_model ) , total = self . _count_model ( self . namespace_model ) , ** kwargs )


signature:
def _get_default_namespace ( self ) -> Optional [ Namespace ] :
    """Get the reference BEL namespace if it exists ."""

implementation: return self . _get_query ( Namespace ) . filter ( Namespace . url == self . _get_namespace_url ( ) ) . one_or_none ( )


signature:
def _make_namespace ( self ) -> Namespace :
    """Make a namespace ."""

implementation:
namespace = Namespace ( name = self . _get_namespace_name ( ) , keyword = self . _get_namespace_keyword ( ) , url = self . _get_namespace_url ( ) , version = str ( time . asctime ( ) ) , )
self . session . add ( namespace )
entries = self . _get_namespace_entries ( namespace )
self . session . add_all ( entries )
t = time . time ( )
log . info ( 'committing models' )
self . session . commit ( )
log . info ( 'committed models in %.2f seconds' , time . time ( ) - t )
return namespace


signature:
def _get_old_entry_identifiers ( namespace : Namespace ) -> Set [ NamespaceEntry ] :
    """Convert a PyBEL generalized namespace entries to a set .
 Default to using the identifier , but can be overridden to use the name instead .
 > > > { term . identifier for term in namespace . entries }"""

implementation: return { term . identifier for term in namespace . entries }


signature:
def _update_namespace ( self , namespace : Namespace ) -> None :
    """Update an already - created namespace .
 Note : Only call this if namespace won ' t be none !"""

implementation:
old_entry_identifiers = self . _get_old_entry_identifiers ( namespace )
new_count = 0
skip_count = 0
for model in self . _iterate_namespace_models ( ) :
    if self . _get_identifier ( model ) in old_entry_identifiers :
        continue
    entry = self . _create_namespace_entry_from_model ( model , namespace = namespace )
    if entry is None or entry . name is None :
        skip_count += 1
        continue
    new_count += 1
    self . session . add ( entry )
t = time . time ( )
log . info ( 'got %d new entries. skipped %d entries missing names. committing models' , new_count , skip_count )
self . session . commit ( )
log . info ( 'committed models in %.2f seconds' , time . time ( ) - t )


signature:
def add_namespace_to_graph ( self , graph : BELGraph ) -> Namespace :
    """Add this manager ' s namespace to the graph ."""

implementation:
namespace = self . upload_bel_namespace ( )
graph . namespace_url [ namespace . keyword ] = namespace . url
# Add this manager as an annotation , too
self . _add_annotation_to_graph ( graph )
return namespace


signature:
def _add_annotation_to_graph ( self , graph : BELGraph ) -> None :
    """Add this manager as an annotation to the graph ."""

implementation:
if 'bio2bel' not in graph . annotation_list :
    graph . annotation_list [ 'bio2bel' ] = set ( )
graph . annotation_list [ 'bio2bel' ] . add ( self . module_name )


signature:
def upload_bel_namespace ( self , update : bool = False ) -> Namespace :
    """Upload the namespace to the PyBEL database .
 : param update : Should the namespace be updated first ?"""

implementation:
if not self . is_populated ( ) :
    self . populate ( )
namespace = self . _get_default_namespace ( )
if namespace is None :
    log . info ( 'making namespace for %s' , self . _get_namespace_name ( ) )
    return self . _make_namespace ( )
if update :
    self . _update_namespace ( namespace )
return namespace


signature:
def drop_bel_namespace ( self ) -> Optional [ Namespace ] :
    """Remove the default namespace if it exists ."""

implementation:
namespace = self . _get_default_namespace ( )
if namespace is not None :
    for entry in tqdm ( namespace . entries , desc = f'deleting entries in {self._get_namespace_name()}' ) :
        self . session . delete ( entry )
    self . session . delete ( namespace )
    log . info ( 'committing deletions' )
    self . session . commit ( )
    return namespace


signature:
def write_bel_namespace ( self , file : TextIO , use_names : bool = False ) -> None :
    """Write as a BEL namespace file ."""

implementation:
if not self . is_populated ( ) :
    self . populate ( )
if use_names and not self . has_names :
    raise ValueError
values = ( self . _get_namespace_name_to_encoding ( desc = 'writing names' ) if use_names else self . _get_namespace_identifier_to_encoding ( desc = 'writing identifiers' ) )
write_namespace ( namespace_name = self . _get_namespace_name ( ) , namespace_keyword = self . _get_namespace_keyword ( ) , namespace_query_url = self . identifiers_url , values = values , file = file , )


signature:
def write_bel_annotation ( self , file : TextIO ) -> None :
    """Write as a BEL annotation file ."""

implementation:
if not self . is_populated ( ) :
    self . populate ( )
values = self . _get_namespace_name_to_encoding ( desc = 'writing names' )
write_annotation ( keyword = self . _get_namespace_keyword ( ) , citation_name = self . _get_namespace_name ( ) , description = '' , values = values , file = file , )


signature:
def write_bel_namespace_mappings ( self , file : TextIO , ** kwargs ) -> None :
    """Write a BEL namespace mapping file ."""

implementation: json . dump ( self . _get_namespace_identifier_to_name ( ** kwargs ) , file , indent = 2 , sort_keys = True )


signature:
def write_directory ( self , directory : str ) -> bool :
    """Write a BEL namespace for identifiers , names , name hash , and mappings to the given directory ."""

implementation:
current_md5_hash = self . get_namespace_hash ( )
md5_hash_path = os . path . join ( directory , f'{self.module_name}.belns.md5' )
if not os . path . exists ( md5_hash_path ) :
    old_md5_hash = None
else :
    with open ( md5_hash_path ) as file :
        old_md5_hash = file . read ( ) . strip ( )
if old_md5_hash == current_md5_hash :
    return False
with open ( os . path . join ( directory , f'{self.module_name}.belns' ) , 'w' ) as file :
    self . write_bel_namespace ( file , use_names = False )
with open ( md5_hash_path , 'w' ) as file :
    print ( current_md5_hash , file = file )
if self . has_names :
    with open ( os . path . join ( directory , f'{self.module_name}-names.belns' ) , 'w' ) as file :
        self . write_bel_namespace ( file , use_names = True )
    with open ( os . path . join ( directory , f'{self.module_name}.belns.mapping' ) , 'w' ) as file :
        self . write_bel_namespace_mappings ( file , desc = 'writing mapping' )
return True


signature:
def get_namespace_hash ( self , hash_fn = hashlib . md5 ) -> str :
    """Get the namespace hash .
 Defaults to MD5."""

implementation:
m = hash_fn ( )
if self . has_names :
    items = self . _get_namespace_name_to_encoding ( desc = 'getting hash' ) . items ( )
else :
    items = self . _get_namespace_identifier_to_encoding ( desc = 'getting hash' ) . items ( )
for name , encoding in items :
    m . update ( f'{name}:{encoding}' . encode ( 'utf8' ) )
return m . hexdigest ( )


signature:
def get_long_description ( ) :
    """Get the long _ description from the README . rst file . Assume UTF - 8 encoding ."""

implementation:
with codecs . open ( os . path . join ( HERE , 'README.rst' ) , encoding = 'utf-8' ) as f :
    long_description = f . read ( )
return long_description


signature:
def dropbox_post_factory ( request ) :
    """receives a UUID via the request and returns either a fresh or an existing dropbox
 for it"""

implementation:
try :
    max_age = int ( request . registry . settings . get ( 'post_token_max_age_seconds' ) )
except Exception :
    max_age = 300
try :
    drop_id = parse_post_token ( token = request . matchdict [ 'token' ] , secret = request . registry . settings [ 'post_secret' ] , max_age = max_age )
except SignatureExpired :
    raise HTTPGone ( 'dropbox expired' )
except Exception : # don ' t be too specific on the reason for the error
    raise HTTPNotFound ( 'no such dropbox' )
dropbox = request . registry . settings [ 'dropbox_container' ] . get_dropbox ( drop_id )
if dropbox . status_int >= 20 :
    raise HTTPGone ( 'dropbox already in processing, no longer accepts data' )
return dropbox


signature:
def dropbox_factory ( request ) :
    """expects the id of an existing dropbox and returns its instance"""

implementation:
try :
    return request . registry . settings [ 'dropbox_container' ] . get_dropbox ( request . matchdict [ 'drop_id' ] )
except KeyError :
    raise HTTPNotFound ( 'no such dropbox' )


signature:
def is_equal ( a , b ) :
    """a constant time comparison implementation taken from
 http : / / codahale . com / a - lesson - in - timing - attacks / and
 Django ' s ` util ` module https : / / github . com / django / django / blob / master / django / utils / crypto . py # L82"""

implementation:
if len ( a ) != len ( b ) :
    return False
result = 0
for x , y in zip ( a , b ) :
    result |= ord ( x ) ^ ord ( y )
return result == 0


signature:
def dropbox_editor_factory ( request ) :
    """this factory also requires the editor token"""

implementation:
dropbox = dropbox_factory ( request )
if is_equal ( dropbox . editor_token , request . matchdict [ 'editor_token' ] . encode ( 'utf-8' ) ) :
    return dropbox
else :
    raise HTTPNotFound ( 'invalid editor token' )


signature:
def sanitize_filename ( filename ) :
    """preserve the file ending , but replace the name with a random token"""

implementation:
# TODO : fix broken splitext ( it reveals everything of the filename after the first ` . ` - doh ! )
token = generate_drop_id ( )
name , extension = splitext ( filename )
if extension :
    return '%s%s' % ( token , extension )
else :
    return token


signature:
def process ( self ) :
    """Calls the external cleanser scripts to ( optionally ) purge the meta data and then
 send the contents of the dropbox via email ."""

implementation:
if self . num_attachments > 0 :
    self . status = u'100 processor running'
    fs_dirty_archive = self . _create_backup ( )
    # calling _ process _ attachments has the side - effect of updating ` send _ attachments `
    self . _process_attachments ( )
    if self . status_int < 500 and not self . send_attachments :
        self . _create_archive ( )
if self . status_int >= 500 and self . status_int < 600 : # cleansing failed
# if configured , we need to move the uncleansed archive to
# the appropriate folder and notify the editors
    if 'dropbox_dirty_archive_url_format' in self . settings : # create _ archive
        shutil . move ( fs_dirty_archive , '%s/%s.zip.pgp' % ( self . container . fs_archive_dirty , self . drop_id ) )
        # update status
        # it ' s now considered ' successful - ish ' again
        self . status = '490 cleanser failure but notify success'
if self . status_int == 800 : # at least one attachment was not supported
# if configured , we need to move the uncleansed archive to
# the appropriate folder and notify the editors
    if 'dropbox_dirty_archive_url_format' in self . settings : # create _ archive
        shutil . move ( fs_dirty_archive , '%s/%s.zip.pgp' % ( self . container . fs_archive_dirty , self . drop_id ) )
if self . status_int < 500 or self . status_int == 800 :
    try :
        if self . _notify_editors ( ) > 0 :
            if self . status_int < 500 :
                self . status = '900 success'
        else :
            self . status = '605 smtp failure'
    except Exception :
        import traceback
        tb = traceback . format_exc ( )
        self . status = '610 smtp error (%s)' % tb
self . cleanup ( )
return self . status