<s> '''<STR_LIT>''' <EOL> from collections import defaultdict <EOL> from hedwig . core import UnaryPredicate , Rule , Example <EOL> from hedwig . core . settings import logger <EOL> from hedwig . stats . significance import is_redundant <EOL> from hedwig . stats . scorefunctions import interesting <EOL> class Learner : <EOL> '''<STR_LIT>''' <EOL> Similarity = '<STR_LIT>' <EOL> Improvement = '<STR_LIT>' <EOL> Default = '<STR_LIT:default>' <EOL> def __init__ ( self , kb , n = None , min_sup = <NUM_LIT:1> , sim = <NUM_LIT:1> , depth = <NUM_LIT:4> , target = None , <EOL> use_negations = False , optimal_subclass = False ) : <EOL> self . kb = kb <EOL> self . n = n <EOL> self . min_sup = min_sup <EOL> self . sim = sim <EOL> self . extending = Learner . Improvement <EOL> self . depth = depth <EOL> self . use_negations = use_negations <EOL> self . optimal_subclass = optimal_subclass <EOL> if kb . is_discrete_target ( ) : <EOL> self . target = list ( self . kb . class_values ) [ <NUM_LIT:0> ] if not target else target <EOL> else : <EOL> self . target = None <EOL> self . pruned_subclasses = self . _pruned_subclasses ( ) <EOL> self . pruned_superclasses_closure = self . _pruned_superclasses ( ) <EOL> self . implicit_roots = self . _implicit_roots ( ) <EOL> def _pruned_subclasses ( self ) : <EOL> min_sup = lambda pred : self . kb . n_members ( pred ) >= self . min_sup <EOL> pruned_subclasses = { } <EOL> for pred in self . kb . predicates : <EOL> subclasses = self . kb . get_subclasses ( pred ) <EOL> pruned_subclasses [ pred ] = filter ( min_sup , subclasses ) <EOL> return pruned_subclasses <EOL> def _pruned_superclasses ( self ) : <EOL> min_sup = lambda pred : self . kb . n_members ( pred ) >= self . min_sup <EOL> pruned_superclasses = { } <EOL> for pred in self . kb . predicates : <EOL> superclasses = self . kb . super_classes ( pred ) <EOL> pruned_superclasses [ pred ] = filter ( min_sup , superclasses ) <EOL> return pruned_superclasses <EOL> def _implicit_roots ( self ) : <EOL> implicit_roots = set ( ) <EOL> n_examples = self . kb . n_examples ( ) <EOL> for pred in self . kb . predicates : <EOL> if self . kb . n_members ( pred ) == n_examples : <EOL> implicit_roots . add ( pred ) <EOL> return implicit_roots <EOL> def get_subclasses ( self , pred ) : <EOL> return self . pruned_subclasses [ pred . label ] <EOL> def get_superclasses ( self , pred ) : <EOL> return self . pruned_superclasses_closure [ pred ] <EOL> def is_implicit_root ( self , pred ) : <EOL> return pred in self . implicit_roots <EOL> def induce ( self ) : <EOL> '''<STR_LIT>''' <EOL> root_pred = self . kb . get_root ( ) <EOL> rules = [ Rule ( self . kb , predicates = [ root_pred ] , target = self . target ) ] <EOL> rules = self . __induce_level ( rules ) <EOL> return filter ( interesting , rules ) <EOL> def __induce_level ( self , rules ) : <EOL> '''<STR_LIT>''' <EOL> while True : <EOL> old_score = self . group_score ( rules ) <EOL> new_rules = rules [ : ] <EOL> for i , rule in enumerate ( rules ) : <EOL> specializations = self . specialize ( rule ) <EOL> self . extend ( new_rules , specializations ) <EOL> rules = sorted ( new_rules , <EOL> key = lambda rule : rule . score , <EOL> reverse = True ) [ : self . n ] <EOL> new_score = self . group_score ( rules ) <EOL> logger . debug ( "<STR_LIT>" % ( old_score , new_score ) ) <EOL> if <NUM_LIT:1> - abs ( old_score / ( new_score + <NUM_LIT> ) ) < <NUM_LIT> : <EOL> break <EOL> return rules <EOL> def extend ( self , rules , specializations ) : <EOL> '''<STR_LIT>''' <EOL> if self . extending == Learner . Default : <EOL> return rules . extend ( specializations ) <EOL> elif self . extending == Learner . Improvement : <EOL> return self . extend_replace_worst ( rules , specializations ) <EOL> elif self . extending == Learner . Similarity : <EOL> return self . extend_with_similarity ( rules , specializations ) <EOL> def extend_with_similarity ( self , rules , specializations ) : <EOL> '''<STR_LIT>''' <EOL> for new_rule in specializations : <EOL> tmp_rules = rules [ : ] <EOL> for rule in tmp_rules : <EOL> sim = rule . similarity ( new_rule ) <EOL> if sim >= self . sim and len ( rules ) > <NUM_LIT:0.5> * self . n : <EOL> break <EOL> else : <EOL> rules . append ( new_rule ) <EOL> def extend_replace_worst ( self , rules , specializations ) : <EOL> '''<STR_LIT>''' <EOL> def is_similar ( new_rule ) : <EOL> for rule in rules [ : ] : <EOL> if rule . similarity ( new_rule ) == <NUM_LIT:1> : <EOL> return True <EOL> return False <EOL> improved = False <EOL> for new_rule in sorted ( specializations , key = lambda rule : rule . score ) : <EOL> worst = sorted ( rules , key = lambda rule : rule . score ) [ <NUM_LIT:0> ] <EOL> if len ( rules ) < self . n : <EOL> rules . append ( new_rule ) <EOL> improved = True <EOL> elif new_rule . score > worst . score and not is_similar ( new_rule ) : <EOL> self . _replace ( rules , worst , new_rule ) <EOL> improved = True <EOL> return improved <EOL> def _replace ( self , rules , worst , new_rule ) : <EOL> idx = rules . index ( worst ) <EOL> rules [ idx ] = new_rule <EOL> def specialize ( self , rule ) : <EOL> '''<STR_LIT>''' <EOL> is_unary = lambda p : isinstance ( p , UnaryPredicate ) <EOL> def specialize_optimal_subclass ( rule ) : <EOL> rules = [ ] <EOL> eligible_preds = rule . shared_var [ rule . latest_var ] <EOL> for pred in filter ( is_unary , eligible_preds ) : <EOL> for sub_class in self . get_subclasses ( pred ) : <EOL> logger . debug ( '<STR_LIT>' % sub_class ) <EOL> new_rule = rule . clone_swap_with_subclass ( pred , sub_class ) <EOL> if self . can_specialize ( new_rule ) : <EOL> rules . append ( new_rule ) <EOL> rules . extend ( specialize_optimal_subclass ( new_rule ) ) <EOL> return rules <EOL> logger . debug ( '<STR_LIT>' % rule ) <EOL> specializations = [ ] <EOL> eligible_preds = rule . shared_var [ rule . latest_var ] <EOL> if not self . optimal_subclass : <EOL> for pred in filter ( is_unary , eligible_preds ) : <EOL> logger . debug ( '<STR_LIT>' % pred . label ) <EOL> for sub_class in self . get_subclasses ( pred ) : <EOL> logger . debug ( '<STR_LIT>' % sub_class ) <EOL> new_rule = rule . clone_swap_with_subclass ( pred , sub_class ) <EOL> if self . can_specialize ( new_rule ) : <EOL> specializations . append ( new_rule ) <EOL> else : <EOL> specializations . extend ( specialize_optimal_subclass ( rule ) ) <EOL> if self . use_negations : <EOL> for pred in filter ( is_unary , eligible_preds ) : <EOL> logger . debug ( '<STR_LIT>' % pred . label ) <EOL> new_rule = rule . clone_negate ( pred ) <EOL> if self . can_specialize ( new_rule ) : <EOL> specializations . append ( new_rule ) <EOL> if not ( len ( eligible_preds ) == <NUM_LIT:1> and <EOL> ( eligible_preds [ <NUM_LIT:0> ] . label == self . kb . get_root ( ) . label or <EOL> self . is_implicit_root ( eligible_preds [ <NUM_LIT:0> ] . label ) ) ) : <EOL> supers = set ( ) <EOL> for pred in eligible_preds : <EOL> supers . update ( self . get_superclasses ( pred . label ) ) <EOL> supers . add ( pred ) <EOL> for lvl in sorted ( self . kb . levels . keys ( ) ) : <EOL> level = self . kb . levels [ lvl ] <EOL> diff = level . difference ( supers ) <EOL> if diff : <EOL> for pred in sorted ( list ( diff ) ) : <EOL> last_pred = rule . predicates [ - <NUM_LIT:1> ] <EOL> new_rule = rule . clone_append ( pred , <EOL> producer_pred = last_pred ) <EOL> if self . can_specialize ( new_rule ) and self . non_redundant ( rule , new_rule ) : <EOL> specializations . append ( new_rule ) <EOL> break <EOL> if isinstance ( rule . predicates [ - <NUM_LIT:1> ] , UnaryPredicate ) : <EOL> specializations . extend ( self . specialize_add_relation ( rule ) ) <EOL> logger . debug ( '<STR_LIT>' <EOL> % [ str ( rule ) for rule in specializations ] ) <EOL> return specializations <EOL> def specialize_add_relation ( self , rule ) : <EOL> '''<STR_LIT>''' <EOL> specializations = [ ] <EOL> for pred in self . kb . binary_predicates : <EOL> last_pred = rule . predicates [ - <NUM_LIT:1> ] <EOL> new_rule = rule . clone_append ( pred , producer_pred = last_pred , <EOL> bin = True ) <EOL> if self . can_specialize ( new_rule ) : <EOL> specializations . append ( new_rule ) <EOL> return specializations <EOL> def can_specialize ( self , rule ) : <EOL> '''<STR_LIT>''' <EOL> return rule . coverage >= self . min_sup and rule . size ( ) <= self . depth <EOL> def non_redundant ( self , rule , new_rule ) : <EOL> '''<STR_LIT>''' <EOL> if new_rule . score < rule . score : <EOL> return False <EOL> if rule . target_type == Example . Ranked : <EOL> return True <EOL> else : <EOL> return not is_redundant ( rule , new_rule ) <EOL> def group_score ( self , rules ) : <EOL> '''<STR_LIT>''' <EOL> return sum ( [ rule . score for rule in rules ] ) </s>
