Merge pull request #456 from digital-land/implement-pipeline-phase-manage

Author: pooleycodes

digital_land/commands.py

@@ -17,7 +17,6 @@
 import duckdb
 
 from digital_land.package.organisation import OrganisationPackage
-from digital_land.check import duplicate_reference_check
 from digital_land.specification import Specification
 from digital_land.collect import Collector
 from digital_land.collection import Collection, resource_path
@@ -32,17 +31,13 @@
 
 from digital_land.package.dataset import DatasetPackage
 from digital_land.package.dataset_parquet import DatasetParquetPackage
-from digital_land.phase.combine import FactCombinePhase
 from digital_land.phase.concat import ConcatFieldPhase
 from digital_land.phase.convert import ConvertPhase, execute
 from digital_land.phase.default import DefaultPhase
 from digital_land.phase.dump import DumpPhase
-from digital_land.phase.factor import FactorPhase
 from digital_land.phase.filter import FilterPhase
 from digital_land.phase.harmonise import HarmonisePhase
 from digital_land.phase.lookup import (
-    EntityLookupPhase,
-    FactLookupPhase,
     PrintLookupPhase,
 )
 from digital_land.phase.map import MapPhase
@@ -51,12 +46,9 @@
 from digital_land.phase.organisation import OrganisationPhase
 from digital_land.phase.parse import ParsePhase
 from digital_land.phase.patch import PatchPhase
-from digital_land.phase.priority import PriorityPhase
-from digital_land.phase.pivot import PivotPhase
 from digital_land.phase.prefix import EntityPrefixPhase
-from digital_land.phase.prune import FieldPrunePhase, EntityPrunePhase, FactPrunePhase
-from digital_land.phase.reference import EntityReferencePhase, FactReferencePhase
-from digital_land.phase.save import SavePhase
+from digital_land.phase.prune import FieldPrunePhase
+from digital_land.phase.reference import EntityReferencePhase
 from digital_land.pipeline import run_pipeline, Lookups, Pipeline
 from digital_land.pipeline.process import convert_tranformed_csv_to_pq
 from digital_land.schema import Schema
@@ -220,7 +212,6 @@ def pipeline_run(
     input_path,
     output_path: Path,
     collection_dir,  # TBD: remove, replaced by endpoints, organisations and entry_date
-    null_path=None,  # TBD: remove this
     issue_dir=None,
     operational_issue_dir="performance/operational_issue/",
     organisation_path=None,
@@ -244,37 +235,18 @@ def pipeline_run(
     if resource is None:
         resource = resource_from_path(input_path)
     dataset = dataset
-    schema = specification.pipeline[pipeline.name]["schema"]
-    intermediate_fieldnames = specification.intermediate_fieldnames(pipeline)
-    issue_log = IssueLog(dataset=dataset, resource=resource)
-    operational_issue_log = OperationalIssueLog(dataset=dataset, resource=resource)
-    column_field_log = ColumnFieldLog(dataset=dataset, resource=resource)
-    dataset_resource_log = DatasetResourceLog(dataset=dataset, resource=resource)
-    converted_resource_log = ConvertedResourceLog(dataset=dataset, resource=resource)
-    api = API(specification=specification)
-    entity_range_min = specification.get_dataset_entity_min(dataset)
-    entity_range_max = specification.get_dataset_entity_max(dataset)
 
-    # load pipeline configuration
-    skip_patterns = pipeline.skip_patterns(resource, endpoints)
-    columns = pipeline.columns(resource, endpoints=endpoints)
-    concats = pipeline.concatenations(resource, endpoints=endpoints)
-    patches = pipeline.patches(resource=resource, endpoints=endpoints)
-    lookups = pipeline.lookups(resource=resource)
-    default_fields = pipeline.default_fields(resource=resource, endpoints=endpoints)
-    default_values = pipeline.default_values(endpoints=endpoints)
-    combine_fields = pipeline.combine_fields(endpoints=endpoints)
-    redirect_lookups = pipeline.redirect_lookups()
-
-    # load config db
-    # TODO get more information from the config
+    # load config db and pass to Pipeline=> TODO move to pipeline class and use more widely
     # TODO in future we need better way of making specification optional for config
     if Path(config_path).exists():
         config = Config(path=config_path, specification=specification)
     else:
-        logging.error("Config path  does not exist")
+        logging.error("Config path does not exist")
         config = None
 
+    pipeline.config = config
+    pipeline.specification = specification
+
     # load organisations
     organisation = Organisation(
         organisation_path=organisation_path, pipeline_dir=Path(pipeline.path)
@@ -288,116 +260,40 @@ def pipeline_run(
         organisations = collection.resource_organisations(resource)
         entry_date = collection.resource_start_date(resource)
 
-    # Load valid category values
+    api = API(specification=specification)
     valid_category_values = api.get_valid_category_values(dataset, pipeline)
 
-    # resource specific default values
-    if len(organisations) == 1:
-        default_values["organisation"] = organisations[0]
-
-    # need an entry-date for all entries and for facts
-    # if a default entry-date isn't set through config then use the entry-date passed
-    # to this function
-    if entry_date:
-        if "entry-date" not in default_values:
-            default_values["entry-date"] = entry_date
-
-    # TODO Migrate all of this into a function in the Pipeline function
-    run_pipeline(
-        ConvertPhase(
-            path=input_path,
-            dataset_resource_log=dataset_resource_log,
-            converted_resource_log=converted_resource_log,
-            output_path=converted_path,
-        ),
-        NormalisePhase(skip_patterns=skip_patterns),
-        ParsePhase(),
-        ConcatFieldPhase(concats=concats, log=column_field_log),
-        FilterPhase(filters=pipeline.filters(resource)),
-        MapPhase(
-            fieldnames=intermediate_fieldnames,
-            columns=columns,
-            log=column_field_log,
-        ),
-        FilterPhase(filters=pipeline.filters(resource, endpoints=endpoints)),
-        PatchPhase(
-            issues=issue_log,
-            patches=patches,
-        ),
-        HarmonisePhase(
-            field_datatype_map=specification.get_field_datatype_map(),
-            issues=issue_log,
-            dataset=dataset,
-            valid_category_values=valid_category_values,
-        ),
-        DefaultPhase(
-            default_fields=default_fields,
-            default_values=default_values,
-            issues=issue_log,
-        ),
-        # TBD: move migrating columns to fields to be immediately after map
-        # this will simplify harmonisation and remove intermediate_fieldnames
-        # but effects brownfield-land and other pipelines which operate on columns
-        MigratePhase(
-            fields=specification.schema_field[schema],
-            migrations=pipeline.migrations(),
-        ),
-        OrganisationPhase(organisation=organisation, issues=issue_log),
-        FieldPrunePhase(fields=specification.current_fieldnames(schema)),
-        EntityReferencePhase(
-            dataset=dataset,
-            prefix=specification.dataset_prefix(dataset),
-            issues=issue_log,
-        ),
-        EntityPrefixPhase(dataset=dataset),
-        EntityLookupPhase(
-            lookups=lookups,
-            redirect_lookups=redirect_lookups,
-            issue_log=issue_log,
-            operational_issue_log=operational_issue_log,
-            entity_range=[entity_range_min, entity_range_max],
-        ),
-        SavePhase(
-            default_output_path("harmonised", input_path),
-            fieldnames=intermediate_fieldnames,
-            enabled=save_harmonised,
-        ),
-        EntityPrunePhase(dataset_resource_log=dataset_resource_log),
-        PriorityPhase(config=config, providers=organisations),
-        PivotPhase(),
-        FactCombinePhase(issue_log=issue_log, fields=combine_fields),
-        FactorPhase(),
-        FactReferencePhase(
-            field_typology_map=specification.get_field_typology_map(),
-            field_prefix_map=specification.get_field_prefix_map(),
-        ),
-        FactLookupPhase(
-            lookups=lookups,
-            redirect_lookups=redirect_lookups,
-            issue_log=issue_log,
-            odp_collections=specification.get_odp_collections(),
-        ),
-        FactPrunePhase(),
-        SavePhase(
-            output_path,
-            fieldnames=specification.factor_fieldnames(),
-        ),
+    # Transform the resource
+    issue_log = pipeline.transform(
+        input_path=input_path,
+        output_path=output_path,
+        organisation=organisation,
+        endpoints=endpoints,
+        organisations=organisations,
+        entry_date=entry_date,
+        resource=resource,
+        converted_path=converted_path,
+        harmonised_output_path=default_output_path("harmonised", input_path),
+        save_harmonised=save_harmonised,
+        valid_category_values=valid_category_values,
     )
 
-    # In the FactCombinePhase, when combine_fields has some values, we check for duplicates and combine values.
-    # If we have done this then we will not call duplicate_reference_check as we have already carried out a
-    # duplicate check and stop messages appearing in issues about reference values not being unique
-    if combine_fields == {}:
-        issue_log = duplicate_reference_check(issues=issue_log, csv_path=output_path)
+    # Save logs in pipeline
+    pipeline.save_logs(
+        issue_path=os.path.join(issue_dir, resource + ".csv"),
+        operational_issue_path=os.path.join(operational_issue_dir, resource + ".csv"),
+        column_field_path=os.path.join(column_field_dir, resource + ".csv"),
+        dataset_resource_path=os.path.join(dataset_resource_dir, resource + ".csv"),
+        converted_resource_path=os.path.join(converted_resource_dir, resource + ".csv"),
+    )
 
-    issue_log.apply_entity_map()
-    issue_log.save(os.path.join(issue_dir, resource + ".csv"))
+    # Parquet seperate save of issue log
     issue_log.save_parquet(os.path.join(output_log_dir, "issue/"))
-    operational_issue_log.save(output_dir=operational_issue_dir)
-    if column_field_dir:
-        column_field_log.save(os.path.join(column_field_dir, resource + ".csv"))
-    dataset_resource_log.save(os.path.join(dataset_resource_dir, resource + ".csv"))
-    converted_resource_log.save(os.path.join(converted_resource_dir, resource + ".csv"))
+
+    # create converted parquet in the var directory
+    cache_dir = Path(organisation_path).parent
+    transformed_parquet_dir = cache_dir / "transformed_parquet" / dataset
+
     # create converted parquet in the var director
     # TODO test without output_path conversation above to make sure we have a test that would've failed
     transformed_parquet_dir = output_path.parent

notebooks/debug_resource_transformation.ipynb

@@ -141,83 +141,10 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 92,
+   "execution_count": null,
    "metadata": {},
-   "outputs": [
-    {
-     "name": "stdout",
-     "output_type": "stream",
-     "text": [
-      "resource 1c192f194a6d7cb044006bbe0d7bb7909eed3783eeb8a53026fc15b9fe31a836 for dataset article-4-direction-area transformed to data/debug_resource_transformation/transformed/article-4-direction-area/1c192f194a6d7cb044006bbe0d7bb7909eed3783eeb8a53026fc15b9fe31a836.csv\n"
-     ]
-    }
-   ],
-   "source": [
-    "\n",
-    "output_path = data_dir / 'transformed' / dataset / f'{resource_hash}.csv'\n",
-    "output_path.parent.mkdir(parents=True, exist_ok=True)\n",
-    "converted_path = data_dir / 'converted' / dataset / f'{resource_hash}.csv'\n",
-    "converted_path.parent.mkdir(parents=True, exist_ok=True)\n",
-    "\n",
-    "# create pipeline object\n",
-    "pipeline = Pipeline(pipeline_dir, dataset)\n",
-    "\n",
-    "# create logs\n",
-    "issue_dir = data_dir / 'issues' / dataset\n",
-    "issue_dir.mkdir(parents=True, exist_ok=True)\n",
-    "operational_issue_dir = data_dir / 'performance' / 'operational_issues'\n",
-    "operational_issue_dir.mkdir(parents=True, exist_ok=True)\n",
-    "column_field_dir = cache_dir / 'column_field' / dataset\n",
-    "column_field_dir.mkdir(parents=True, exist_ok=True)\n",
-    "dataset_resource_dir = cache_dir / 'dataset_resource' / dataset\n",
-    "dataset_resource_dir.mkdir(parents=True, exist_ok=True)\n",
-    "converted_resource_dir = cache_dir / 'converted_resource' / dataset\n",
-    "converted_resource_dir.mkdir(parents=True, exist_ok=True)\n",
-    "output_log_dir = data_dir / 'log'\n",
-    "output_log_dir.mkdir(parents=True, exist_ok=True)\n",
-    "\n",
-    "# get endpoints from the collection TODO include redirects\n",
-    "collection = Collection(directory = collection_dir)\n",
-    "collection.load()\n",
-    "endpoints = collection.resource_endpoints(resource_hash)\n",
-    "organisations = collection.resource_organisations(resource_hash)\n",
-    "entry_date = collection.resource_start_date(resource_hash)\n",
-    "\n",
-    "# build config from downloaded files \n",
-    "config_path = cache_dir / 'config.sqlite3'\n",
-    "config = Config(path=config_path, specification=spec)\n",
-    "config.create()\n",
-    "tables = {key: pipeline.path for key in config.tables.keys()}\n",
-    "config.load(tables)\n",
-    "\n",
-    "pipeline_run(\n",
-    "    dataset=dataset,\n",
-    "    pipeline=pipeline,\n",
-    "    specification=spec,\n",
-    "    input_path=resource_path,\n",
-    "    output_path=output_path,\n",
-    "    collection_dir=collection_dir,  # TBD: remove, replaced by endpoints, organisations and entry_date\n",
-    "    null_path=None,  # TBD: remove this\n",
-    "    issue_dir=issue_dir,\n",
-    "    operational_issue_dir=operational_issue_dir,\n",
-    "    organisation_path=org_path,\n",
-    "    save_harmonised=False,\n",
-    "    #  TBD save all logs in  a log directory, this will mean only one path passed in.\n",
-    "    column_field_dir=column_field_dir,\n",
-    "    dataset_resource_dir=dataset_resource_dir,\n",
-    "    converted_resource_dir=converted_resource_dir,\n",
-    "    cache_dir=cache_dir,\n",
-    "    endpoints=endpoints,\n",
-    "    organisations=organisations,\n",
-    "    entry_date=entry_date,\n",
-    "    config_path=config_path,\n",
-    "    resource=resource_hash,\n",
-    "    output_log_dir=output_log_dir,\n",
-    "    converted_path=converted_path,\n",
-    ")\n",
-    "\n",
-    "print(f'resource {resource_hash} for dataset {dataset} transformed to {output_path}')"
-   ]
+   "outputs": [],
+   "source": "\noutput_path = data_dir / 'transformed' / dataset / f'{resource_hash}.csv'\noutput_path.parent.mkdir(parents=True, exist_ok=True)\nconverted_path = data_dir / 'converted' / dataset / f'{resource_hash}.csv'\nconverted_path.parent.mkdir(parents=True, exist_ok=True)\n\n# create pipeline object\npipeline = Pipeline(pipeline_dir, dataset)\n\n# create logs\nissue_dir = data_dir / 'issues' / dataset\nissue_dir.mkdir(parents=True, exist_ok=True)\noperational_issue_dir = data_dir / 'performance' / 'operational_issues'\noperational_issue_dir.mkdir(parents=True, exist_ok=True)\ncolumn_field_dir = cache_dir / 'column_field' / dataset\ncolumn_field_dir.mkdir(parents=True, exist_ok=True)\ndataset_resource_dir = cache_dir / 'dataset_resource' / dataset\ndataset_resource_dir.mkdir(parents=True, exist_ok=True)\nconverted_resource_dir = cache_dir / 'converted_resource' / dataset\nconverted_resource_dir.mkdir(parents=True, exist_ok=True)\noutput_log_dir = data_dir / 'log'\noutput_log_dir.mkdir(parents=True, exist_ok=True)\n\n# get endpoints from the collection TODO include redirects\ncollection = Collection(directory = collection_dir)\ncollection.load()\nendpoints = collection.resource_endpoints(resource_hash)\norganisations = collection.resource_organisations(resource_hash)\nentry_date = collection.resource_start_date(resource_hash)\n\n# build config from downloaded files \nconfig_path = cache_dir / 'config.sqlite3'\nconfig = Config(path=config_path, specification=spec)\nconfig.create()\ntables = {key: pipeline.path for key in config.tables.keys()}\nconfig.load(tables)\n\npipeline_run(\n    dataset=dataset,\n    pipeline=pipeline,\n    specification=spec,\n    input_path=resource_path,\n    output_path=output_path,\n    collection_dir=collection_dir,  # TBD: remove, replaced by endpoints, organisations and entry_date\n    issue_dir=issue_dir,\n    operational_issue_dir=operational_issue_dir,\n    organisation_path=org_path,\n    save_harmonised=False,\n    #  TBD save all logs in  a log directory, this will mean only one path passed in.\n    column_field_dir=column_field_dir,\n    dataset_resource_dir=dataset_resource_dir,\n    converted_resource_dir=converted_resource_dir,\n    cache_dir=cache_dir,\n    endpoints=endpoints,\n    organisations=organisations,\n    entry_date=entry_date,\n    config_path=config_path,\n    resource=resource_hash,\n    output_log_dir=output_log_dir,\n    converted_path=converted_path,\n)\n\nprint(f'resource {resource_hash} for dataset {dataset} transformed to {output_path}')"
   },
   {
    "cell_type": "code",
@@ -248,4 +175,4 @@
  },
  "nbformat": 4,
  "nbformat_minor": 2
-}
+}