#NEWLINE# def stream( self , new_data , rollover = None ) : #NEWLINE# #INDENT# import numpy as np #NEWLINE# newkeys = set ( new_data . keys ( ) ) #NEWLINE# oldkeys = set ( self . data . keys ( ) ) #NEWLINE# if ( newkeys != oldkeys ) : #NEWLINE# #INDENT# missing = ( oldkeys - newkeys ) #NEWLINE# extra = ( newkeys - oldkeys ) #NEWLINE# if ( missing and extra ) : #NEWLINE# #INDENT# raise ValueError ( ( 'Must stream updates to all existing columns (missing: %s, extra: %s)' % ( ', ' . join ( sorted ( missing ) ) , ', ' . join ( sorted ( extra ) ) ) ) ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if missing : #NEWLINE# #INDENT# raise ValueError ( ( 'Must stream updates to all existing columns (missing: %s)' % ', ' . join ( sorted ( missing ) ) ) ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# raise ValueError ( ( 'Must stream updates to all existing columns (extra: %s)' % ', ' . join ( sorted ( extra ) ) ) ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# lengths = set ( ) #NEWLINE# for x in new_data . values ( ) : #NEWLINE# #INDENT# if isinstance ( x , np . ndarray ) : #NEWLINE# #INDENT# if ( len ( x . shape ) != 1 ) : #NEWLINE# #INDENT# raise ValueError ( ( 'stream(...) only supports 1d sequences, got ndarray with size %r' % ( x . shape , ) ) ) #NEWLINE# #UNINDENT# lengths . add ( x . shape [ 0 ] ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# lengths . add ( len ( new_data ) ) #NEWLINE# #UNINDENT# #UNINDENT# if ( len ( lengths ) > 1 ) : #NEWLINE# #INDENT# raise ValueError ( 'All streaming column updates must be the same length' ) #NEWLINE# #UNINDENT# self . data . _stream ( self . document , self , new_data , rollover )