Processing Data functions

This article is a deep dive into the compatible clean/metadata wrapper, pd_process_data(), and the functions that consume its output: pd_deflate_pipeline() (the second, batch-deflation stage), pd_deflation()/deflation(), and log_report(). Code in this article is illustrative and does not execute when the article is built — running it requires a configured working release and write access to PIP storage.

For the end-to-end orchestration and how this wrapper fits with the other two, see PIP Data Pipeline: Orchestration Overview. For the internal mechanics of DLW acquisition and validation (the step before this one), see Validating Data.

`pd_run_pipeline()`: incremental top-level execution

Use pd_run_pipeline() after DLW validation:

result <- pd_run_pipeline(
  force = FALSE,
  force_surveys = NULL,
  bootstrap = FALSE,
  checkpoint_size = 25L,
  checkpoint_seconds = Inf,
  verbose = FALSE
)

The only durable nodes are clean:<survey_id>, metadata:<pip_id>, and deflate:<pip_id>. Internal load, PFW merge, recode, auxiliary attachment, and save functions are fingerprint components rather than cached nodes. The result reports current/stale/forced planning state and cached/runnable/success/failed/ skipped/blocked execution state. Cached clean nodes do not load household data.

Stamp owns immutable artifact versions and exact receipts. The C2 dependency manifest is the only pipdata currentness/provenance index. Success is published only after exact receipt verification, inventory reconciliation, and a manifest checkpoint. A Colombia 2018 CPI change refreshes matching Colombia 2018 metadata and deflate nodes only.

force_surveys is additive to normal invalidation. Explicit bootstrap is required for absent or unknown legacy provenance. Recoverable failures block descendants while independent siblings can continue. A retry starts a new authoritative replan from the last valid manifest; no run cursor is persisted. Production remains blocked until signed target Windows/SMB fencing and immutable unique-rename evidence are complete.

The existing wrappers below remain compatible. They keep their signatures, durable aliases, positional behavior, and master-inventory return types.

Important: pd_process_data() cleans surveys and attaches metadata — it does not deflate welfare. Deflation is a separate pipeline stage: use pd_deflate_pipeline() (below) to batch-deflate the whole inventory, or pd_deflation() (see Deflation is a separate step below) for a single survey.

`pd_process_data()`: clean surveys and build metadata

pd_process_data() iterates a validated DLW inventory, merges auxiliary data with each survey, cleans it, attaches metadata, saves versioned outputs, and returns the updated PIP master inventory:

new_pip_inv <- pd_process_data(
  inv          = NULL, # NULL loads pipload::load_gmd_valid_inv() internally
  aux_measures = c("pfw", "cpi", "ppp", "pop", "gdp", "pce"),
  force        = FALSE,
  verbose      = TRUE
)

inv — the completed DLW validation inventory (as produced by Validating Data’s pipdata_validate_gmd()). When NULL (the default), it is loaded internally via pipload::load_gmd_valid_inv(). Before planning, only available rows with status valid or invalid are retained. Recognized legacy blank/unavailable retry rows are removed, and malformed completed rows abort.
aux_measures — which auxiliary datasets to load and merge/attach.
force — when TRUE, temporarily switches stamp versioning to "timestamp" and bypasses the master-inventory comparison, forcing every survey in inv to be reprocessed.

What happens inside

Guard completed validation input before snapshots, dependency planning, force selection, or row lookup. The guard also runs inside execution preparation for internal callers, so legacy execution-control rows cannot create cleaning or metadata actions. Existing policy still permits both completed valid and invalid rows.
Load auxiliary data for each measure in aux_measures via pipload::load_aux_data().
Determine which surveys need (re-)processing — valid_dlw_load() detects auxiliary-file changes (logging an aux_changes_inf entry when found), filters inv to the relevant modules (ALL/GROUP/HIST/GPWG/BIN), keeps the latest version of each survey, and — unless force = TRUE — drops surveys already present in the master inventory.
Sync the recode spec once via sync_recode_spec(), before the loop. This compares the package’s bundled recode spec (see What is the recode spec? below) against the latest version saved in stamp: if there is no stamp version yet, or the package version has changed (compared by hash), it saves a new stamp version; otherwise it reuses the existing one. Either way it returns the active spec once, which is then threaded down to every survey’s pd_dlw_clean() call — so the comparison/save only happens a single time per pd_process_data() run, instead of once per survey.
Process each survey with process_data(), one at a time: inv_dlw_load() (load raw survey) -> pd_cpfw_merge() (merge Price Framework metadata) -> pd_dlw_clean() (S3-dispatched cleaning, using the shared recode spec) -> pd_aux_attr() (attach CPI/PPP/population/GDP/PCE metadata) -> save_pip_data(alias = "pip") / save_pip_data(alias = "pip_meta") (write cleaned data and metadata to PIP storage). A per-survey failure is caught (as a piperr or generic error), logged to "pipdata_log" with the survey id and error class, and the survey is skipped — it does not abort the rest of the run. No deflation step is part of this chain.
Log a processing summary (process_summary_inf: totals, successes, failures) and, if any surveys failed, a null_svys_inf entry listing them.
Rebuild the PIP master inventory via build_pip_inventory() from the successful surveys’ stamp catalog entries, and return it.

pd_change_report() applies the same completed-inventory guard without writing artifacts:

change_plan <- pd_change_report(
  inv = pipload::load_gmd_valid_inv(),
  master = pipload::load_pip_master_inventory()
)

What is the recode spec?

The recode spec is the set of recoding rules for the raw DLW variables that pd_dlw_clean() cleans into PIP variables. It’s defined in inst/extdata/recode_spec.yml, bundled with the package: for each variable it declares a type, a recode_type (one of range_clamp, binary_map, haven_labels, binned_from_continuous, quantile_from_continuous, or indicator), and the type-specific parameters that rule needs (e.g. a valid_range for range_clamp, a mapping for binary_map/haven_labels). apply_recode_spec() (called internally by pd_dlw_clean()) applies every rule that matches a column present in the survey. Replace-type rules (range_clamp, binary_map, haven_labels) recode the source column in place (renaming it to the target variable name when they differ, e.g. urban → area); derive-type rules (binned_from_continuous, quantile_from_continuous) keep the source column and add the derived variable alongside it (e.g. age stays, age_group is added).

Deflation is a separate step

pd_deflation()/deflation() deflates a single cleaned survey’s welfare values using CPI, PPP, and population auxiliary data. It is not called automatically by pd_process_data() — it must be run afterwards, per survey, once cleaned data exists in PIP storage. There are two input modes:

# Mode A: pass a cleaned survey directly (aux auto-loaded from the master
# inventory when cpi/ppp/pop are NULL)
dt <- pipload::pip_read(id = "BOL_2022_EH_INC_ALL", alias = "pip")
bol_deflated <- pd_deflation(dt)

# Legacy Mode A: pass explicit aux tables instead of auto-loading them
ppp <- pipload::pip_load_aux("ppp")
cpi <- pipload::pip_load_aux("cpi")
pop <- pipload::pip_load_aux("pop")
bol_deflated <- pd_deflation(dt, cpi = cpi, ppp = ppp, pop = pop)

# Mode B: load the survey by id and deflate in one call
bol_deflated <- pd_deflation(pip_id = "BOL_2022_EH_INC_ALL")

In Mode A (dt supplied, cpi/ppp/pop left NULL), pd_deflation() resolves the matching metadata version for the survey from the master inventory and loads CPI/PPP/population automatically. In Mode B (pip_id supplied instead of dt), the survey itself is also loaded from stamp before deflation. pd_deflation() resolves the correct S3 method (deflation.pipmd() for micro data, deflation.pipgd() for group data) based on the survey’s class and applies it; deflation failures are caught and logged, returning NA for that survey rather than aborting.

{pipload} also provides load_pip_deflated_data(), a convenience wrapper around pd_deflation()’s Mode B: it locates a survey (by id_name or by filter arguments such as country_code/surveyid_year/module), loads it, and deflates it in one call — useful when you don’t already have the survey’s pip_id in hand:

bol_deflated <- pipload::load_pip_deflated_data(id_name = "BOL_2022_EH_INC_ALL")

# Or filter by country/year/module instead of a known id_name
bol_deflated <- pipload::load_pip_deflated_data(
  country_code  = "BOL",
  surveyid_year = 2022,
  module        = "ALL"
)

To deflate many surveys in a batch, use pd_deflate_pipeline(), the second pipeline stage (see below), which loops pd_deflation() over the master inventory.

`pd_deflate_pipeline()`: batch-deflate the whole inventory

pd_deflate_pipeline() is the batch orchestrator that iterates over the PIP master inventory of cleaned surveys, deflates each via pd_deflation() (simple Mode B, pip_id only), detects and skips failures (including NA returns), saves every deflated survey to the dedicated "pip_deflated" stamp alias, updates the master inventory with deflation columns, and logs a deflate_summary_inf summary entry:

new_pip_inv <- pd_deflate_pipeline(
  inv     = NULL, # NULL loads the master inventory internally
  force   = FALSE,
  verbose = TRUE
)

inv — the master inventory (as produced by pd_process_data() → build_pip_inventory()). When NULL (the default), it is loaded internally via pipload::load_pip_master_inventory().
force — when TRUE, every row is re-deflated regardless of the deflated column. When FALSE (the default), only surveys whose deflated is NA or FALSE are deflated.
verbose — controls verbosity of downstream pipload I/O calls.

Relationship to pd_process_data(): deflation is an independent second stage. pd_process_data() cleans surveys and creates metadata; it does not call pd_deflate_pipeline(). Run the pipeline wrapper explicitly after cleaning (e.g. in Pipdata_script.R). The "pip_deflated" stamp alias must be registered before the first run (e.g. via pipdata_dlw_process() or an explicit stamp::st_init() pointing at pip_repository/pip_deflated/).

On success, the returned inventory has deflated = TRUE, content_hash_deflated (the hash of the saved "pip_deflated" artifact), and the aux_cpi_hash_at_deflation, aux_ppp_hash_at_deflation, aux_pop_hash_at_deflation snapshots for each deflated survey. The "pipdata_log" receives a deflate_summary_inf entry with n_total, n_success, n_failed, surveys_success, and surveys_failed, which log_report() renders as a Deflation Summary section.

`log_report()`: summarize the `"pipdata_log"`

log_report() parses a piplog object and writes (or returns) a structured markdown report:

# log defaults to pipfun::log_filter(name = "pipdata_log") when not supplied
report <- log_report()

# Or write directly to a file
log_report(
  path      = file.path("log_reports", "log_report.md"),
  overwrite = TRUE
)

The report includes, when the corresponding logmeta entry is present in the log: latest-attempt DLW acquisition and validation summaries, stage-aware run warnings, run metadata (time window, totals), the processing summary (process_summary_inf), the deflation summary (deflate_summary_inf), auxiliary file changes (aux_changes_inf), a summary table by error/info type, a country-level breakdown of errors, inventory verification counts (inv_update_inf), skipped-survey details (skipped_svys_data/skipped_svys_metadata), and the list of surveys that failed processing (null_svys_inf). Sections are silently omitted when their logmeta entry is absent. Exact DLW completion entries are preferred; older logs use fallback confined to the selected latest attempt. Dedicated DLW sections separate invalid classifications from execution failures, and DLW discriminators are excluded from generic type/country sections to prevent stale history and duplicate reporting.

Scope: log_report() parses entries written under the "pipdata_log" name by pipdata_dlw_process(), pd_process_data()/process_data(), and pd_deflate_pipeline()/deflate_one(). The report is stage-aware and shows DLW acquisition/validation outcomes together with cleaning and deflation results. Auxiliary refresh logging from pipaux remains outside this report.

pd_change_report() applies the same completed-validation input guard before building its read-only dependency plan. This protects both execution and report planning when a legacy gmd_valid_inv still contains blank/unavailable retry rows from an earlier pipeline version.

pd_run_pipeline(): incremental top-level execution

pd_process_data(): clean surveys and build metadata