Mercurial > repos > matt-shirley > ncbi_epi_browse
comparison ncbi_connector.py @ 9:58917de44665 draft
Uploaded
author | matt-shirley |
---|---|
date | Mon, 30 Sep 2013 12:46:06 -0400 |
parents | |
children |
comparison
equal
deleted
inserted
replaced
8:3e7669c9ecbc | 9:58917de44665 |
---|---|
1 #!/usr/bin/env python | |
2 # Retrieves data from external data source applications and stores in a dataset file. | |
3 # Data source application parameters are temporarily stored in the dataset file. | |
4 import socket | |
5 import urllib | |
6 import sys | |
7 import os | |
8 import optparse | |
9 import xml.etree.ElementTree as ElementTree | |
10 from galaxy import eggs #eggs needs to be imported so that galaxy.util can find docutils egg... | |
11 from galaxy.util.json import from_json_string, to_json_string | |
12 from galaxy.util import get_charset_from_http_headers | |
13 import galaxy.model # need to import model before sniff to resolve a circular import dependency | |
14 from galaxy.datatypes import sniff | |
15 from galaxy.datatypes.registry import Registry | |
16 from galaxy.jobs import TOOL_PROVIDED_JOB_METADATA_FILE | |
17 | |
18 assert sys.version_info[:2] >= ( 2, 4 ) | |
19 | |
20 def stop_err( msg ): | |
21 sys.stderr.write( msg ) | |
22 sys.exit() | |
23 | |
24 GALAXY_PARAM_PREFIX = 'GALAXY' | |
25 GALAXY_ROOT_DIR = os.path.realpath( os.path.join( os.path.split( os.path.realpath( __file__ ) )[0], '..', '..' ) ) | |
26 GALAXY_DATATYPES_CONF_FILE = os.path.join( GALAXY_ROOT_DIR, 'datatypes_conf.xml' ) | |
27 | |
28 def load_input_parameters( filename, erase_file = True ): | |
29 datasource_params = {} | |
30 try: | |
31 json_params = from_json_string( open( filename, 'r' ).read() ) | |
32 datasource_params = json_params.get( 'param_dict' ) | |
33 except: | |
34 json_params = None | |
35 for line in open( filename, 'r' ): | |
36 try: | |
37 line = line.strip() | |
38 fields = line.split( '\t' ) | |
39 datasource_params[ fields[0] ] = fields[1] | |
40 except: | |
41 continue | |
42 if erase_file: | |
43 open( filename, 'w' ).close() #open file for writing, then close, removes params from file | |
44 return json_params, datasource_params | |
45 | |
46 def deconstruct_multi_filename( multi_filename ): | |
47 keys = ['primary', 'id', 'name', 'visible', 'file_type', 'dbkey'] | |
48 return ( dict( zip( keys, multi_filename.split('_') ) ) ) | |
49 | |
50 def construct_multi_filename( id, name, file_type, dbkey=None): | |
51 """ Implementation of *Number of Output datasets cannot be determined until tool run* from documentation_. | |
52 .. _documentation: http://wiki.galaxyproject.org/Admin/Tools/Multiple%20Output%20Files | |
53 """ | |
54 if dbkey: | |
55 filename = "%s_%s_%s_%s_%s_%s" % ( 'primary', id, name, 'visible', file_type, dbkey ) | |
56 else: | |
57 filename = "%s_%s_%s_%s_%s" % ( 'primary', id, name, 'visible', file_type ) | |
58 | |
59 return filename | |
60 | |
61 def xml_save_to_newfile_directory( xmlfile, directory, id, enhanced_handling=False, datatypes_registry=None ): | |
62 """ Open xmlfile, parse all URLs to fetch. Fetch each file, saving to ``directory``. | |
63 Save first file for last and return for ``page``. | |
64 | |
65 Schema | |
66 | |
67 :: | |
68 | |
69 <?xml version="1.0"?> | |
70 <!DOCTYPE downloads [ | |
71 <!ELEMENT downloads (download)> | |
72 <!ELEMENT download (resource,url,meta)> | |
73 <!ELEMENT resource (#PCDATA)> | |
74 <!ELEMENT url (#PCDATA)> | |
75 <!ELEMENT id (#PCDATA)> | |
76 <!ELEMENT meta (id,format,type,summary,feature,genome,technique,instrument,assay,sample,description,PMID)> | |
77 <!ELEMENT id (#PCDATA)> | |
78 <!ELEMENT format (#PCDATA)> | |
79 <!ELEMENT type (#PCDATA)> | |
80 <!ELEMENT summary (#PCDATA)> | |
81 <!ELEMENT feature (#PCDATA)> | |
82 <!ELEMENT genome (#PCDATA)> | |
83 <!ELEMENT technique (#PCDATA)> | |
84 <!ELEMENT instrument (#PCDATA)> | |
85 <!ELEMENT assay (#PCDATA)> | |
86 <!ELEMENT sample (#PCDATA)> | |
87 <!ELEMENT description (#PCDATA)> | |
88 <!ELEMENT PMID (#PCDATA)> | |
89 ]> | |
90 """ | |
91 root = ElementTree.fromstring(xmlfile.read()) | |
92 if root.tag != 'downloads': | |
93 stop_err( 'The remote data source application has not sent back a URL parameter in the request.' ) | |
94 # traverse xml schema to find URLs, names, and dbkeys | |
95 files_to_fetch = [] | |
96 complete = False | |
97 | |
98 for child in root: | |
99 if (child.tag == 'download') and (complete == True): | |
100 files_to_fetch.append( ( construct_multi_filename( id, name, file_type, dbkey ), URL ) ) | |
101 | |
102 for sub in child: | |
103 if sub.tag == 'url': | |
104 URL = sub.text | |
105 elif sub.tag == 'meta': | |
106 | |
107 for meta in sub: | |
108 if meta.tag == 'id': | |
109 name = meta.text | |
110 elif meta.tag == 'genome': | |
111 dbkey = meta.text | |
112 elif meta.tag == 'format': | |
113 file_type = meta.text | |
114 # hit the end of our schema | |
115 files_to_fetch.append( ( construct_multi_filename( id, name, file_type, dbkey ), URL ) ) | |
116 | |
117 if len(files_to_fetch) > 1: | |
118 for filename, URL in files_to_fetch[1:]: | |
119 try: | |
120 cur_filename = os.path.join( directory, filename ) | |
121 page = urllib.urlopen( URL ) | |
122 multi_dict = deconstruct_multi_filename( filename ) | |
123 cur_filename, is_multi_byte = sniff.stream_to_open_named_file( page, os.open( cur_filename, os.O_WRONLY | os.O_CREAT ), cur_filename, source_encoding=get_charset_from_http_headers( page.headers ) ) | |
124 if enhanced_handling: | |
125 ext = sniff.handle_uploaded_dataset_file( cur_filename, datatypes_registry, ext = multi_dict['file_type'], is_multi_byte = is_multi_byte ) | |
126 except Exception, e: | |
127 stop_err( 'Unable to fetch %s:\n%s' % ( URL, e ) ) | |
128 | |
129 # pass page back to main | |
130 return ( files_to_fetch[0][0], urllib.urlopen( files_to_fetch[0][1] ) ) | |
131 | |
132 def __main__(): | |
133 # Parse the command line options | |
134 usage = "Usage: ncbi_connector.py filename max_size [options]" | |
135 parser = optparse.OptionParser(usage = usage) | |
136 parser.add_option("-x", "--xml", | |
137 action="store_true", dest="xmlfile", help="filename defines external resources as xml") | |
138 parser.add_option("-i", "--id", type="string", | |
139 action="store", dest="id", help="output id") | |
140 parser.add_option("-p", "--path", type="string", | |
141 action="store", dest="newfilepath", help="new file path") | |
142 | |
143 (options, args) = parser.parse_args() | |
144 | |
145 filename = args[0] | |
146 try: | |
147 max_file_size = int( args[1] ) | |
148 except: | |
149 max_file_size = 0 | |
150 | |
151 job_params, params = load_input_parameters( filename ) | |
152 if job_params is None: #using an older tabular file | |
153 enhanced_handling = False | |
154 job_params = dict( param_dict = params ) | |
155 job_params[ 'output_data' ] = [ dict( out_data_name = 'output', | |
156 ext = 'data', | |
157 file_name = filename, | |
158 extra_files_path = None ) ] | |
159 job_params[ 'job_config' ] = dict( GALAXY_ROOT_DIR=GALAXY_ROOT_DIR, GALAXY_DATATYPES_CONF_FILE=GALAXY_DATATYPES_CONF_FILE, TOOL_PROVIDED_JOB_METADATA_FILE = TOOL_PROVIDED_JOB_METADATA_FILE ) | |
160 else: | |
161 enhanced_handling = True | |
162 json_file = open( job_params[ 'job_config' ][ 'TOOL_PROVIDED_JOB_METADATA_FILE' ], 'w' ) #specially named file for output junk to pass onto set metadata | |
163 | |
164 datatypes_registry = Registry() | |
165 datatypes_registry.load_datatypes( root_dir = job_params[ 'job_config' ][ 'GALAXY_ROOT_DIR' ], config = job_params[ 'job_config' ][ 'GALAXY_DATATYPES_CONF_FILE' ] ) | |
166 | |
167 URL = params.get( 'URL', None ) #using exactly URL indicates that only one dataset is being downloaded | |
168 URL_method = params.get( 'URL_method', None ) | |
169 | |
170 # The Python support for fetching resources from the web is layered. urllib uses the httplib | |
171 # library, which in turn uses the socket library. As of Python 2.3 you can specify how long | |
172 # a socket should wait for a response before timing out. By default the socket module has no | |
173 # timeout and can hang. Currently, the socket timeout is not exposed at the httplib or urllib2 | |
174 # levels. However, you can set the default timeout ( in seconds ) globally for all sockets by | |
175 # doing the following. | |
176 socket.setdefaulttimeout( 600 ) | |
177 | |
178 for data_dict in job_params[ 'output_data' ]: | |
179 cur_filename = data_dict.get( 'file_name', filename ) | |
180 cur_URL = params.get( '%s|%s|URL' % ( GALAXY_PARAM_PREFIX, data_dict[ 'out_data_name' ] ), URL ) | |
181 if not cur_URL: | |
182 open( cur_filename, 'w' ).write( "" ) | |
183 stop_err( 'The remote data source application has not sent back a URL parameter in the request.' ) | |
184 | |
185 # The following calls to urllib.urlopen() will use the above default timeout | |
186 try: | |
187 if not URL_method or URL_method == 'get': | |
188 page = urllib.urlopen( cur_URL ) | |
189 elif URL_method == 'post': | |
190 page = urllib.urlopen( cur_URL, urllib.urlencode( params ) ) | |
191 except Exception, e: | |
192 stop_err( 'The remote data source application may be off line, please try again later. Error: %s' % str( e ) ) | |
193 if max_file_size: | |
194 file_size = int( page.info().get( 'Content-Length', 0 ) ) | |
195 if file_size > max_file_size: | |
196 stop_err( 'The size of the data (%d bytes) you have requested exceeds the maximum allowed (%d bytes) on this server.' % ( file_size, max_file_size ) ) | |
197 # If xmlfile is provided, fetch files 2 through n and save to new_file_path, replace page with file 1 | |
198 if options.xmlfile: | |
199 multi_filename, page = xml_save_to_newfile_directory( page, options.newfilepath, options.id, enhanced_handling, datatypes_registry ) | |
200 multi_dict = deconstruct_multi_filename( multi_filename ) | |
201 | |
202 #do sniff stream for multi_byte | |
203 try: | |
204 cur_filename, is_multi_byte = sniff.stream_to_open_named_file( page, os.open( cur_filename, os.O_WRONLY | os.O_CREAT ), cur_filename, source_encoding=get_charset_from_http_headers( page.headers ) ) | |
205 except Exception, e: | |
206 stop_err( 'Unable to fetch %s:\n%s' % ( cur_URL, e ) ) | |
207 | |
208 #here import checks that upload tool performs | |
209 if enhanced_handling: | |
210 try: | |
211 ext = sniff.handle_uploaded_dataset_file( filename, datatypes_registry, ext = data_dict[ 'ext' ], is_multi_byte = is_multi_byte ) | |
212 except Exception, e: | |
213 stop_err( str( e ) ) | |
214 info = dict( type = 'dataset', | |
215 dataset_id = data_dict[ 'dataset_id' ], | |
216 ext = ext) | |
217 | |
218 json_file.write( "%s\n" % to_json_string( info ) ) | |
219 | |
220 if __name__ == "__main__": __main__() |