Multiprocessing works but without error handling in the worker class.

grompy.yaml

@@ -7,7 +7,7 @@ parcel_info:
   shape_file: /home/wit015/Data/groenmonitor/BRP/BRP_10rows.shp
   table_name: parcel_info
 datasets:
-  sentinel2_values:
+  sentinel2_reflectance_values:
     dsn: sqlite:////home/wit015/Data/groenmonitor/sentinel2_reflectance_values.db3
     bands:
       NDVI: /home/wit015/Data/groenmonitor/Optisch/zonal_stats_mean_2019_ADC.csv
@@ -21,7 +21,7 @@ datasets:
       B11: /home/wit015/Data/groenmonitor/Optisch/zonal_stats_mean_B11_2019_ADC.csv
       B12: /home/wit015/Data/groenmonitor/Optisch/zonal_stats_mean_B12_2019_ADC.csv
       B8A: /home/wit015/Data/groenmonitor/Optisch/zonal_stats_mean_B8A_2019_ADC.csv
-  sentinel2_std:
+  sentinel2_reflectance_std:
     dsn: sqlite:////home/wit015/Data/groenmonitor/sentinel2_reflectance_std.db3
     bands:
       NDVI: /home/wit015/Data/groenmonitor/Optisch/zonal_stats_std_2019_ADC.csv
@@ -43,7 +43,7 @@ datasets:
       VV_std: /home/wit015/Data/groenmonitor/Radar/zonal_stats_std_VV_2019_ADC.csv
       VH_std: /home/wit015/Data/groenmonitor/Radar/zonal_stats_std_VH_2019_ADC.csv
   sentinel1_coherence:
-    dsn: sqlite:////home/wit015/Data/groenmonitor/sentinel1_backscatter.db3
+    dsn: sqlite:////home/wit015/Data/groenmonitor/sentinel1_coherence.db3
     bands:
         S1A_VV: /home/wit015/Data/groenmonitor/Radar/zonal_stats_mean_coh_S1A_VV_ALL_2019_ADC.csv
         S1A_VV_std: /home/wit015/Data/groenmonitor/Radar/zonal_stats_std_coh_S1A_VV_ALL_2019_ADC.csv

grompy/load_data.py

+import sys
 from pathlib import Path
 from csv import DictReader
 import time
@@ -107,8 +108,7 @@ def write_to_database(engine, table_name, mean_csv_readers, nlines):
 
 
 def load_satellite_csv(dataset_name, dsn, bands):
-    # nlines = count_lines(bands)  # 803016
-    nlines = 803016
+    nlines = count_lines(bands)  # 803016
     mean_csv_readers = {}
     for column_name, csv_fname in bands.items():
         mean_csv_readers[column_name] = DictReader(open(csv_fname))
@@ -127,11 +127,21 @@ def load_data(yaml_file):
     grompy_conf = yaml.safe_load(open(yaml_file))
     parcel_info = grompy_conf.pop("parcel_info")
     load_parcel_info(**parcel_info)
-    i =0
 
-    # pool = mp.Pool(mp.cpu_count())
+    process_list = []
     for dataset_name, description in grompy_conf["datasets"].items():
-        i += 1
-        if i <=2:
-            continue
-        load_satellite_csv(dataset_name, **description)
\ No newline at end of file
+        print(f"Starting loading of: {dataset_name}")
+        p = mp.Process(target=load_satellite_csv, args=(dataset_name,), kwargs=description)
+        process_list.append(p)
+    for p in process_list:
+        p.start()
+
+    while any([p.is_alive() for p in process_list]):
+        try:
+            print("Data loading running.....")
+            time.sleep(10)
+        except KeyboardInterrupt:
+            for p in process_list:
+                p.kill()
+            sys.exit()
+

grompy/util.py

@@ -48,16 +48,13 @@ def printProgressBar (iteration, total, prefix = '', suffix = '', decimals = 1,
         print()
 
 
-def prepare_db(dsn, table_name, bands, has_sensor_info):
+def prepare_db(dsn, table_name, bands):
     engine = create_engine(dsn)
     meta = MetaData(engine)
     tbl = Table(table_name, meta,
                 Column('fieldID', Integer, primary_key=True, nullable=False),
                 Column('day', Date, primary_key=True, nullable=False),
                 )
-    if has_sensor_info:
-        tbl.append_column(Column('sensor', Text, primary_key=True, nullable=False))
-
     for col_name in bands:
         tbl.append_column(Column(col_name, Float, nullable=True))