fix: clearing some of the mongodb database before the other tasks; re…

…moved some prints

models/models_storage.py

@@ -71,21 +71,17 @@ def filenames(self,rel,extension):
             "trained_models",
             f"DSPNAE_{rel.name}*.{extension}"
         )
-        print("filenames_glob",filenames_glob)
         ret = {}
         filenames = glob.glob(filenames_glob)
         for curr in filenames:
             m = re.match(f'.*-v(\d+).{extension}', curr)
             if m:
                 # has version number
-                print(curr, m.groups())
                 version = int(m.groups()[0])
             else:
                 # no version number in filename: this was the first
-                print(f'filename {curr} not matching versioned pattern')
                 version = 0
             ret[version] = curr
-        print("ret",ret)
         return ret
 
     def kwargs_filenames(self,rel):

preprocess/dag.py

@@ -104,6 +104,11 @@ def parse(page, ti):
     large_mp.send(ti, rels_vals)
     large_mp.clear_recv(ti, f"download_{page}")
 
+def clear(ti, rel):
+    db = persistency.mongo_db()
+
+    # remove all data previously stored for this relation
+    db[rel].remove({})
 
 def persist(page, ti):
     """
@@ -118,9 +123,6 @@ def persist(page, ti):
     # FIXME: per-rel tasks
     for rel, sets in data.items():
 
-        # remove all data previously stored for this relation
-        db[rel].remove({})
-
         for activity_id, set_ in sets.items():
 
             # remove previously stored activity, if present FIXME: deprecated?
@@ -183,13 +185,12 @@ def encode(rel, ti):
     # remove existing data in the collection
     coll_out.remove({})
 
-    # how much time does each item require to be encoded
-
     for document in coll_in.find(no_cursor_timeout=True):
         document = dict(document)  # copy
         set_ = document['set_']
         set_size = get_set_size(set_)
 
+        # how much time does each item require to be encoded
         start = time.time()
 
         for field in rel.fields:
@@ -341,6 +342,15 @@ def to_tsets(rel, ti):
         op_kwargs={}
     )
 
+    t_clear = {}
+    for rel in rels:
+        t_clear[rel] = PythonOperator(
+            task_id="clear",
+            python_callable=clear,
+            start_date=days_ago(2),
+            op_kwargs={'rel': rel}
+        )
+
     t_persist = {}
     for page in pages:
         start = page*config.download_page_size
@@ -364,7 +374,9 @@ def to_tsets(rel, ti):
             start_date=days_ago(2),
             op_kwargs={'page': page}
         )
-        t_download >> t_parse >> t_persist[page]
+        for rel in rels:
+            t_clear[rel] >> t_download
+        t_download>> t_parse >> t_persist[page]
 
     for rel in rels:
         t_to_npa = PythonOperator(