From ac30608b4bdbc06d7f9383a2a23d475527449fd7 Mon Sep 17 00:00:00 2001 From: Sophie Pike Date: Wed, 9 Sep 2026 16:33:51 +0100 Subject: [PATCH 1/2] feat: updated the example_2 code to account for correct logging and run_directory location searches. Also provides stage_results for use in demo --- examples/pipeline_2/conf.yaml | 8 +----- examples/pipeline_2/data/orders_cleaned.csv | 7 ----- examples/pipeline_2/data/orders_prepped.csv | 7 ----- examples/pipeline_2/scripts/0_clean_data.py | 26 +++++++++++++++--- examples/pipeline_2/scripts/1_derive_vars.py | 22 +++++++++++++-- examples/pipeline_2/scripts/2_reporting.py | 28 +++++++++++++++----- 6 files changed, 64 insertions(+), 34 deletions(-) delete mode 100644 examples/pipeline_2/data/orders_cleaned.csv delete mode 100644 examples/pipeline_2/data/orders_prepped.csv diff --git a/examples/pipeline_2/conf.yaml b/examples/pipeline_2/conf.yaml index 58d076b..2b9ed4c 100644 --- a/examples/pipeline_2/conf.yaml +++ b/examples/pipeline_2/conf.yaml @@ -17,7 +17,7 @@ pipeline_variables: working_dir: examples/pipeline_2 project_root: examples/pipeline_2 data_dir: examples/pipeline_2/data - output_dir: examples/pipeline_2/outputs + output_dir: examples/pipeline_2/runs log_dir: examples/pipeline_2/logs overwrite: True metadata: @@ -48,11 +48,7 @@ stage_configuration: - "age" - "dob" - "address" - output_location: examples/pipeline_2/data/orders_cleaned.csv - input_location: examples/pipeline_2/data/orders.csv 1_derive_vars: - input_location: examples/pipeline_2/data/orders_cleaned.csv - output_location: examples/pipeline_2/data/orders_prepped.csv delivery_times: north: 14 south: 4 @@ -82,8 +78,6 @@ stage_configuration: total_production_cost: "Total_production_cost" order_profit: "Order_profit" 2_reporting: - input_location: examples/pipeline_2/data/orders_prepped.csv - report_location: examples/pipeline_2/outputs/order_analysis.md region: "Region" order_profit: "Order_profit" quantity: "Quantity" diff --git a/examples/pipeline_2/data/orders_cleaned.csv b/examples/pipeline_2/data/orders_cleaned.csv deleted file mode 100644 index c3e721b..0000000 --- a/examples/pipeline_2/data/orders_cleaned.csv +++ /dev/null @@ -1,7 +0,0 @@ -Order_id,Region,Product,Quantity,Unit_price,Order_date -1001,north,notebook,2,3.5,2026-06-01 -1002,south,pen,5,1.2,2026-06-01 -1003,north,notebook,1,3.5,2026-06-02 -1004,west,folder,3,2.75,2026-06-03 -1005,east,pen,4,1.2,2026-06-03 -1006,north,notebook,2,3.5,2026-06-04 diff --git a/examples/pipeline_2/data/orders_prepped.csv b/examples/pipeline_2/data/orders_prepped.csv deleted file mode 100644 index b6c5d70..0000000 --- a/examples/pipeline_2/data/orders_prepped.csv +++ /dev/null @@ -1,7 +0,0 @@ -Order_id,Region,Product,Quantity,Unit_price,Order_date,Estimated_delvery_date,Delivery_day,Total_cost,Order_day,Order_month,Large_order,Small_order,Postage,Total_production_cost,Order_profit -1001,north,notebook,2,3.5,2026-06-01,2026-06-15,Monday,7.0,Monday,June,False,False,2.5,2.0,2.5 -1002,south,pen,5,1.2,2026-06-01,2026-06-05,Friday,6.0,Monday,June,True,False,5.0,1.5,-0.5 -1003,north,notebook,1,3.5,2026-06-02,2026-06-16,Tuesday,3.5,Tuesday,June,False,True,1.0,1.0,1.5 -1004,west,folder,3,2.75,2026-06-03,2026-06-10,Wednesday,8.25,Wednesday,June,False,False,2.5,2.25,3.5 -1005,east,pen,4,1.2,2026-06-03,2026-06-10,Wednesday,4.8,Wednesday,June,True,False,5.0,1.2,-1.4000000000000004 -1006,north,notebook,2,3.5,2026-06-04,2026-06-18,Thursday,7.0,Thursday,June,False,False,2.5,2.0,2.5 diff --git a/examples/pipeline_2/scripts/0_clean_data.py b/examples/pipeline_2/scripts/0_clean_data.py index 50ad0a3..f54eb4d 100644 --- a/examples/pipeline_2/scripts/0_clean_data.py +++ b/examples/pipeline_2/scripts/0_clean_data.py @@ -32,20 +32,38 @@ def standardise_columns(df): def main(context=None): + # Load stage configuration config = context.get_stage_config("0_clean_data") - print(config) - orders = pd.read_csv(config["input_location"]) + # Calculate location for run outputs + output_root = context.resolve_output_root() + full_output_location = output_root / "orders_cleaned.csv" + # Calculate location of data input + data_dir = context.get_data_dir() + orders = pd.read_csv(data_dir / "orders.csv") + + # Source variable lists from stage configuration expected_variables = config["expected_variables"] identifiable_cols = config["identifiable_cols"] + # Run functions required for this stage check_variables(orders, expected_variables) print(orders.dtypes) orders = remove_identifiable(orders, identifiable_cols) orders = standardise_columns(orders) - orders.to_csv(config["output_location"], index=False) + + # Save cleaned data to run specific output location defined earlier in function + orders.to_csv(full_output_location, index=False) + + return { + "output_location": str(full_output_location), + "expected_variables": expected_variables, + "identifiable_cols": identifiable_cols, + "record_count": len(orders), + "columns": list(orders.columns), + } if __name__ == "__main__": - main() + main(context=None) diff --git a/examples/pipeline_2/scripts/1_derive_vars.py b/examples/pipeline_2/scripts/1_derive_vars.py index 94d4d42..95163f2 100644 --- a/examples/pipeline_2/scripts/1_derive_vars.py +++ b/examples/pipeline_2/scripts/1_derive_vars.py @@ -58,11 +58,20 @@ def profit_per_order(df): def main(context=None): + # Get stage configuration for this stage config = context.get_stage_config("1_derive_vars") - df = pd.read_csv(config["input_location"]) + # Source data path for previous stage results which are required + # for this stage and read those in + data_path = context.resolve_given_path( + "0_clean_data", "output_location", "orders_cleaned.csv", context.get_data_dir() + ) + df = pd.read_csv(data_path) + + # Source variable list from stage configuration for this stage delivery_times = config["delivery_times"] + # Run relevant functions for this stage df = correct_date_time(df) df = estimate_delivery(df, delivery_times) df = total_cost(df) @@ -71,7 +80,16 @@ def main(context=None): df = postage_cost(df) df = production_cost(df) df = profit_per_order(df) - df.to_csv(config["output_location"], index=False) + + # Save results to output location calculated based on run_directory + output_root = context.resolve_output_root() + df.to_csv(output_root / "orders_prepped.csv", index=False) + + return { + "output_location": str(output_root / "orders_prepped.csv"), + "record_count": len(df), + "columns": list(df.columns), + } if __name__ == "__main__": diff --git a/examples/pipeline_2/scripts/2_reporting.py b/examples/pipeline_2/scripts/2_reporting.py index ba71526..311f616 100644 --- a/examples/pipeline_2/scripts/2_reporting.py +++ b/examples/pipeline_2/scripts/2_reporting.py @@ -1,5 +1,3 @@ -from pathlib import Path - import pandas as pd @@ -116,20 +114,27 @@ def curate_report(report, values): ) -def write_report(report): - report_file = Path("examples/pipeline_2/outputs/order_analysis.md") +def write_report(report, output_path): + report_file = output_path report_file.write_text("\n".join(report), encoding="utf-8") -def main(): - orders = pd.read_csv("examples/pipeline_2/data/orders_prepped.csv") +def main(context=None): + # Read in processed data from stage "1_derive_vars" + data_loc = context.resolve_given_path( + "1_derive_vars", "output_location", "orders_prepped.csv", context.get_data_dir() + ) + orders = pd.read_csv(data_loc) + # Set empty list and dictionary to store report and values report = [] values = {} + # Set the number format for the report to 2 decimal places num_format = "{:.2f}" + # Run functions required for the stage per_region_profits(orders, values, num_format) per_region_quantity(orders, values) orders_per_day(orders, values) @@ -138,7 +143,16 @@ def main(): total_summaries(orders, values, num_format) profit_per_product(orders, values, num_format) curate_report(report, values) - write_report(report) + + # Calculate the output location for the final report and write to it + output_path = context.resolve_output_root() / "order_analysis.md" + write_report(report, output_path) + + return { + "report_location": str(output_path), + "total_orders": int(values["total_count"]), + "total_profit": float(values["total_profit"]), + } if __name__ == "__main__": From 48c3d38121e7a3a8af1fba4608e75021841c9841 Mon Sep 17 00:00:00 2001 From: Sophie Pike Date: Wed, 9 Sep 2026 16:44:27 +0100 Subject: [PATCH 2/2] fix: remove default None value in main() function calls to ensure valid runs only --- examples/pipeline_2/scripts/0_clean_data.py | 2 +- examples/pipeline_2/scripts/1_derive_vars.py | 2 +- examples/pipeline_2/scripts/2_reporting.py | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/examples/pipeline_2/scripts/0_clean_data.py b/examples/pipeline_2/scripts/0_clean_data.py index f54eb4d..5027787 100644 --- a/examples/pipeline_2/scripts/0_clean_data.py +++ b/examples/pipeline_2/scripts/0_clean_data.py @@ -31,7 +31,7 @@ def standardise_columns(df): return df -def main(context=None): +def main(context): # Load stage configuration config = context.get_stage_config("0_clean_data") diff --git a/examples/pipeline_2/scripts/1_derive_vars.py b/examples/pipeline_2/scripts/1_derive_vars.py index 95163f2..0edf60f 100644 --- a/examples/pipeline_2/scripts/1_derive_vars.py +++ b/examples/pipeline_2/scripts/1_derive_vars.py @@ -57,7 +57,7 @@ def profit_per_order(df): return df -def main(context=None): +def main(context): # Get stage configuration for this stage config = context.get_stage_config("1_derive_vars") diff --git a/examples/pipeline_2/scripts/2_reporting.py b/examples/pipeline_2/scripts/2_reporting.py index 311f616..2007dc8 100644 --- a/examples/pipeline_2/scripts/2_reporting.py +++ b/examples/pipeline_2/scripts/2_reporting.py @@ -120,7 +120,7 @@ def write_report(report, output_path): report_file.write_text("\n".join(report), encoding="utf-8") -def main(context=None): +def main(context): # Read in processed data from stage "1_derive_vars" data_loc = context.resolve_given_path( "1_derive_vars", "output_location", "orders_prepped.csv", context.get_data_dir()