update disk file structure

Author: antheas

src/pasteur/kedro/pipelines/dataset.py

@@ -29,7 +29,7 @@ def create_dataset_pipeline(
             for t in dataset.raw_tables
         ]
         outputs += [
-            D("typed", f"{dataset}.typed.{t}", ["orig", "typed", dataset, t], type="pq")
+            D("typed", f"{dataset}.typed.{t}", ["ds", dataset, "typed", t], type="pq")
             for t in dataset.raw_tables
         ]
         prefix = "typed."
@@ -49,7 +49,7 @@ def create_dataset_pipeline(
     ]
 
     outputs += [
-        D("interim", f"{dataset}.{t}", ["orig", "interim", dataset, t], type="pq")
+        D("interim", f"{dataset}.{t}", ["ds", dataset, "tables", t], type="pq")
         for t in tables
     ]
     meta_tables = PipelineMeta(
@@ -72,7 +72,7 @@ def create_dataset_pipeline(
 
     meta_keys = PipelineMeta(
         pipe,
-        [D("keys", f"{dataset}.keys", ["orig", "keys", dataset])],
+        [D("keys", f"{dataset}.keys", ["ds", dataset, "keys"])],
     )
 
     return meta_tables + meta_keys

src/pasteur/kedro/pipelines/metrics.py

@@ -124,13 +124,13 @@ def _create_fit_pipeline(
             D(
                 "measure",
                 f"{view}.msr.{name}",
-                ["measure", "dataset", view, name, "metric"],
+                ['view', view, 'msr', name, "metric"],
                 type="pkl",
             ),
             D(
                 "measure",
                 f"{view}.msr.{name}_pre",
-                ["measure", "dataset", view, name, "pre"],
+                ['view', view, 'msr', name, "pre"],
                 type="pkl",
             ),
         ]
@@ -173,7 +173,7 @@ def _create_process_pipeline(
             D(
                 "measure",
                 f"{view}.{syn_split}.{name}_data",
-                ["synth", "measure", "dataset", view, name, "data"],
+                ['synth', view, syn_split, 'msr', name],
                 type="pkl",
                 versioned=True,
             ),

src/pasteur/kedro/pipelines/synth.py

@@ -50,14 +50,14 @@ def create_synth_pipeline(
         D(
             "synth_models",
             f"{view}.{fr.name}.model",
-            ["synth", "models", f"{view}.{fr.name}"],
+            ["synth", view, fr.name, 'model'],
             versioned=True,
             type="pkl",
         ),
         D(
             "synth_output",
             f"{view}.{fr.name}.enc",
-            ["synth", "enc", f"{view}.{fr.name}"],
+            ["synth", view, fr.name, 'enc'],
             versioned=True,
             type="multi",
         ),

src/pasteur/kedro/pipelines/transform.py

@@ -135,11 +135,11 @@ def create_fit_pipeline(
     return PipelineMeta(
         pipeline(trn_fit_nodes + enc_fit_nodes, tags=TAGS_TRANSFORM),
         [
-            D("transformers", f"{view}.trn.{t}", ["views", "trn", view, t], type="pkl")
+            D("transformers", f"{view}.trn.{t}", ["views", view, "trn",  t], type="pkl")
             for t in view.tables
         ]
         + [
-            D("encoders", f"{view}.enc.{enc}", ["views", "enc", view, enc], type="pkl")
+            D("encoders", f"{view}.enc.{enc}", ["views", view, 'enc', enc], type="pkl")
             for enc in encs
         ],
     )
@@ -177,21 +177,21 @@ def create_transform_pipeline(
                 D(
                     "split_transformed",
                     f"{view}.{split}.ctx_{t}",
-                    ["views", "ctx", f"{view}.{split}", t],
+                    ["views", view, split, "ctx", t],
                 )
             )
             outputs.append(
                 D(
                     "split_transformed",
                     f"{view}.{split}.bst_{t}",
-                    ["views", "bst", f"{view}.{split}", t],
+                    ["views", view, split, "bst", t],
                 )
             )
             outputs.append(
                 D(
                     "split_transformed",
                     f"{view}.{split}.ids_{t}",
-                    ["views", "ids", f"{view}.{split}", t],
+                    ["views", view, split, "ids", t],
                 )
             )
 
@@ -218,7 +218,7 @@ def create_transform_pipeline(
                 # FIXME: Pass proper layer properly, don't infer
                 "synth_reencoded" if retransform else "split_encoded",
                 f"{view}.{split}.{enc}",
-                ["synth" if retransform else "views", enc, f"{view}.{split}"],
+                ["synth" if retransform else "views", view, split],
                 versioned=retransform,
             )
         )
@@ -272,26 +272,26 @@ def create_reverse_pipeline(view: View, alg: str, enc: str):
                 D(
                     "synth_decoded",
                     f"{view}.{alg}.bst_{t}",
-                    ["synth", "bst", f"{view}.{alg}", t],
+                    ["synth", view, alg, "bst", t],
                     versioned=True,
                 ),
                 D(
                     "synth_decoded",
                     f"{view}.{alg}.ids_{t}",
-                    ["synth", "ids", f"{view}.{alg}", t],
+                    ["synth", view, alg, "ids", t],
                     versioned=True,
                 ),
                 D(
                     "synth_decoded",
                     f"{view}.{alg}.ctx_{t}",
-                    ["synth", "ctx", f"{view}.{alg}", t],
+                    ["synth", view, alg, "ctx", t],
                     versioned=True,
                     type="multi",
                 ),
                 D(
                     "synth_reversed",
                     f"{view}.{alg}.{t}",
-                    ["synth", "dec", f"{view}.{alg}", t],
+                    ["synth", view, alg, 'tables', t],
                     versioned=True,
                 ),
             ]

src/pasteur/kedro/pipelines/views.py

@@ -60,7 +60,7 @@ def create_view_pipeline(view: View):
             ]
         ),
         [
-            D("primary", f"{view}.view.{t}", ["views", "primary", view, t], type="pq")
+            D("primary", f"{view}.view.{t}", ["view", view, "tables", t], type="pq")
             for t in view.tables
         ],
     )
@@ -81,7 +81,7 @@ def create_meta_pipeline(view: View):
             ],
             tags=TAGS_VIEW_META,
         ),
-        [D("metadata", f"{view}.metadata", ["views", "metadata", view], type="pkl")],
+        [D("metadata", f"{view}.metadata", ["view", view, "metadata"], type="pkl")],
     )
 
 
@@ -121,7 +121,7 @@ def create_keys_pipeline(view: View, splits: list[str]):
 
     return PipelineMeta(
         pipe,
-        [D("keys", f"{view}.keys.{s}", ["views", "keys", view, s]) for s in splits],
+        [D("keys", f"{view}.keys.{s}", ["view", view, "keys", s]) for s in splits],
     )
 
 
@@ -152,7 +152,7 @@ def create_filter_pipeline(view: View, splits: list[str]):
             namespace=view.name,
         ),
         [
-            D("splits", f"{view}.{s}.{t}", ["views", "primary", f"{view}.{s}", t])
+            D("splits", f"{view}.{s}.{t}", ["views", view, s, 'tables', t])
             for t in tables
             for s in splits
         ],