<s> from __future__ import print_function <EOL> import argparse <EOL> import random <EOL> import tempfile <EOL> import time <EOL> import sys <EOL> from collections import Counter <EOL> import numpy <EOL> import pylab <EOL> import crosscat . tests . enumerate_utils as eu <EOL> import crosscat . tests . plot_utils as pu <EOL> import crosscat . utils . sample_utils as su <EOL> import crosscat . utils . data_utils as du <EOL> import crosscat . cython_code . State as State <EOL> def get_next_seed ( rng , max_val = <NUM_LIT> ) : <EOL> return rng . randint ( max_val ) <EOL> def run_test ( seed , n = <NUM_LIT:1000> , d_type = '<STR_LIT>' , observed = False ) : <EOL> if d_type == '<STR_LIT>' : <EOL> run_test_continuous ( n , observed , seed ) <EOL> elif d_type == '<STR_LIT>' : <EOL> run_test_multinomial ( n , observed , seed ) <EOL> def generate_multinomial_data ( seed , n_cols , n_rows , n_views ) : <EOL> rng = numpy . random . RandomState ( seed ) <EOL> cols_to_views = [ <NUM_LIT:0> for _ in range ( n_cols ) ] <EOL> rows_in_views_to_cols = [ ] <EOL> for view in range ( n_views ) : <EOL> partition = eu . CRP ( n_rows , <NUM_LIT> , get_next_seed ( rng ) ) <EOL> random . shuffle ( partition , rng . uniform ) <EOL> rows_in_views_to_cols . append ( partition ) <EOL> data = numpy . zeros ( ( n_rows , n_cols ) , dtype = float ) <EOL> for col in range ( n_cols ) : <EOL> view = cols_to_views [ col ] <EOL> for row in range ( n_rows ) : <EOL> cluster = rows_in_views_to_cols [ view ] [ row ] <EOL> data [ row , col ] = cluster <EOL> T = data . tolist ( ) <EOL> M_r = du . gen_M_r_from_T ( T ) <EOL> M_c = du . gen_M_c_from_T ( T ) <EOL> T , M_c = du . convert_columns_to_multinomial ( T , M_c , list ( range ( n_cols ) ) ) <EOL> return T , M_r , M_c <EOL> def run_test_continuous ( n , observed , seed ) : <EOL> rng = numpy . random . RandomState ( seed ) <EOL> n_rows = <NUM_LIT> <EOL> n_cols = <NUM_LIT> <EOL> if observed : <EOL> query_row = <NUM_LIT:10> <EOL> else : <EOL> query_row = n_rows <EOL> query_column = <NUM_LIT:1> <EOL> Q = [ ( query_row , query_column ) ] <EOL> T , M_r , M_c = du . gen_factorial_data_objects ( get_next_seed ( rng ) , <NUM_LIT:2> , <NUM_LIT:2> , n_rows , <NUM_LIT:1> ) <EOL> state = State . p_State ( M_c , T ) <EOL> T_array = numpy . array ( T ) <EOL> X_L = state . get_X_L ( ) <EOL> X_D = state . get_X_D ( ) <EOL> Y = [ ] <EOL> gns = lambda : get_next_seed ( rng ) <EOL> samples = su . simple_predictive_sample ( M_c , X_L , X_D , Y , Q , gns , n = n ) <EOL> X_array = numpy . sort ( numpy . array ( samples ) ) <EOL> std_X = numpy . std ( X_array ) <EOL> mean_X = numpy . mean ( X_array ) <EOL> X_filter_low = numpy . nonzero ( X_array < mean_X - <NUM_LIT> * std_X ) [ <NUM_LIT:0> ] <EOL> X_filter_high = numpy . nonzero ( X_array > mean_X + <NUM_LIT> * std_X ) [ <NUM_LIT:0> ] <EOL> X_filter = numpy . hstack ( ( X_filter_low , X_filter_high ) ) <EOL> X_array = numpy . delete ( X_array , X_filter ) <EOL> X_array = numpy . sort ( X_array ) <EOL> X = X_array . tolist ( ) <EOL> Qs = [ ] ; <EOL> for x in X : <EOL> Qtmp = ( query_row , query_column , x ) <EOL> Qs . append ( Qtmp ) <EOL> densities = numpy . exp ( su . simple_predictive_probability ( M_c , X_L , X_D , Y , Qs ) ) <EOL> area_density = <NUM_LIT:0> ; <EOL> for i in range ( len ( X ) - <NUM_LIT:1> ) : <EOL> area_density += ( X [ i + <NUM_LIT:1> ] - X [ i ] ) * ( densities [ i + <NUM_LIT:1> ] + densities [ i ] ) / <NUM_LIT> <EOL> print ( "<STR_LIT>" + str ( area_density ) ) <EOL> print ( "<STR_LIT>" ) <EOL> pylab . figure ( facecolor = '<STR_LIT>' ) <EOL> pdf , bins , patches = pylab . hist ( X , <NUM_LIT:100> , normed = <NUM_LIT:1> , histtype = '<STR_LIT>' , label = '<STR_LIT>' , alpha = <NUM_LIT> , color = [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> pylab . scatter ( X , densities , c = "<STR_LIT>" , label = "<STR_LIT>" , edgecolor = '<STR_LIT:none>' ) <EOL> pylab . legend ( loc = '<STR_LIT>' , fontsize = '<STR_LIT>' ) <EOL> pylab . xlabel ( '<STR_LIT:value>' ) <EOL> pylab . ylabel ( '<STR_LIT>' ) <EOL> pylab . title ( '<STR_LIT>' ) <EOL> pylab . show ( ) <EOL> fd , fig_filename = tempfile . mkstemp ( prefix = '<STR_LIT>' , <EOL> suffix = '<STR_LIT>' , dir = '<STR_LIT:.>' ) <EOL> pylab . savefig ( fig_filename ) <EOL> def run_test_multinomial ( n , observed , seed ) : <EOL> rng = numpy . random . RandomState ( seed ) <EOL> n_rows = <NUM_LIT> <EOL> n_cols = <NUM_LIT> <EOL> if observed : <EOL> query_row = <NUM_LIT:10> <EOL> else : <EOL> query_row = n_rows <EOL> query_column = <NUM_LIT:1> <EOL> Q = [ ( query_row , query_column ) ] <EOL> T , M_r , M_c = generate_multinomial_data ( get_next_seed ( rng ) , <NUM_LIT:2> , n_rows , <NUM_LIT:1> ) <EOL> state = State . p_State ( M_c , T ) <EOL> X_L = state . get_X_L ( ) <EOL> X_D = state . get_X_D ( ) <EOL> Y = [ ] <EOL> gns = lambda : get_next_seed ( rng ) <EOL> samples = su . simple_predictive_sample ( M_c , X_L , X_D , Y , Q , gns , n = n ) <EOL> X_array = numpy . sort ( numpy . array ( samples ) ) <EOL> X = numpy . unique ( X_array ) <EOL> X = X . tolist ( ) <EOL> Qs = [ ] ; <EOL> for x in X : <EOL> Qtmp = ( query_row , query_column , x ) <EOL> Qs . append ( Qtmp ) <EOL> densities = numpy . exp ( su . simple_predictive_probability ( M_c , X_L , X_D , Y , Qs ) ) <EOL> print ( "<STR_LIT>" % ( numpy . sum ( densities ) ) ) <EOL> pylab . clf ( ) <EOL> mbins = numpy . unique ( X_array ) <EOL> mbins = numpy . append ( mbins , max ( mbins ) + <NUM_LIT:1> ) <EOL> pdf , bins = numpy . histogram ( X_array , mbins ) <EOL> pdf = pdf / float ( numpy . sum ( pdf ) ) <EOL> pylab . bar ( mbins [ <NUM_LIT:0> : - <NUM_LIT:1> ] , pdf , label = "<STR_LIT>" , alpha = <NUM_LIT> ) <EOL> pylab . scatter ( X , densities , c = "<STR_LIT>" , label = "<STR_LIT>" , edgecolor = '<STR_LIT:none>' ) <EOL> pylab . legend ( loc = '<STR_LIT>' , fontsize = '<STR_LIT>' ) <EOL> pylab . xlabel ( '<STR_LIT:value>' ) <EOL> pylab . ylabel ( '<STR_LIT>' ) <EOL> pylab . title ( '<STR_LIT>' ) <EOL> pylab . show ( ) <EOL> fd , fig_filename = tempfile . mkstemp ( prefix = '<STR_LIT>' , <EOL> suffix = '<STR_LIT>' , dir = '<STR_LIT:.>' ) <EOL> pylab . savefig ( fig_filename ) <EOL> rng = numpy . random . RandomState ( ) <EOL> run_test ( seed = get_next_seed ( rng ) , n = <NUM_LIT> , d_type = '<STR_LIT>' , observed = False ) <EOL> run_test ( seed = get_next_seed ( rng ) , n = <NUM_LIT> , d_type = '<STR_LIT>' , observed = True ) <EOL> run_test ( seed = get_next_seed ( rng ) , n = <NUM_LIT> , d_type = '<STR_LIT>' , observed = False ) <EOL> run_test ( seed = get_next_seed ( rng ) , n = <NUM_LIT> , d_type = '<STR_LIT>' , observed = True ) </s>
