Get attribute.

Append child element.

Find element by path.

List files in the given path. Args: path: Path to list files from pattern: Optional glob pattern to filter files

Get or create a standardized logger for a pipeline. This function provides a simple way to get a consistently configured logger for any

Set attribute.

Test implementation of the run method.

Parse XML string.

OPTIMAL: Query parquet and create DuckDB table directly - NO DataFrame conversion. This is the recommended method for maximum perfor

Close database connection.

Resolve the latest object matching an arbitrary R2 glob.

OPTIMIZED: Upload JSON data with streaming via s3fs with retry logic. Uses streaming approach for optimal performance: - No

Create a DuckDB table directly from storage parquet file.

Upload DuckDB table directly to cloud storage without DataFrame conversion. Supports both Parquet and CSV formats based on file extension.

Get or create a configured logger instance with the specified log level. Creates a singleton logger instance if it doesn't exist yet

Close database connection.

Close the DuckDB connection.

Close database connection.

Save data using unified cloud storage access layer. ✅ MIGRATION: Now uses StorageAccess for optimal performance with shared connecti

Get path to latest gold data file for a given dataset.

Resolve the latest parquet object for a timestamped dataset prefix. Most gold datasets live under: `<prefix>/<timestamp>/<filename>`

Validate and transform geometries using DuckDB-spatial for optimal performance. This function validates geometries in a DuckDB table, fixes

Execute SQL and return results as list of dicts.

Context manager for temporary file download with guaranteed cleanup.

Check current resource usage with GitHub Actions compatibility.

Ensure stdout and stderr are flushed.

Get cached s3fs filesystem instance for Cloudflare R2. Requires R2_ACCESS_KEY_ID, R2_SECRET_ACCESS_KEY, and R2_ACCOUNT_ID environment va

Buffer raw data for later consolidated export.

Generate a ranking from SQL that returns cvr_number, company_name, municipality, value.

Create complete metadata for a dataset

Extract a single record dict from a detail HTML page. `slug_to_name` maps `kontrol_tag` slugs (e.g. ``"ikast-brande"``) to their display name

OPTIMIZED: Download JSON data with streaming via s3fs. Uses streaming approach for optimal performance: - No temp file creat

Perform chunked spatial join with 5-stage pipeline.

Set up standardized logging for pipeline operations. This replaces the inconsistent logging setups found across different pipelines with

Connect to DuckDB.

Find all elements by path.

Create output directories if they don't exist.

Create sub-element.

Normalize a CVR number to 8-digit string format. Handles: - XML hex encoding from FVM Excel exports (e.g., "_x0034_" → "4") - Float

Detect CRS from coordinate bounds. Analyzes coordinate ranges to determine the likely CRS and coordinate order. Args: min_x: Mi

Validate geometries and normalize to EPSG:25832 (Danish UTM) for processing. This function follows the OPTIMAL CRS strategy: - Keep data

Return SQL subquery for application intensity per GRUKO. intensity_type: 'fluorinated' — sum of fluorinated-only ingredients (for TFA)

Save the gold data to cloud storage using DuckDB's COPY command.

Parse a GML geometry element into a WKT string using DuckDB-spatial. This method extracts coordinates from a GML posList element and

Geocode an address using its DAWA address ID. Args: adresse_id: UUID address identifier from CVR Returns:

Fetch comprehensive company data for a specific CVR number. Args: cvr_number: 8-digit CVR number fetch_all_f

Close database connection.

Main orchestrator for CHR gold layer processing. Args: export_timestamp: Export timestamp for file naming silver_dir: Path t

Write JSON to local output dir or R2. Args: data: JSON-serializable data path: Relative path under api/v1/, e.g. "hom

Create a temporary file and track it for cleanup. Args: suffix: File suffix/extension prefix: File prefix Re

Save stage output to cloud storage using optimized export. Args: table_name: DuckDB table name to export out

Strip protocol prefix from a storage path, returning a bare bucket/key string. Handles ``r2://`` and ``s3://`` prefixes. Bare paths (no prefix)

Load a parquet file into a DuckDB table. Args: parquet_path: Path to parquet file (cloud or local) table_name: Name f

Load parquet file(s) from R2 prefix into a DuckDB table. Returns row count.

Mock upload JSON.

Ground deposition as % of applied dose at given distance (arable crops). Uses the Rautmann power-law: D(d) = A × d^B Clamped to minimum 1m di

Find the first matching column name from candidates.

Find timestamped directories ranked by recency (newest first). Args: base_path: Base storage path like r2://bucket/silver/dataset

Return SQL subquery for application intensity per GRUKO for a given substance. Shared by run_correlations(), run_temporal_lag(), and run_multivar

Parse an XML feature element into a feature dictionary with WKT geometry. This method extracts geometry and attributes from an XML f

Read data from the bronze layer, preferring in-memory data if available. This method first checks if bronze_data is provided (in-mem

Geocode an address using its DAWA address ID with caching. Args: adresse_id: UUID address identifier from CVR R

Get FVM username, password, VetStat certificate, and private key for robust authentication.

Setup cloud storage authentication for a DuckDB connection. Tries R2 credentials first (TYPE r2), falls back to legacy cloud HMAC (TYPE GCS).

Replicate silver's transform_dma_json using pure DuckDB. Silver uses pyarrow → DuckDB. We skip pyarrow and use DuckDB JSON directly to test t

Rename columns according to conventions using SQL.

Safely drop a table if it exists.

Create the common GLRCHRWSInfoInbound structure.

Get the configured logger. Returns: Logger instance

Load parquet file(s) from R2 prefix into a DuckDB table. Returns row count.

Capture index + taxonomy, then download up to `limit` detail pages. Returns a manifest dict (also written to metadata.json).

Update output dataset names to include year suffix.

Create pending pesticide rows table with nopesticides=1 records filtered out.

Create the disaggregated results table with original schema plus municipality support.

Get current memory usage in GB.

Load a silver dataset into DuckDB using the standard BaseSource method. Args: dataset_name: Name of the dataset (e.g., '

Add a GHG sub-source emission to this category.