<s> import re <EOL> from string import Template <EOL> import numpy <EOL> from pyparsing import Word , alphanums , Suppress , Optional , CharsNotIn , Group , nums , ZeroOrMore , OneOrMore , cppStyleComment , Literal , printables <EOL> from pgmpy . models import BayesianModel <EOL> from pgmpy . factors import TabularCPD <EOL> from pgmpy . extern . six . moves import map , range <EOL> class BIFReader ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , path = None , string = None ) : <EOL> """<STR_LIT>""" <EOL> if path : <EOL> with open ( path , '<STR_LIT:r>' ) as network : <EOL> self . network = network . read ( ) <EOL> elif string : <EOL> self . network = string <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if '<STR_LIT:">' in self . network : <EOL> self . network = self . network . replace ( '<STR_LIT:">' , '<STR_LIT:U+0020>' ) <EOL> if '<STR_LIT>' in self . network or '<STR_LIT>' in self . network : <EOL> self . network = cppStyleComment . suppress ( ) . transformString ( self . network ) <EOL> self . name_expr , self . state_expr , self . property_expr = self . get_variable_grammar ( ) <EOL> self . probability_expr , self . cpd_expr = self . get_probability_grammar ( ) <EOL> self . network_name = self . get_network_name ( ) <EOL> self . variable_names = self . get_variables ( ) <EOL> self . variable_states = self . get_states ( ) <EOL> self . variable_properties = self . get_property ( ) <EOL> self . variable_parents = self . get_parents ( ) <EOL> self . variable_cpds = self . get_cpd ( ) <EOL> self . variable_edges = self . get_edges ( ) <EOL> def get_variable_grammar ( self ) : <EOL> """<STR_LIT>""" <EOL> word_expr = Word ( alphanums + '<STR_LIT:_>' + '<STR_LIT:->' ) <EOL> word_expr2 = Word ( initChars = printables , excludeChars = [ '<STR_LIT:{>' , '<STR_LIT:}>' , '<STR_LIT:U+002C>' , '<STR_LIT:U+0020>' ] ) <EOL> name_expr = Suppress ( '<STR_LIT>' ) + word_expr + Suppress ( '<STR_LIT:{>' ) <EOL> state_expr = ZeroOrMore ( word_expr2 + Optional ( Suppress ( "<STR_LIT:U+002C>" ) ) ) <EOL> variable_state_expr = Suppress ( '<STR_LIT:type>' ) + Suppress ( word_expr ) + Suppress ( '<STR_LIT:[>' ) + Suppress ( Word ( nums ) ) + Suppress ( '<STR_LIT:]>' ) + Suppress ( '<STR_LIT:{>' ) + Group ( state_expr ) + Suppress ( '<STR_LIT:}>' ) + Suppress ( '<STR_LIT:;>' ) <EOL> property_expr = Suppress ( '<STR_LIT>' ) + CharsNotIn ( '<STR_LIT:;>' ) + Suppress ( '<STR_LIT:;>' ) <EOL> return name_expr , variable_state_expr , property_expr <EOL> def get_probability_grammar ( self ) : <EOL> """<STR_LIT>""" <EOL> word_expr = Word ( alphanums + '<STR_LIT:->' + '<STR_LIT:_>' ) + Suppress ( Optional ( "<STR_LIT:|>" ) ) + Suppress ( Optional ( "<STR_LIT:U+002C>" ) ) <EOL> word_expr2 = Word ( initChars = printables , excludeChars = [ '<STR_LIT:U+002C>' , '<STR_LIT:)>' , '<STR_LIT:U+0020>' , '<STR_LIT:(>' ] ) + Suppress ( Optional ( "<STR_LIT:U+002C>" ) ) <EOL> num_expr = Word ( nums + '<STR_LIT:->' + '<STR_LIT:+>' + '<STR_LIT:e>' + '<STR_LIT:E>' + '<STR_LIT:.>' ) + Suppress ( Optional ( "<STR_LIT:U+002C>" ) ) <EOL> probability_expr = Suppress ( '<STR_LIT>' ) + Suppress ( '<STR_LIT:(>' ) + OneOrMore ( word_expr ) + Suppress ( '<STR_LIT:)>' ) <EOL> optional_expr = Suppress ( '<STR_LIT:(>' ) + Suppress ( OneOrMore ( word_expr2 ) ) + Suppress ( '<STR_LIT:)>' ) <EOL> probab_attributes = optional_expr | Suppress ( '<STR_LIT>' ) <EOL> cpd_expr = probab_attributes + OneOrMore ( num_expr ) <EOL> return probability_expr , cpd_expr <EOL> def variable_block ( self ) : <EOL> start = re . finditer ( '<STR_LIT>' , self . network ) <EOL> for index in start : <EOL> end = self . network . find ( '<STR_LIT>' , index . start ( ) ) <EOL> yield self . network [ index . start ( ) : end ] <EOL> def probability_block ( self ) : <EOL> start = re . finditer ( '<STR_LIT>' , self . network ) <EOL> for index in start : <EOL> end = self . network . find ( '<STR_LIT>' , index . start ( ) ) <EOL> yield self . network [ index . start ( ) : end ] <EOL> def get_network_name ( self ) : <EOL> """<STR_LIT>""" <EOL> start = self . network . find ( '<STR_LIT>' ) <EOL> end = self . network . find ( '<STR_LIT>' , start ) <EOL> network_attribute = Suppress ( '<STR_LIT>' ) + Word ( alphanums + '<STR_LIT:_>' + '<STR_LIT:->' ) + '<STR_LIT:{>' <EOL> network_name = network_attribute . searchString ( self . network [ start : end ] ) [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> return network_name <EOL> def get_variables ( self ) : <EOL> """<STR_LIT>""" <EOL> variable_names = [ ] <EOL> for block in self . variable_block ( ) : <EOL> name = self . name_expr . searchString ( block ) [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> variable_names . append ( name ) <EOL> return variable_names <EOL> def get_states ( self ) : <EOL> """<STR_LIT>""" <EOL> variable_states = { } <EOL> for block in self . variable_block ( ) : <EOL> name = self . name_expr . searchString ( block ) [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> variable_states [ name ] = list ( self . state_expr . searchString ( block ) [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] ) <EOL> return variable_states <EOL> def get_property ( self ) : <EOL> """<STR_LIT>""" <EOL> variable_properties = { } <EOL> for block in self . variable_block ( ) : <EOL> name = self . name_expr . searchString ( block ) [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> properties = self . property_expr . searchString ( block ) <EOL> variable_properties [ name ] = [ y . strip ( ) for x in properties for y in x ] <EOL> return variable_properties <EOL> def get_parents ( self ) : <EOL> """<STR_LIT>""" <EOL> variable_parents = { } <EOL> for block in self . probability_block ( ) : <EOL> names = self . probability_expr . searchString ( block . split ( '<STR_LIT:\n>' ) [ <NUM_LIT:0> ] ) [ <NUM_LIT:0> ] <EOL> variable_parents [ names [ <NUM_LIT:0> ] ] = names [ <NUM_LIT:1> : ] <EOL> return variable_parents <EOL> def get_cpd ( self ) : <EOL> """<STR_LIT>""" <EOL> variable_cpds = { } <EOL> for block in self . probability_block ( ) : <EOL> name = self . probability_expr . searchString ( block ) [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> cpds = self . cpd_expr . searchString ( block ) <EOL> arr = [ float ( j ) for i in cpds for j in i ] <EOL> if '<STR_LIT>' in block : <EOL> arr = numpy . array ( arr ) <EOL> arr = arr . reshape ( ( len ( self . variable_states [ name ] ) , <EOL> arr . size // len ( self . variable_states [ name ] ) ) ) <EOL> else : <EOL> length = len ( self . variable_states [ name ] ) <EOL> reshape_arr = [ [ ] for i in range ( length ) ] <EOL> for i , val in enumerate ( arr ) : <EOL> reshape_arr [ i % length ] . append ( val ) <EOL> arr = reshape_arr <EOL> arr = numpy . array ( arr ) <EOL> variable_cpds [ name ] = arr <EOL> return variable_cpds <EOL> def get_edges ( self ) : <EOL> """<STR_LIT>""" <EOL> edges = [ [ value , key ] for key in self . variable_parents . keys ( ) <EOL> for value in self . variable_parents [ key ] ] <EOL> return edges <EOL> def get_model ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> model = BayesianModel ( self . variable_edges ) <EOL> model . name = self . network_name <EOL> model . add_nodes_from ( self . variable_names ) <EOL> tabular_cpds = [ ] <EOL> for var in sorted ( self . variable_cpds . keys ( ) ) : <EOL> values = self . variable_cpds [ var ] <EOL> cpd = TabularCPD ( var , len ( self . variable_states [ var ] ) , values , <EOL> evidence = self . variable_parents [ var ] , <EOL> evidence_card = [ len ( self . variable_states [ evidence_var ] ) <EOL> for evidence_var in self . variable_parents [ var ] ] ) <EOL> tabular_cpds . append ( cpd ) <EOL> model . add_cpds ( * tabular_cpds ) <EOL> for node , properties in self . variable_properties . items ( ) : <EOL> for prop in properties : <EOL> prop_name , prop_value = map ( lambda t : t . strip ( ) , prop . split ( '<STR_LIT:=>' ) ) <EOL> model . node [ node ] [ prop_name ] = prop_value <EOL> return model <EOL> except AttributeError : <EOL> raise AttributeError ( '<STR_LIT>' ) <EOL> class BIFWriter ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , model ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( model , BayesianModel ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> self . model = model <EOL> if not self . model . name : <EOL> self . network_name = '<STR_LIT>' <EOL> else : <EOL> self . network_name = self . model . name <EOL> self . variable_states = self . get_states ( ) <EOL> self . property_tag = self . get_properties ( ) <EOL> self . variable_parents = self . get_parents ( ) <EOL> self . tables = self . get_cpds ( ) <EOL> def BIF_templates ( self ) : <EOL> """<STR_LIT>""" <EOL> network_template = Template ( '<STR_LIT>' ) <EOL> variable_template = Template ( """<STR_LIT>""" ) <EOL> property_template = Template ( '<STR_LIT>' ) <EOL> probability_template = Template ( """<STR_LIT>""" ) <EOL> return network_template , variable_template , property_template , probability_template <EOL> def __str__ ( self ) : <EOL> """<STR_LIT>""" <EOL> network_template , variable_template , property_template , probability_template = self . BIF_templates ( ) <EOL> network = '<STR_LIT>' <EOL> network += network_template . substitute ( name = self . network_name ) <EOL> variables = self . model . nodes ( ) <EOL> for var in sorted ( variables ) : <EOL> no_of_states = str ( len ( self . variable_states [ var ] ) ) <EOL> states = '<STR_LIT:U+002CU+0020>' . join ( self . variable_states [ var ] ) <EOL> if not self . property_tag [ var ] : <EOL> properties = '<STR_LIT>' <EOL> else : <EOL> properties = '<STR_LIT>' <EOL> for prop_val in self . property_tag [ var ] : <EOL> properties += property_template . substitute ( prop = prop_val ) <EOL> network += variable_template . substitute ( name = var , no_of_states = no_of_states , <EOL> states = states , properties = properties ) <EOL> for var in sorted ( variables ) : <EOL> if not self . variable_parents [ var ] : <EOL> parents = '<STR_LIT>' <EOL> seprator = '<STR_LIT>' <EOL> else : <EOL> parents = '<STR_LIT:U+002CU+0020>' . join ( self . variable_parents [ var ] ) <EOL> seprator = '<STR_LIT>' <EOL> cpd = '<STR_LIT:U+002CU+0020>' . join ( map ( str , self . tables [ var ] ) ) <EOL> network += probability_template . substitute ( variable_ = var , seprator_ = seprator , <EOL> parents = parents , values = cpd ) <EOL> return network <EOL> def get_variables ( self ) : <EOL> """<STR_LIT>""" <EOL> variables = self . model . nodes ( ) <EOL> return variables <EOL> def get_states ( self ) : <EOL> """<STR_LIT>""" <EOL> variable_states = { } <EOL> cpds = self . model . get_cpds ( ) <EOL> for cpd in cpds : <EOL> variable = cpd . variable <EOL> variable_states [ variable ] = [ ] <EOL> for state in range ( cpd . get_cardinality ( [ variable ] ) [ variable ] ) : <EOL> variable_states [ variable ] . append ( str ( variable ) + '<STR_LIT:_>' + str ( state ) ) <EOL> return variable_states <EOL> def get_properties ( self ) : <EOL> """<STR_LIT>""" <EOL> variables = self . model . nodes ( ) <EOL> property_tag = { } <EOL> for variable in sorted ( variables ) : <EOL> properties = self . model . node [ variable ] <EOL> property_tag [ variable ] = [ ] <EOL> for prop , val in properties . items ( ) : <EOL> property_tag [ variable ] . append ( str ( prop ) + "<STR_LIT>" + str ( val ) ) <EOL> return property_tag <EOL> def get_parents ( self ) : <EOL> """<STR_LIT>""" <EOL> cpds = self . model . get_cpds ( ) <EOL> cpds . sort ( key = lambda x : x . variable ) <EOL> variable_parents = { } <EOL> for cpd in cpds : <EOL> variable_parents [ cpd . variable ] = [ ] <EOL> for parent in sorted ( cpd . variables [ : <NUM_LIT:0> : - <NUM_LIT:1> ] ) : <EOL> variable_parents [ cpd . variable ] . append ( parent ) <EOL> return variable_parents <EOL> def get_cpds ( self ) : <EOL> """<STR_LIT>""" <EOL> cpds = self . model . get_cpds ( ) <EOL> tables = { } <EOL> for cpd in cpds : <EOL> tables [ cpd . variable ] = cpd . values . ravel ( ) <EOL> return tables <EOL> def write_bif ( self , filename ) : <EOL> """<STR_LIT>""" <EOL> writer = self . __str__ ( ) <EOL> with open ( filename , '<STR_LIT:w>' ) as fout : <EOL> fout . write ( writer ) </s>
