<s> import os <EOL> import re <EOL> import json <EOL> from collections import OrderedDict <EOL> import numpy as np <EOL> from scipy . io import loadmat , savemat <EOL> from ... lib . array_utils import GrowingLILMatrix <EOL> FEATURE_DIRECTORY = '<STR_LIT>' <EOL> FEATURES = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> FEATURES_INDEX = { '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT>' : <NUM_LIT:2> , '<STR_LIT>' : <NUM_LIT:3> , <EOL> '<STR_LIT>' : <NUM_LIT:4> } <EOL> def _string_to_object_view ( s ) : <EOL> return tuple ( [ int ( i ) for i in s . split ( ) ] ) <EOL> def _string_to_features ( s ) : <EOL> return [ float ( i ) for i in s . split ( ) ] <EOL> class WrongNumberOfFeatures ( Exception ) : <EOL> pass <EOL> class Frame ( object ) : <EOL> def __init__ ( self , filename , label , timestamp , object_views ) : <EOL> if len ( object_views ) == <NUM_LIT:4> : <EOL> object_views = ( None , object_views [ <NUM_LIT:0> ] , object_views [ <NUM_LIT:1> ] , <EOL> object_views [ <NUM_LIT:2> ] , object_views [ <NUM_LIT:3> ] ) <EOL> self . label = label <EOL> self . timestamp = timestamp <EOL> self . views = object_views <EOL> self . filename = filename <EOL> def get_full_path ( self , path ) : <EOL> return os . path . join ( path , FEATURE_DIRECTORY , self . filename ) <EOL> def read_features ( self , path ) : <EOL> with open ( self . get_full_path ( path ) , '<STR_LIT:r>' ) as f : <EOL> lines = [ l for l in f . readlines ( ) [ : : - <NUM_LIT:1> ] if not l . isspace ( ) ] <EOL> lines . reverse ( ) <EOL> if len ( lines ) != len ( self . views ) * len ( FEATURES ) : <EOL> raise WrongNumberOfFeatures ( ) <EOL> return [ [ _string_to_features ( lines . pop ( ) ) for f in FEATURES ] <EOL> for o in self . views ] <EOL> def to_dict ( self ) : <EOL> return { '<STR_LIT>' : self . views , <EOL> '<STR_LIT:label>' : self . label , <EOL> '<STR_LIT>' : self . timestamp , <EOL> '<STR_LIT:filename>' : self . filename } <EOL> @ classmethod <EOL> def from_line ( cls , line ) : <EOL> blocks = line . split ( '<STR_LIT:|>' ) <EOL> filename = blocks [ <NUM_LIT:0> ] [ : - <NUM_LIT:1> ] <EOL> views = [ _string_to_object_view ( s ) for s in blocks [ <NUM_LIT:1> : ] ] <EOL> label , timestamp = Frame . _parse_filename ( filename ) <EOL> return Frame ( filename , label , timestamp , views ) <EOL> @ classmethod <EOL> def from_dict ( cls , d ) : <EOL> return Frame ( d [ '<STR_LIT:filename>' ] , d [ '<STR_LIT:label>' ] , <EOL> d [ '<STR_LIT>' ] , d [ '<STR_LIT>' ] ) <EOL> @ classmethod <EOL> def _parse_filename ( cls , name ) : <EOL> m = re . search ( r"<STR_LIT>" , name ) <EOL> if m : <EOL> return ( int ( m . group ( '<STR_LIT:label>' ) ) , float ( m . group ( '<STR_LIT:time>' ) ) ) <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" <EOL> % name ) <EOL> class ObjectDB ( object ) : <EOL> def __init__ ( self , object_names = None ) : <EOL> self . frames = [ ] <EOL> self . histos = [ GrowingLILMatrix ( ) for dummy in FEATURES ] <EOL> self . histos_by_frame = [ None for dummy in FEATURES ] <EOL> self . object_names = object_names <EOL> def add_frame ( self , frame ) : <EOL> self . frames . append ( frame ) <EOL> def add_feats ( self , feats ) : <EOL> for feat , histo in zip ( feats , self . histos ) : <EOL> histo . add_row ( feat ) <EOL> def save ( self , path , name ) : <EOL> meta_file = os . path . join ( path , name + '<STR_LIT>' ) <EOL> data_file = os . path . join ( path , name + '<STR_LIT>' ) <EOL> d = OrderedDict ( ) <EOL> if self . object_names is not None : <EOL> d [ '<STR_LIT>' ] = self . object_names <EOL> d [ '<STR_LIT>' ] = name + '<STR_LIT>' <EOL> d [ '<STR_LIT>' ] = [ frame . to_dict ( ) for frame in self . frames ] <EOL> with open ( meta_file , '<STR_LIT:w>' ) as f : <EOL> json . dump ( d , f , indent = <NUM_LIT:2> ) <EOL> matrices = dict ( zip ( FEATURES , self . histos ) ) <EOL> matrices . update ( zip ( map ( '<STR_LIT>' . __add__ , FEATURES ) , <EOL> self . histos_by_frame ) ) <EOL> savemat ( data_file , matrices ) <EOL> def get_histos_matrix ( self , histo_type ) : <EOL> return self . histos [ FEATURES_INDEX [ histo_type ] ] <EOL> def get_histos_matrix_by_frame ( self , histo_type ) : <EOL> return self . histos_by_frame [ FEATURES_INDEX [ histo_type ] ] <EOL> def _compute_histos_matrix_by_frame ( self , histo_type ) : <EOL> """<STR_LIT>""" <EOL> idx_fst = np . cumsum ( [ len ( f . views ) for f in self . frames ] ) <EOL> idx_fst = np . roll ( idx_fst , <NUM_LIT:1> ) <EOL> idx_fst [ <NUM_LIT:0> ] = <NUM_LIT:0> <EOL> orig_histos = self . get_histos_matrix ( histo_type ) . tocsc ( ) <EOL> histos = orig_histos [ idx_fst , : ] . tolil ( ) <EOL> for i , ( j , f ) in zip ( idx_fst , enumerate ( self . frames ) ) : <EOL> if len ( f . views ) > <NUM_LIT:1> : <EOL> new_histo = orig_histos [ i : ( i + len ( f . views ) ) , : ] . mean ( axis = <NUM_LIT:0> ) <EOL> histos [ j , : ] = new_histo <EOL> self . histos_by_frame [ FEATURES_INDEX [ histo_type ] ] = histos . tocsc ( ) <EOL> @ classmethod <EOL> def build_from ( cls , path , object_names = None , verbosity = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> db = ObjectDB ( object_names = object_names ) <EOL> with open ( path , '<STR_LIT:r>' ) as f : <EOL> frames = [ Frame . from_line ( l ) for l in f . readlines ( ) <EOL> if not l . isspace ( ) ] <EOL> skipped = <NUM_LIT:0> <EOL> for f in frames : <EOL> try : <EOL> for obj_feat in f . read_features ( os . path . dirname ( path ) ) : <EOL> db . add_feats ( obj_feat ) <EOL> db . add_frame ( f ) <EOL> except IOError as e : <EOL> skipped += <NUM_LIT:1> <EOL> if verbosity is '<STR_LIT>' : <EOL> print ( "<STR_LIT>" % ( f . filename , e . strerror ) ) <EOL> except WrongNumberOfFeatures : <EOL> skipped += <NUM_LIT:1> <EOL> if verbosity is '<STR_LIT>' : <EOL> print ( "<STR_LIT>" <EOL> % f . filename ) <EOL> if skipped > <NUM_LIT:0> and verbosity is not '<STR_LIT>' : <EOL> print ( "<STR_LIT>" % skipped ) <EOL> for feat in FEATURES : <EOL> if verbosity is not '<STR_LIT>' : <EOL> print ( "<STR_LIT>" % feat ) <EOL> db . _compute_histos_matrix_by_frame ( feat ) <EOL> return db <EOL> @ classmethod <EOL> def load ( cls , path ) : <EOL> with open ( path , '<STR_LIT:r>' ) as f : <EOL> meta = json . load ( f ) <EOL> object_names = meta . get ( '<STR_LIT>' , None ) <EOL> db = ObjectDB ( object_names = object_names ) <EOL> for f in meta [ '<STR_LIT>' ] : <EOL> db . add_frame ( Frame . from_dict ( f ) ) <EOL> if '<STR_LIT>' in meta : <EOL> data = loadmat ( os . path . join ( os . path . dirname ( path ) , <EOL> meta [ '<STR_LIT>' ] ) ) <EOL> db . histos = [ data [ feat ] for feat in FEATURES ] <EOL> db . histos_by_frame = [ data [ '<STR_LIT>' + feat ] <EOL> for feat in FEATURES ] <EOL> return db </s>
