<s> import os <EOL> import rdflib <EOL> import shutil <EOL> import tempfile <EOL> import zipfile <EOL> from collections import Counter <EOL> from fnmatch import fnmatch <EOL> from rdflib . plugins . parsers . notation3 import BadSyntax <EOL> from urlparse import urlparse <EOL> class NIDMUpload : <EOL> def __init__ ( self , zip_path , tmp_dir = None , load = True ) : <EOL> self . path = zip_path <EOL> self . provn = None <EOL> self . ttl = None <EOL> self . ttl_relpath = '<STR_LIT>' <EOL> self . workdir = tempfile . mkdtemp ( ) if tmp_dir is None else tmp_dir <EOL> self . zip = None <EOL> self . raw_ttl = '<STR_LIT>' <EOL> self . valid_ttl = False <EOL> self . unzipped = False <EOL> self . contrasts = [ ] <EOL> self . statmaps = [ ] <EOL> if load : <EOL> self . get_statmaps ( ) <EOL> class ParseException ( Exception ) : <EOL> pass <EOL> class NoStatMapsException ( Exception ) : <EOL> pass <EOL> def parse_metafiles ( self , extract_ttl = False ) : <EOL> try : <EOL> self . zip = zipfile . ZipFile ( self . path ) <EOL> except Exception : <EOL> raise self . ParseException ( "<STR_LIT>" ) <EOL> metafiles = { } <EOL> for ext in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> metafiles [ ext ] = [ v for v in self . zip . infolist ( ) <EOL> if fnmatch ( v . filename , '<STR_LIT:*>' + ext ) <EOL> and not fnmatch ( v . filename , '<STR_LIT>' ) <EOL> ] <EOL> if len ( metafiles [ ext ] ) > <NUM_LIT:1> : <EOL> raise self . ParseException ( <EOL> "<STR_LIT>" . format ( ext ) ) <EOL> if not metafiles [ ext ] : <EOL> raise self . ParseException ( <EOL> "<STR_LIT>" . format ( ext ) ) <EOL> self . ttl = metafiles [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> self . provn = metafiles [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> self . raw_ttl = self . fix_spm12_ttl ( self . zip . read ( metafiles [ '<STR_LIT>' ] [ <NUM_LIT:0> ] ) ) <EOL> self . ttl_relpath = self . parse_ttl_relative_path ( metafiles [ '<STR_LIT>' ] [ <NUM_LIT:0> ] . filename ) <EOL> return self . raw_ttl if extract_ttl else True <EOL> def parse_contrasts ( self ) : <EOL> query = """<STR_LIT>""" <EOL> nidm_g = rdflib . Graph ( ) <EOL> try : <EOL> nidm_g . parse ( data = self . raw_ttl , format = '<STR_LIT>' ) <EOL> self . valid_ttl = True <EOL> except BadSyntax : <EOL> raise self . ParseException ( "<STR_LIT>" ) <EOL> c_results = nidm_g . query ( query ) <EOL> for row in c_results . bindings : <EOL> c_row = { } <EOL> for key , val in sorted ( row . items ( ) ) : <EOL> c_row [ str ( key ) ] = val . decode ( ) <EOL> self . contrasts . append ( c_row ) <EOL> self . contrasts = { v [ '<STR_LIT>' ] : v for v in self . contrasts } . values ( ) <EOL> return self . contrasts <EOL> def unpack_nidm_zip ( self ) : <EOL> if not self . ttl or not self . provn : <EOL> self . parse_metafiles ( ) <EOL> if not self . contrasts : <EOL> self . parse_contrasts ( ) <EOL> try : <EOL> self . zip . extractall ( path = self . workdir ) <EOL> self . unzipped = True <EOL> except Exception , e : <EOL> raise self . ParseException ( "<STR_LIT>" % e ) <EOL> def get_statmaps ( self ) : <EOL> if not self . unzipped : <EOL> self . unpack_nidm_zip ( ) <EOL> if not self . contrasts : <EOL> raise self . NoStatMapsException ( "<STR_LIT>" ) <EOL> incidences = Counter ( [ v [ '<STR_LIT>' ] for v in self . contrasts ] ) <EOL> for contrast in self . contrasts : <EOL> map = { '<STR_LIT:name>' : contrast [ '<STR_LIT>' ] , <EOL> '<STR_LIT:type>' : self . parse_statmap_type ( contrast [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT:file>' : self . validate_statmap_uri ( contrast ) } <EOL> if incidences [ contrast [ '<STR_LIT>' ] ] > <NUM_LIT:1> : <EOL> map [ '<STR_LIT:name>' ] = self . get_unique_mapname ( contrast ) <EOL> self . statmaps . append ( map ) <EOL> return self . statmaps <EOL> def validate_statmap_uri ( self , contrast ) : <EOL> rel_path = self . uri_to_path ( contrast [ '<STR_LIT>' ] ) <EOL> path = self . expand_path ( rel_path ) <EOL> if not os . path . exists ( path ) : <EOL> raise self . ParseException ( <EOL> "<STR_LIT>" . format ( <EOL> self . parse_statmap_type ( contrast [ '<STR_LIT>' ] ) ) ) <EOL> return path <EOL> def expand_path ( self , path ) : <EOL> for sep in [ '<STR_LIT:.>' , '<STR_LIT:/>' ] : <EOL> if path . startswith ( sep ) : <EOL> path = path . replace ( sep , self . ttl_relpath , <NUM_LIT:1> ) <EOL> return os . path . join ( self . workdir , path ) <EOL> def copy_to_dest ( self , dest ) : <EOL> source = os . path . join ( self . workdir , self . ttl_relpath ) <EOL> flist = [ v for v in os . listdir ( source ) if self . valid_path ( v ) ] <EOL> for dfile in flist : <EOL> sourcepath = os . path . join ( source , dfile ) <EOL> destpath = os . path . join ( dest , dfile ) <EOL> if not os . path . exists ( destpath ) : <EOL> shutil . move ( sourcepath , destpath ) <EOL> def cleanup ( self ) : <EOL> shutil . rmtree ( self . workdir ) <EOL> @ classmethod <EOL> def uri_to_path ( self , map_uri ) : <EOL> print map_uri <EOL> uri = urlparse ( map_uri ) <EOL> if not uri . path : <EOL> raise self . ParseException ( "<STR_LIT>" ) <EOL> return '<STR_LIT>' . format ( uri . netloc , uri . path ) <EOL> @ classmethod <EOL> def get_unique_mapname ( self , contrast ) : <EOL> type = self . parse_statmap_type ( contrast [ '<STR_LIT>' ] ) <EOL> return '<STR_LIT>' . format ( contrast [ '<STR_LIT>' ] , type ) <EOL> @ staticmethod <EOL> def parse_statmap_type ( stattype_url ) : <EOL> stato_dict = { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT:T>" , <EOL> "<STR_LIT>" : "<STR_LIT:F>" } <EOL> return stato_dict [ stattype_url ] <EOL> @ staticmethod <EOL> def parse_ttl_relative_path ( path ) : <EOL> relpath = os . path . dirname ( path ) <EOL> for sep in [ '<STR_LIT:.>' , '<STR_LIT:.>' , '<STR_LIT:/>' ] : <EOL> if relpath . startswith ( sep ) : <EOL> relpath = relpath . replace ( sep , '<STR_LIT>' , <NUM_LIT:1> ) <EOL> return relpath <EOL> @ staticmethod <EOL> def valid_path ( path ) : <EOL> if fnmatch ( path , '<STR_LIT>' ) : <EOL> return False <EOL> if path [ <NUM_LIT:0> ] is '<STR_LIT:.>' and path [ <NUM_LIT:1> ] is not ( '<STR_LIT:/>' or '<STR_LIT:.>' ) : <EOL> return False <EOL> return True <EOL> @ staticmethod <EOL> def print_sparql_results ( res ) : <EOL> for idx , row in enumerate ( res . bindings ) : <EOL> rowfmt = [ ] <EOL> print "<STR_LIT>" % idx <EOL> for key , val in sorted ( row . items ( ) ) : <EOL> rowfmt . append ( '<STR_LIT>' % ( key , val . decode ( ) ) ) <EOL> print '<STR_LIT:\n>' . join ( rowfmt ) <EOL> @ staticmethod <EOL> def fix_spm12_ttl ( ttl ) : <EOL> return ttl . replace ( "<STR_LIT>" , "<STR_LIT>" ) </s>
