<s> """<STR_LIT>""" <EOL> import unittest <EOL> from windml . preprocessing . nrel_repair import NRELRepair <EOL> from windml . datasets . aemo import AEMO <EOL> from windml . datasets . nrel import NREL <EOL> from windml . preprocessing . missing_data_finder import MissingDataFinder <EOL> from windml . preprocessing . linear_interpolation import LinearInterpolation <EOL> from windml . preprocessing . mar_destroyer import MARDestroyer <EOL> from windml . preprocessing . nmar_destroyer import NMARDestroyer <EOL> from windml . preprocessing . marthres_destroyer import MARThresDestroyer <EOL> from windml . preprocessing . topologic_interpolation import TopologicInterpolation <EOL> from windml . preprocessing . forward_copy import ForwardCopy <EOL> from windml . preprocessing . backward_copy import BackwardCopy <EOL> from windml . preprocessing . mreg_interpolation import MRegInterpolation <EOL> from sklearn . neighbors import KNeighborsRegressor <EOL> class TestPreprocessing ( unittest . TestCase ) : <EOL> def test_nrel_repair ( self ) : <EOL> ds = NREL ( ) <EOL> target = ds . get_turbine ( NREL . park_id [ '<STR_LIT>' ] , <NUM_LIT> ) <EOL> measurements = target . get_measurements ( ) [ : <NUM_LIT> ] <EOL> measurements = NRELRepair ( ) . repair ( measurements ) <EOL> assert ( NRELRepair ( ) . validate ( measurements ) ) <EOL> def test_mreg_interpolation_multi ( self ) : <EOL> park_id = NREL . park_id [ '<STR_LIT>' ] <EOL> windpark = NREL ( ) . get_windpark ( park_id , <NUM_LIT:3> , <NUM_LIT> ) <EOL> target = windpark . get_target ( ) <EOL> timestep = <NUM_LIT> <EOL> measurements = target . get_measurements ( ) [ <NUM_LIT> : <NUM_LIT> ] <EOL> damaged , indices = MARDestroyer ( ) . destroy ( measurements , percentage = <NUM_LIT> ) <EOL> before_misses = MissingDataFinder ( ) . find ( damaged , timestep ) <EOL> neighbors = windpark . get_turbines ( ) [ : - <NUM_LIT:1> ] <EOL> count_neighbors = len ( neighbors ) <EOL> reg = '<STR_LIT>' <EOL> regargs = { '<STR_LIT:n>' : <NUM_LIT:10> , '<STR_LIT>' : '<STR_LIT>' } <EOL> processed = <NUM_LIT:0> <EOL> missed = { k : count_neighbors for k in indices } <EOL> exclude = [ ] <EOL> damaged_nseries = [ ] <EOL> for neighbor in neighbors : <EOL> nseries = neighbor . get_measurements ( ) [ <NUM_LIT> : <NUM_LIT> ] <EOL> damaged , indices = MARDestroyer ( ) . destroy ( nseries , percentage = <NUM_LIT> , exclude = exclude ) <EOL> for index in indices : <EOL> if ( index not in missed . keys ( ) ) : <EOL> missed [ index ] = count_neighbors <EOL> missed [ index ] -= <NUM_LIT:1> <EOL> if ( missed [ index ] == <NUM_LIT:1> ) : <EOL> exclude . append ( index ) <EOL> damaged_nseries . append ( damaged ) <EOL> t_hat = MRegInterpolation ( ) . interpolate ( damaged , timestep = timestep , neighbor_series = damaged_nseries , reg = reg , regargs = regargs ) <EOL> after_misses = MissingDataFinder ( ) . find ( t_hat , timestep ) <EOL> assert ( len ( after_misses ) < <NUM_LIT:1> ) <EOL> def test_mreg_interpolation ( self ) : <EOL> park_id = NREL . park_id [ '<STR_LIT>' ] <EOL> windpark = NREL ( ) . get_windpark ( park_id , <NUM_LIT:3> , <NUM_LIT> ) <EOL> target = windpark . get_target ( ) <EOL> timestep = <NUM_LIT> <EOL> measurements = target . get_measurements ( ) [ <NUM_LIT> : <NUM_LIT> ] <EOL> damaged , indices = MARDestroyer ( ) . destroy ( measurements , percentage = <NUM_LIT> ) <EOL> before_misses = MissingDataFinder ( ) . find ( damaged , timestep ) <EOL> neighbors = windpark . get_turbines ( ) [ : - <NUM_LIT:1> ] <EOL> reg = '<STR_LIT>' <EOL> regargs = { '<STR_LIT:n>' : <NUM_LIT:10> , '<STR_LIT>' : '<STR_LIT>' } <EOL> nseries = [ t . get_measurements ( ) [ <NUM_LIT> : <NUM_LIT> ] for t in neighbors ] <EOL> t_hat = MRegInterpolation ( ) . interpolate ( damaged , timestep = timestep , neighbor_series = nseries , reg = reg , regargs = regargs ) <EOL> after_misses = MissingDataFinder ( ) . find ( t_hat , timestep ) <EOL> assert ( len ( after_misses ) < <NUM_LIT:1> ) <EOL> def test_linear_interpolation ( self ) : <EOL> park_id = NREL . park_id [ '<STR_LIT>' ] <EOL> windpark = NREL ( ) . get_windpark ( park_id , <NUM_LIT:10> , <NUM_LIT> ) <EOL> target = windpark . get_target ( ) <EOL> timestep = <NUM_LIT> <EOL> measurements = target . get_measurements ( ) [ <NUM_LIT> : <NUM_LIT> ] <EOL> damaged , indices = MARDestroyer ( ) . destroy ( measurements , percentage = <NUM_LIT> ) <EOL> before_misses = MissingDataFinder ( ) . find ( damaged , timestep ) <EOL> t_hat = LinearInterpolation ( ) . interpolate ( damaged , timestep = timestep ) <EOL> after_misses = MissingDataFinder ( ) . find ( t_hat , timestep ) <EOL> assert ( measurements . shape [ <NUM_LIT:0> ] == t_hat . shape [ <NUM_LIT:0> ] ) <EOL> assert ( len ( after_misses ) < <NUM_LIT:1> ) <EOL> def test_forward_copy_interpolation ( self ) : <EOL> park_id = NREL . park_id [ '<STR_LIT>' ] <EOL> windpark = NREL ( ) . get_windpark ( park_id , <NUM_LIT:10> , <NUM_LIT> ) <EOL> target = windpark . get_target ( ) <EOL> timestep = <NUM_LIT> <EOL> measurements = target . get_measurements ( ) [ <NUM_LIT> : <NUM_LIT> ] <EOL> damaged , indices = MARDestroyer ( ) . destroy ( measurements , percentage = <NUM_LIT> ) <EOL> before_misses = MissingDataFinder ( ) . find ( damaged , timestep ) <EOL> t_hat = ForwardCopy ( ) . interpolate ( measurements , timestep = timestep ) <EOL> after_misses = MissingDataFinder ( ) . find ( t_hat , timestep ) <EOL> assert ( measurements . shape [ <NUM_LIT:0> ] == t_hat . shape [ <NUM_LIT:0> ] ) <EOL> assert ( len ( after_misses ) < <NUM_LIT:1> ) <EOL> def test_backward_copy_interpolation ( self ) : <EOL> park_id = NREL . park_id [ '<STR_LIT>' ] <EOL> windpark = NREL ( ) . get_windpark ( park_id , <NUM_LIT:10> , <NUM_LIT> ) <EOL> target = windpark . get_target ( ) <EOL> timestep = <NUM_LIT> <EOL> measurements = target . get_measurements ( ) [ <NUM_LIT> : <NUM_LIT> ] <EOL> damaged , indices = MARDestroyer ( ) . destroy ( measurements , percentage = <NUM_LIT> ) <EOL> before_misses = MissingDataFinder ( ) . find ( damaged , timestep ) <EOL> t_hat = BackwardCopy ( ) . interpolate ( measurements , timestep = timestep ) <EOL> after_misses = MissingDataFinder ( ) . find ( t_hat , timestep ) <EOL> assert ( measurements . shape [ <NUM_LIT:0> ] == t_hat . shape [ <NUM_LIT:0> ] ) <EOL> assert ( len ( after_misses ) < <NUM_LIT:1> ) <EOL> def test_topological_interpolation ( self ) : <EOL> park_id = NREL . park_id [ '<STR_LIT>' ] <EOL> windpark = NREL ( ) . get_windpark ( park_id , <NUM_LIT:10> , <NUM_LIT> ) <EOL> target = windpark . get_target ( ) <EOL> timestep = <NUM_LIT> <EOL> measurements = target . get_measurements ( ) [ <NUM_LIT> : <NUM_LIT> ] <EOL> damaged , indices = NMARDestroyer ( ) . destroy ( measurements , percentage = <NUM_LIT> , min_length = <NUM_LIT:10> , max_length = <NUM_LIT:100> ) <EOL> tloc = ( target . longitude , target . latitude ) <EOL> neighbors = windpark . get_turbines ( ) [ : - <NUM_LIT:1> ] <EOL> nseries = [ t . get_measurements ( ) [ <NUM_LIT> : <NUM_LIT> ] for t in neighbors ] <EOL> nlocs = [ ( t . longitude , t . latitude ) for t in neighbors ] <EOL> t_hat = TopologicInterpolation ( ) . interpolate ( damaged , method = "<STR_LIT>" , timestep = timestep , location = tloc , neighbor_series = nseries , neighbor_locations = nlocs ) <EOL> misses = MissingDataFinder ( ) . find ( t_hat , timestep ) <EOL> assert ( measurements . shape [ <NUM_LIT:0> ] == t_hat . shape [ <NUM_LIT:0> ] ) <EOL> assert ( len ( misses ) < <NUM_LIT:1> ) <EOL> def test_mar_destroyer ( self ) : <EOL> turbine = NREL ( ) . get_turbine ( NREL . park_id [ '<STR_LIT>' ] , <NUM_LIT> ) <EOL> timeseries = turbine . get_measurements ( ) [ : <NUM_LIT:1000> ] <EOL> damaged , indices = MARDestroyer ( ) . destroy ( timeseries , percentage = <NUM_LIT> ) <EOL> misses = MissingDataFinder ( ) . find ( damaged , <NUM_LIT> ) <EOL> assert ( len ( misses ) > <NUM_LIT:0> ) <EOL> def test_marthres_destroyer ( self ) : <EOL> turbine = NREL ( ) . get_turbine ( NREL . park_id [ '<STR_LIT>' ] , <NUM_LIT> ) <EOL> timeseries = turbine . get_measurements ( ) [ : <NUM_LIT:1000> ] <EOL> damaged , indices = MARThresDestroyer ( ) . destroy ( timeseries , percentage = <NUM_LIT> , lower_bound = <NUM_LIT:0> , upper_bound = <NUM_LIT:20> ) <EOL> misses = MissingDataFinder ( ) . find ( damaged , <NUM_LIT> ) <EOL> assert ( len ( misses ) > <NUM_LIT:0> ) <EOL> def test_nmar_destroyer ( self ) : <EOL> turbine = NREL ( ) . get_turbine ( NREL . park_id [ '<STR_LIT>' ] , <NUM_LIT> ) <EOL> timeseries = turbine . get_measurements ( ) [ : <NUM_LIT:1000> ] <EOL> damaged , indices = NMARDestroyer ( ) . destroy ( timeseries , percentage = <NUM_LIT> , min_length = <NUM_LIT:10> , max_length = <NUM_LIT:50> ) <EOL> misses = MissingDataFinder ( ) . find ( damaged , <NUM_LIT> ) <EOL> assert ( len ( misses ) > <NUM_LIT:0> ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
