<s> from __future__ import division <EOL> from future . utils import viewitems , viewkeys , viewvalues <EOL> from collections import defaultdict <EOL> import itertools <EOL> import functools <EOL> from . import blocking , predicates , core , index , backport <EOL> import numpy <EOL> import logging <EOL> import random <EOL> logger = logging . getLogger ( __name__ ) <EOL> def findUncertainPairs ( field_distances , classifier , bias = <NUM_LIT:0.5> ) : <EOL> """<STR_LIT>""" <EOL> probability = classifier . predict_proba ( field_distances ) [ : , - <NUM_LIT:1> ] <EOL> p_max = ( <NUM_LIT:1> - bias ) <EOL> logger . info ( p_max ) <EOL> informativity = numpy . copy ( probability ) <EOL> informativity [ probability < p_max ] /= p_max <EOL> informativity [ probability >= p_max ] = ( <NUM_LIT:1> - probability [ probability >= p_max ] ) / ( <NUM_LIT:1> - p_max ) <EOL> return numpy . argmax ( informativity ) <EOL> class ActiveLearning ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , candidates , data_model , num_processes ) : <EOL> self . candidates = candidates <EOL> pool = backport . Pool ( num_processes ) <EOL> self . field_distances = numpy . concatenate ( <EOL> pool . map ( data_model . distances , <EOL> chunker ( candidates , <NUM_LIT:100> ) , <EOL> <NUM_LIT:2> ) ) <EOL> pool . terminate ( ) <EOL> def uncertainPairs ( self , classifier , dupe_proportion ) : <EOL> uncertain_index = findUncertainPairs ( self . field_distances , <EOL> classifier , <EOL> dupe_proportion ) <EOL> self . field_distances = numpy . delete ( self . field_distances , <EOL> uncertain_index , axis = <NUM_LIT:0> ) <EOL> uncertain_pairs = [ self . candidates . pop ( uncertain_index ) ] <EOL> return uncertain_pairs <EOL> class BlockLearner ( object ) : <EOL> def learn ( self , matches , max_comparisons , recall ) : <EOL> '''<STR_LIT>''' <EOL> if len ( self . blocker . predicates ) <= <NUM_LIT> : <EOL> compound_length = <NUM_LIT:3> <EOL> else : <EOL> compound_length = <NUM_LIT:2> <EOL> self . blocker . indexAll ( { i : record <EOL> for i , record <EOL> in enumerate ( self . unroll ( matches ) ) } ) <EOL> dupe_cover = cover ( self . blocker , matches , compound_length ) <EOL> comparison_count = self . comparisons ( self . total_cover , compound_length ) <EOL> dupe_cover = { pred : pairs <EOL> for pred , pairs <EOL> in viewitems ( dupe_cover ) <EOL> if comparison_count [ pred ] < max_comparisons } <EOL> if not dupe_cover : <EOL> raise ValueError ( NO_PREDICATES_ERROR ) <EOL> uncoverable_dupes = set ( matches ) - set . union ( * viewvalues ( dupe_cover ) ) <EOL> epsilon = int ( ( <NUM_LIT:1.0> - recall ) * len ( matches ) ) <EOL> if len ( uncoverable_dupes ) > epsilon : <EOL> logger . warning ( OUT_OF_PREDICATES_WARNING ) <EOL> logger . debug ( uncoverable_dupes ) <EOL> epsilon = <NUM_LIT:0> <EOL> else : <EOL> epsilon -= len ( uncoverable_dupes ) <EOL> chvatal_set = greedy ( dupe_cover . copy ( ) , comparison_count , epsilon ) <EOL> dupe_cover = { pred : dupe_cover [ pred ] for pred in chvatal_set } <EOL> final_predicates = tuple ( dominating ( dupe_cover ) ) <EOL> logger . info ( '<STR_LIT>' ) <EOL> for predicate in final_predicates : <EOL> logger . info ( predicate ) <EOL> return final_predicates <EOL> def comparisons ( self , cover , compound_length ) : <EOL> CP = predicates . CompoundPredicate <EOL> block_index = { } <EOL> for predicate , blocks in viewitems ( cover ) : <EOL> block_index [ predicate ] = { } <EOL> for block_id , blocks in viewitems ( blocks ) : <EOL> for record_id in self . _blocks ( blocks ) : <EOL> block_index [ predicate ] . setdefault ( record_id , <EOL> set ( ) ) . add ( block_id ) <EOL> compounder = self . Compounder ( cover , block_index ) <EOL> comparison_count = { } <EOL> simple_predicates = sorted ( cover , key = str ) <EOL> for i in range ( <NUM_LIT:2> , compound_length + <NUM_LIT:1> ) : <EOL> for combo in itertools . combinations ( simple_predicates , i ) : <EOL> comparison_count [ CP ( combo ) ] = self . estimate ( compounder ( combo ) ) <EOL> for pred in simple_predicates : <EOL> comparison_count [ pred ] = self . estimate ( viewvalues ( cover [ pred ] ) ) <EOL> return comparison_count <EOL> class Compounder ( object ) : <EOL> def __init__ ( self , cover , block_index ) : <EOL> self . cover = cover <EOL> self . block_index = block_index <EOL> self . _cached_predicate = None <EOL> self . _cached_blocks = None <EOL> def __call__ ( self , compound_predicate ) : <EOL> a , b = compound_predicate [ : - <NUM_LIT:1> ] , compound_predicate [ - <NUM_LIT:1> ] <EOL> if len ( a ) > <NUM_LIT:1> : <EOL> if a == self . _cached_predicate : <EOL> a_blocks = self . _cached_blocks <EOL> else : <EOL> a_blocks = self . _cached_blocks = list ( self ( a ) ) <EOL> self . _cached_predicate = a <EOL> else : <EOL> a_blocks = viewvalues ( self . cover [ a [ <NUM_LIT:0> ] ] ) <EOL> return self . overlap ( a_blocks , b ) <EOL> class DedupeCompounder ( Compounder ) : <EOL> def overlap ( self , a_blocks , b ) : <EOL> b_index = self . block_index [ b ] <EOL> b_index_get = b_index . get <EOL> cover_b = self . cover [ b ] <EOL> null_set = set ( ) <EOL> for x_ids in a_blocks : <EOL> seen_y = set ( ) <EOL> for record_id in x_ids : <EOL> b_blocks = b_index_get ( record_id , null_set ) <EOL> for y in b_blocks : <EOL> if y not in seen_y : <EOL> yield x_ids & cover_b [ y ] <EOL> seen_y |= b_blocks <EOL> class DedupeBlockLearner ( BlockLearner ) : <EOL> Compounder = DedupeCompounder <EOL> def __init__ ( self , predicates , sampled_records ) : <EOL> blocker = blocking . Blocker ( predicates ) <EOL> blocker . indexAll ( sampled_records ) <EOL> self . total_cover = self . coveredRecords ( blocker , sampled_records ) <EOL> self . multiplier = sampled_records . original_length / len ( sampled_records ) <EOL> self . blocker = blocking . Blocker ( predicates ) <EOL> @ staticmethod <EOL> def unroll ( matches ) : <EOL> return set ( ) . union ( * matches ) <EOL> @ staticmethod <EOL> def _blocks ( blocks ) : <EOL> return blocks <EOL> @ staticmethod <EOL> def coveredRecords ( blocker , records ) : <EOL> CP = predicates . CompoundPredicate <EOL> cover = { } <EOL> for predicate in blocker . predicates : <EOL> cover [ predicate ] = { } <EOL> for id , record in viewitems ( records ) : <EOL> blocks = predicate ( record ) <EOL> for block in blocks : <EOL> cover [ predicate ] . setdefault ( block , set ( ) ) . add ( id ) <EOL> return cover <EOL> def estimate ( self , blocks ) : <EOL> lengths = numpy . fromiter ( ( len ( ids ) for ids in blocks ) , int ) <EOL> return numpy . sum ( ( lengths * self . multiplier ) * <EOL> ( lengths * self . multiplier - <NUM_LIT:1> ) / <EOL> <NUM_LIT:2> ) <EOL> class RecordLinkCompounder ( Compounder ) : <EOL> def overlap ( self , a_blocks , b ) : <EOL> b_index = self . block_index [ b ] <EOL> b_index_get = b_index . get <EOL> cover_b = self . cover [ b ] <EOL> null_set = set ( ) <EOL> for first , second in a_blocks : <EOL> seen_y = set ( ) <EOL> for record_id in first : <EOL> b_blocks = b_index_get ( record_id , null_set ) <EOL> for y in b_blocks : <EOL> if y not in seen_y : <EOL> yield first & cover_b [ y ] [ <NUM_LIT:0> ] , second & cover_b [ y ] [ <NUM_LIT:1> ] <EOL> seen_y |= b_blocks <EOL> class RecordLinkBlockLearner ( BlockLearner ) : <EOL> Compounder = RecordLinkCompounder <EOL> def __init__ ( self , predicates , sampled_records_1 , sampled_records_2 ) : <EOL> blocker = blocking . Blocker ( predicates ) <EOL> blocker . indexAll ( sampled_records_2 ) <EOL> self . total_cover = self . coveredRecords ( blocker , <EOL> sampled_records_1 , <EOL> sampled_records_2 ) <EOL> self . multiplier_1 = sampled_records_1 . original_length / len ( sampled_records_1 ) <EOL> self . multiplier_2 = sampled_records_2 . original_length / len ( sampled_records_2 ) <EOL> self . blocker = blocking . Blocker ( predicates ) <EOL> @ staticmethod <EOL> def unroll ( matches ) : <EOL> return { record_2 for _ , record_2 in matches } <EOL> @ staticmethod <EOL> def _blocks ( blocks ) : <EOL> return blocks [ <NUM_LIT:0> ] <EOL> @ staticmethod <EOL> def coveredRecords ( blocker , records_1 , records_2 ) : <EOL> CP = predicates . CompoundPredicate <EOL> cover = { } <EOL> for predicate in blocker . predicates : <EOL> cover [ predicate ] = { } <EOL> for id , record in viewitems ( records_2 ) : <EOL> blocks = predicate ( record ) <EOL> for block in blocks : <EOL> cover [ predicate ] . setdefault ( block , ( set ( ) , set ( ) ) ) [ <NUM_LIT:1> ] . add ( id ) <EOL> current_blocks = set ( cover [ predicate ] ) <EOL> for id , record in viewitems ( records_1 ) : <EOL> blocks = set ( predicate ( record ) ) <EOL> for block in blocks & current_blocks : <EOL> cover [ predicate ] [ block ] [ <NUM_LIT:0> ] . add ( id ) <EOL> return cover <EOL> def estimate ( self , blocks ) : <EOL> A , B = core . iunzip ( blocks , <NUM_LIT:2> ) <EOL> lengths_A = numpy . fromiter ( ( len ( ids ) for ids in A ) , float ) <EOL> lengths_B = numpy . fromiter ( ( len ( ids ) for ids in B ) , float ) <EOL> return numpy . sum ( ( lengths_A * self . multiplier_1 ) * <EOL> ( lengths_B * self . multiplier_2 ) ) <EOL> def greedy ( dupe_cover , comparison_count , epsilon ) : <EOL> uncovered_dupes = set . union ( * dupe_cover . values ( ) ) <EOL> final_predicates = set ( ) <EOL> while len ( uncovered_dupes ) > epsilon and dupe_cover : <EOL> cost = lambda p : comparison_count [ p ] / len ( dupe_cover [ p ] ) <EOL> best_predicate = min ( dupe_cover , key = cost ) <EOL> final_predicates . add ( best_predicate ) <EOL> covered = dupe_cover . pop ( best_predicate ) <EOL> uncovered_dupes = uncovered_dupes - covered <EOL> remaining_cover ( dupe_cover , covered ) <EOL> logger . debug ( best_predicate ) <EOL> logger . debug ( '<STR_LIT>' , <EOL> { '<STR_LIT>' : len ( uncovered_dupes ) } ) <EOL> return final_predicates <EOL> def dominating ( dupe_cover ) : <EOL> uncovered_dupes = set . union ( * dupe_cover . values ( ) ) <EOL> final_predicates = set ( ) <EOL> while uncovered_dupes : <EOL> score = lambda p : len ( dupe_cover [ p ] ) <EOL> best_predicate = max ( dupe_cover , key = score ) <EOL> final_predicates . add ( best_predicate ) <EOL> covered = dupe_cover . pop ( best_predicate ) <EOL> uncovered_dupes = uncovered_dupes - covered <EOL> remaining_cover ( dupe_cover , covered ) <EOL> return final_predicates <EOL> def cover ( blocker , pairs , compound_length ) : <EOL> cover = coveredPairs ( blocker . predicates , pairs ) <EOL> cover = compound ( cover , compound_length ) <EOL> remaining_cover ( cover ) <EOL> return cover <EOL> def coveredPairs ( predicates , pairs ) : <EOL> cover = { } <EOL> pairs = sorted ( pairs ) <EOL> for predicate in predicates : <EOL> rec_1 = None <EOL> for pair in pairs : <EOL> record_1 , record_2 = pair <EOL> if record_1 != rec_1 : <EOL> blocks_1 = set ( predicate ( record_1 ) ) <EOL> rec_1 = record_1 <EOL> if blocks_1 : <EOL> blocks_2 = predicate ( record_2 ) <EOL> field_preds = blocks_1 & set ( blocks_2 ) <EOL> if field_preds : <EOL> cover . setdefault ( predicate , set ( ) ) . add ( pair ) <EOL> return cover <EOL> def compound ( cover , compound_length ) : <EOL> simple_predicates = sorted ( cover , key = str ) <EOL> CP = predicates . CompoundPredicate <EOL> for i in range ( <NUM_LIT:2> , compound_length + <NUM_LIT:1> ) : <EOL> compound_predicates = itertools . combinations ( simple_predicates , i ) <EOL> for compound_predicate in compound_predicates : <EOL> a , b = compound_predicate [ : - <NUM_LIT:1> ] , compound_predicate [ - <NUM_LIT:1> ] <EOL> if len ( a ) == <NUM_LIT:1> : <EOL> a = a [ <NUM_LIT:0> ] <EOL> cover [ CP ( compound_predicate ) ] = cover [ a ] & cover [ b ] <EOL> return cover <EOL> def chunker ( seq , size ) : <EOL> return ( seq [ pos : pos + size ] for pos in range ( <NUM_LIT:0> , len ( seq ) , size ) ) <EOL> def remaining_cover ( coverage , covered = set ( ) ) : <EOL> null_covers = [ ] <EOL> for predicate , uncovered in viewitems ( coverage ) : <EOL> uncovered -= covered <EOL> if not uncovered : <EOL> null_covers . append ( predicate ) <EOL> for predicate in null_covers : <EOL> del coverage [ predicate ] <EOL> OUT_OF_PREDICATES_WARNING = "<STR_LIT>" <EOL> NO_PREDICATES_ERROR = "<STR_LIT>" </s>
