Changeset 2691

Timestamp:

01/16/12 14:59:45 (4 months ago)

Author:

jussi

Message:

very much unfinished business moving to memory based sqlite for workers

Location:

simo/branches/memory_worker

Files:

• dev/cfg/buildout.cfg (modified) (1 diff)
• src/runner2.py (modified) (6 diffs)

simo/branches/memory_worker/dev/cfg/buildout.cfg

@@ -111 +111 @@
 [geos]
 recipe = hexagonit.recipe.cmmi
-url = http://download.osgeo.org/geos/geos-3.3.1.tar.bz2
+url = http://download.osgeo.org/geos/geos-3.3.2.tar.bz2
 
 [proj4]

simo/branches/memory_worker/src/runner2.py

@@ -87 +87 @@
         db_type, host, port, user, pw, name, schema = self.db_info
         if db_type.upper() == 'POSTGRESQL':
-            db = runner.PostgreSQL(db_name='config', db_host=host, 
-                                   db_port=int(port), user=user, pw=pw, 
-                                   wipe=False, spatial=False, 
+            db = runner.PostgreSQL(db_name='config', db_host=host,
+                                   db_port=int(port), user=user, pw=pw,
+                                   wipe=False, spatial=False,
                                    db_database=name, db_schema=schema)
         elif db_type.upper() == 'ORACLE':
-            db = runner.Oracle(db_name='config', db_host=host, user=user, 
+            db = runner.Oracle(db_name='config', db_host=host, user=user,
                                pw=pw, wipe=False, spatial=False)
         else:
@@ -131 +131 @@
         self.cfg = cfg
         if db_type.upper() == 'POSTGRESQL':
-            db = runner.PostgreSQL(db_name='config', db_host=host, 
-                                   db_port=int(port), user=user, pw=pw, 
-                                   wipe=True, spatial=False, 
+            db = runner.PostgreSQL(db_name='config', db_host=host,
+                                   db_port=int(port), user=user, pw=pw,
+                                   wipe=True, spatial=False,
                                    db_database=name, db_schema=schema)
         elif db_type.upper() == 'ORACLE':
-            db = runner.Oracle(db_name='config', db_host=host, user=user, 
+            db = runner.Oracle(db_name='config', db_host=host, user=user,
                                pw=pw, wipe=True, spatial=False)
         else:
@@ -199 +199 @@
         """
         max_errors = self.cfg['error_handling.max_error_count_total']
+        import_date = self.cfg['import.import_date']
         basedir = os.getcwd()
         sim_dir = self.cfg['program_execution.base_folder']
@@ -208 +209 @@
                       % datamapping_def
                 raise Exception(msg)
-            data_imp = DataImporter(self.input_db, dm, True, self.logger,
-                                    'data import', self.lx, max_errors)
+            data_imp = DataImporter(self.input_db, dm, import_date,
+                                    self.logger, 'data import', self.lx,
+                                    max_errors)
             return data_imp
         finally:
             os.chdir(basedir)
+
+    mappingdef = self.cfg['import.mapping_definition']
+    format = self.cfg['import.format']
+    folder = self.cfg['import.data_dir']
+    datafiles = self.cfg['import.data_file']
+    separator = self.cfg['import.data_delimiter']
+    skipfirst = self.cfg['import.skip_first_line']
+    forced_ops = self.cfg['import.operations.operation_import']
+    if forced_ops:
+        # Import forced operations into input database
+        oformat = self.cfg['import.operations.operation_format']
+        ofile = self.cfg['import.operations.operation_file']
+        omap = self.cfg['import.operations.operation_modelchains']
+        oconv = self.cfg['import.operations.operation_conversion']
+    infomsg = 'Initializing SIMO importer!'
+    self.logger.log_message(log_name, 'info', infomsg)
+    simlevel = self.sim_level_name
+    mapping = self.struct_db.get_obj(sc.TEXT2DATA_DEF, mappingdef)
+    if mapping is None:
+        errormsg = "Data mapping definition '%s' couldn't be loaded" \
+                   % mappingdef
+        self.logger.log_message(log_name, 'error', errormsg)
+        return False
+    imp = DataImporter(self.input_db, mapping, import_date, self.logger,
+                       log_name, self.lx, max_errors)
+
+    initial_date = self.cfg['import.data_date']
+    month_first = self.cfg.get('import.month_first')
+    if month_first is None: month_first = False
 
     def _load_op_importer(self, text2operation=None,
@@ -339 +370 @@
             pass
 
-def do_import(runner_obj):
+def do_import(runner_obj, ds, op_ds):
     """
-    Run import on the database _if_ there is nothing imported yet.
+    Import data and forced operations into input dbs from data source
+
+    runner_obj -- executor of import
+    ds -- simulation data in string buffer
+    op_ds -- operaton data in string buffer
     """
     basedir = os.getcwd()
@@ -360 +395 @@
     finally:
         os.chdir(basedir)
+
+    dataset = {'format': 'by_level',
+               'buffers': [buff],
+               'sim_level': 'simulation',
+               'level_ind': [level_ind],
+               'data_date': data_date,
+               'clear_old': True
+              }
+    opdata = {'format': 'text',
+              'opbuffer': buff,
+              'separator': None
+             }
+
+    buffers = []
+    for file in datafiles:
+        fpath = os.path.join(folder, file)
+        buffers.append(open(fpath))
+    imp.import_data(format, buffers, separator=separator,
+                    skip_first=skipfirst, sim_level=simlevel,
+                    data_date=initial_date, month_first=month_first)
+    cont = not imp.terminate_import
+    if not imp.terminate_import and forced_ops:
+        conv_def, op2mc = None, None
+        if oformat in ('text',):
+            conv_def = self.struct_db.get_obj(sc.TEXT2OPERATION_DEF, oconv)
+            if conv_def is None and oformat:
+                errormsg = "Operation mapping definition '%s' couldn't "\
+                           "be loaded"  % oconv
+                self.logger.log_message(log_name, 'error', errormsg)
+                return False
+        elif oformat in ('xml', 'db'):
+            op2mc = self.struct_db.get_obj(sc.OPERATION2MODELCHAINS_DEF,
+                                           omap)
+        imp = OperationImporter(self.input_db, self.fops_db, oformat,
+                                conv_def, op2mc, self.logger, log_name,
+                                max_errors)
+        if oformat != 'db':
+            op_buffer = open(ofile)
+        else:
+            op_buffer = SQLiteDB(ofile, wipe=False, commit=False)
+        imp.import_operations(oformat, op_buffer, separator)
+        cont = not imp.terminate_import
+    return cont
 
 def get_runner(initial_setup, inifile, db_info, do_wipe=False,