Project

General

Profile

1
#!/usr/bin/env python
2
# Scrubs the taxonlabels in VegBIEN using TNRS.
3
# Runs continuously until no new rows are added after max_pause.
4

    
5
import os.path
6
import StringIO
7
import sys
8
import time
9

    
10
sys.path.append(os.path.dirname(__file__)+"/../lib")
11

    
12
import csvs
13
import dates
14
import opts
15
import profiling
16
import sql
17
import sql_gen
18
import sql_io
19
import streams
20
import strings
21
import tnrs
22

    
23
# Config
24
pause = 2*60*60 # sec; = 2 hr
25
max_pause = 9*60*60 # sec; = 9 hr; must be >= max partition import time (1.5 hr)
26
assert pause <= max_pause
27

    
28
tnrs_input = sql_gen.Table('tnrs_input_name')
29
tnrs_data = sql_gen.Table('tnrs')
30
time_col_name = 'Time_submitted'
31

    
32
def main():
33
    # Input
34
    env_names = []
35
    db_config = opts.get_env_vars(sql.db_config_names, None, env_names)
36
    verbosity = float(opts.get_env_var('verbosity', 3, env_names))
37
    wait = opts.env_flag('wait', False, env_names)
38
    if not 'engine' in db_config: raise SystemExit('Usage: '
39
        +opts.env_usage(env_names)+' '+sys.argv[0]+' 2>>log')
40
    
41
    def log(msg, level=1):
42
        '''Higher level -> more verbose'''
43
        if level <= verbosity:
44
            sys.stderr.write(strings.to_raw_str(msg.rstrip('\n')+'\n'))
45
    
46
    # Connect to DB
47
    db = sql.connect(db_config, log_debug=log)
48
    
49
    tnrs_profiler = profiling.ItersProfiler(iter_text='name')
50
    
51
    # Iterate over unscrubbed verbatim taxonlabels
52
    total_pause = 0
53
    while True:
54
        # Fetch next set
55
        cur = sql.select(db, tnrs_input, limit=tnrs.max_names, cacheable=False)
56
        this_ct = cur.rowcount
57
        log('Processing '+str(this_ct)+' taxonlabels')
58
        if this_ct == 0:
59
            if not wait: break
60
            log('Waited '+str(total_pause)+' sec total')
61
            total_pause += pause
62
            if total_pause > max_pause: break
63
            log('Waiting '+str(pause)+' sec...')
64
            time.sleep(pause) # wait for more rows
65
            continue # try again
66
        # otherwise, rows found
67
        total_pause = 0
68
        names = list(sql.values(cur))
69
        
70
        # Run TNRS
71
        log('Making TNRS request')
72
        now_str = str(dates.now())
73
        tnrs_profiler.start()
74
        try: stream = tnrs.repeated_tnrs_request(names)
75
        finally:
76
            tnrs_profiler.stop(iter_ct=this_ct)
77
            log('Cumulatively: '+tnrs_profiler.msg())
78
        
79
        log('Storing TNRS response data')
80
        reader, header = csvs.reader_and_header(stream)
81
        header.insert(0, time_col_name)
82
        reader = csvs.ColInsertFilter(reader, lambda row, row_num: now_str)
83
        sql_io.append_csv(db, tnrs_data, reader, header)
84

    
85
main()
(66-66/72)