synpop_tools

Databricks synpop generation utilities.

This module provides utility functions for the synpop generation process.

Requires: - databricks-sdk - time

Author: Gino F. Fazzi gino.franco.fazzi@audienceproject.com Facundo Fernández facundo.fernandez@audienceproject.com

synpop_tools.synpop_retrieve_available_countries(format='country', stage=None)

Retrieve the countries and versions available for synpop generation.

Reads available_countries.json and, when stage is given, narrows the result to the country-version pairs recorded in the approvals table.

Parameters:

Name	Type	Description	Default
format	str	Shape of the returned value. One of "country", "country,version", or "country,version,job".	'country'
stage	str | None	Optional approvals-table filter. "approved" keeps rows whose decision is approve, "ingested" keeps rows with any status, and "active" keeps rows whose status is active. When None the full curated list is returned without consulting the approvals table.	None

Returns:

Type	Description
list of str or dict	For "country", a sorted list of country codes. For "country,version", a sorted list of "Country|Version" strings. For "country,version,job", a nested mapping of country to version to job ID.

Raises:

Type	Description
Exception	If stage is not one of the accepted values, or if format is not one of the accepted values.

Examples:

>>> synpop_retrieve_available_countries(format="country,version", stage="active")
['DE|v1', 'DK|v2']

synpop_tools.synpop_get_widget_countries()

Return the curated country-version list formatted for a Databricks widget.

Returns:

Type	Description
list of str	Sorted "Country|Version" entries with a leading "(All)" option for use as dbutils.widgets choices.

synpop_tools.safe_download_tercet_files(file_url, output_path)

Download and validate a GISCO TERCET ZIP archive with cURL.

Parameters:

Name	Type	Description	Default
file_url		Path relative to the GISCO NUTS-2024 base URL.	required
output_path		Local directory in which to recreate file_url and store the archive.	required

Returns:

Type	Description
	None. An existing file with a ZIP signature is reused.

Raises:

Type	Description
RuntimeError	If cURL fails, GISCO returns HTML, or the downloaded file does not have a ZIP signature.

Notes

The cURL request emulates a browser, follows redirects, retries transient failures five times, and has a five-minute maximum duration.

synpop_tools.EurostatAPI

Eurostat API wrapper for retrieving European statistical datasets.

This class provides methods to interact with the Eurostat API, retrieve datasets, format them into tables, and execute the synpop extract process.

Attributes:

Name	Type	Description
BASE_URL	str	Base URL for the Eurostat API endpoint.
ENDPOINTS	dict	Mapping of table names to Eurostat dataset identifiers.
country_code	list	List of supported European country names.

Example

eurostat = EurostatAPI() df = eurostat.get_table("pjanedu")

synpop_tools.EurostatAPI.__init__()

Initialize the EurostatAPI with default configuration.

Sets up the base URL, endpoint mappings, and supported country codes.

synpop_tools.EurostatAPI.get_table(table, dtype='Pandas', labels=True, return_update_date=False, params=None)

Retrieve a Eurostat table and optionally return its update date.

Fetches data from the Eurostat API for the specified table, converts it to a DataFrame, and optionally returns the dataset's update timestamp.

Parameters:

Name	Type	Description	Default
table	str	The table name key (e.g., 'pjanedu', 'edat', 'd2jan', 'isoc', 'pjangrp3').	required
dtype	str	The output DataFrame type. Must be 'Pandas' or 'PySpark'. Defaults to 'Pandas'.	'Pandas'
labels	bool	Whether to use human-readable labels for dimension values. Defaults to True.	True
return_update_date	bool	If True, returns a tuple of (update_date, DataFrame). Defaults to False.	False
params	dict	Additional query parameters to pass to the Eurostat API. Defaults to None.	None

Returns:

Type	Description
	DataFrame or tuple: If return_update_date is False, returns a DataFrame of the requested table. If True, returns a tuple of (update_date: str, df: DataFrame).

Raises:

Type	Description
Exception	If an invalid dtype is provided.

Example

eurostat = EurostatAPI() df = eurostat.get_table("pjanedu", dtype="Pandas") update_date, df = eurostat.get_table("edat", return_update_date=True)

synpop_tools.EurostatAPI.get_endpoint_data(table, params)

Fetch raw JSON data from a Eurostat API endpoint.

Makes a GET request to the Eurostat API and returns the raw JSON response.

Parameters:

Name	Type	Description	Default
table	str	The table name key (e.g., 'pjanedu', 'edat', 'd2jan', 'isoc', 'pjangrp3').	required
params	dict	Query parameters to pass to the API. Defaults to an empty dict.	required

Returns:

Name	Type	Description
dict		The JSON response from the Eurostat API containing the dataset structure with dimensions, sizes, and values.

Raises:

Type	Description
HTTPError	If the API request fails.

Example

data = eurostat.get_endpoint_data("pjanedu", {"sex": "F"})

synpop_tools.EurostatAPI.format_JSON_stat_data_to_table(dataset, use_labels=True)

Convert Eurostat JSON-stat format to a Pandas DataFrame.

Takes a JSON-stat dataset object (which has a nested structure with dimensions, sizes, and values) and converts it to a flat, friendly Pandas DataFrame.

Parameters:

Name	Type	Description	Default
dataset	dict	The JSON-stat dataset object from the Eurostat API.	required
use_labels	bool	Whether to use human-readable labels for dimension values. If False, uses the raw dimension keys. Defaults to True.	True

Returns:

Type	Description
	pd.DataFrame: A DataFrame with columns for each dimension plus a 'value' column containing the data values.

Note

• Handles both dense (list) and sparse (dict) value representations
• Creates a MultiIndex from dimension combinations, then flattens it
• Dimension order is preserved from the original dataset

Example

data = eurostat.get_endpoint_data("pjanedu") df = eurostat.format_JSON_stat_data_to_table(data)

synpop_tools.EurostatAPI.execute_synpop_extract(s3_base_path, logger=None)

Execute the Eurostat synpop data extract and upload to S3.

Retrieves the latest Eurostat data for all configured tables, checks for updates against the stored metadata, and uploads new data to S3 if available.

This method: 1. Reads the current metadata from S3 to determine last update dates 2. Iterates through all Eurostat tables (pjanedu, edat, d2jan, isoc, pjangrp3) 3. Checks if new data is available (by comparing update dates) 4. Applies table-specific transformations and filters 5. Writes new data to S3 and updates the metadata file

Parameters:

Name	Type	Description	Default
s3_base_path	str	The base S3 path where Eurostat data should be stored (e.g., 's3://ap-synthetic-population/dev/raw_data/eurostat').	required
logger	Logger	A logger instance for tracking progress and reporting status. If not provided, logging is disabled. Defaults to None.	None

Raises:

Type	Description
Exception	If metadata cannot be read or if data retrieval fails.

Note

• Only tables with data newer than the stored metadata are updated
• Each table has specific filters applied (e.g., removing 'TOTAL' age)
• Metadata is updated in S3 after all processing completes

Example

eurostat = EurostatAPI() eurostat.execute_synpop_extract( ... s3_base_path="s3://bucket/eurostat", ... logger=my_logger ... )

synpop_tools.EurostatAPI.__archive_datapoint(table_name, update_date, s3_base_path)

Copy a Eurostat extract into its dated S3 archive location.

Parameters:

Name	Type	Description	Default
table_name		Eurostat table name used by the source CSV.	required
update_date		Datetime used to construct the archive filename.	required
s3_base_path		Root S3 path containing the eurostat directory.	required

Returns:

Type	Description
	None.

Notes

This method uses dbutils.fs.cp and therefore requires a Databricks runtime with access to the source and archive paths.

synpop_tools.NonEurostatHandler

Base handler for non-Eurostat census data sources.

This class provides a generic interface for handling census data from sources other than Eurostat. It manages file downloads, extraction, and S3 uploads for various country-specific data sources.

Attributes:

Name	Type	Description
root_path	str	The root directory for data storage.
data_path	str	The path to the data subdirectory.
country_code	str	The ISO country code for the data source.
datapoints	dict	Dictionary mapping datapoint names to their metadata (URL, download pattern, release dates, etc.).

Note

This is an abstract base class. Specific country implementations should inherit from it and define their own datapoints.

synpop_tools.NonEurostatHandler.__init__(root_path='/Workspace/dame/Synthetic Population')

Initialize the NonEurostatHandler with a root data path.

Parameters:

Name	Type	Description	Default
root_path	str	The root directory for data storage. Defaults to the DAME Synthetic Population Generation path.	'/Workspace/dame/Synthetic Population'

synpop_tools.NonEurostatHandler.set_root_path(path)

Set the root path for data storage and create data directory if needed.

Parameters:

Name	Type	Description	Default
path	str	The root directory path for data storage.	required

synpop_tools.NonEurostatHandler.get_available_datapoints()

Get a list of available datapoint names for this data source.

Returns:

Name	Type	Description
list		List of datapoint names that are configured for this source.

Raises:

Type	Description
AssertionError	If a datapoint name doesn't match its sanitized form.

synpop_tools.NonEurostatHandler.download_file_from_url(download_url, filename, verbose=False)

Download a file from a URL to the local data path.

Downloads a file from the specified URL and saves it to the data directory with the given filename. Uses streaming to handle large files efficiently.

Parameters:

Name	Type	Description	Default
download_url	str	The URL to download the file from.	required
filename	str	The name to save the file as (in data_path).	required
verbose	bool	If True, prints confirmation message. Defaults to False.	False

Raises:

Type	Description
HTTPError	If the download request fails.

Example

handler.download_file_from_url( ... "https://example.com/data.csv", ... "country-data.csv", ... verbose=True ... )

synpop_tools.NonEurostatHandler.upload_datapoint_to_s3(datapoint, s3_base_path)

Upload a downloaded datapoint to S3 storage.

Moves the local datapoint file from the data directory to the specified S3 path.

Parameters:

Name	Type	Description	Default
datapoint	str	The name of the datapoint to upload.	required
s3_base_path	str	The base S3 path where data should be stored.	required

Note

Uses dbutils.fs.mv to move the file, which copies it to S3 and removes the local copy.

Example

handler.upload_datapoint_to_s3( ... "population_data.csv", ... "s3://bucket/census/AU" ... )

synpop_tools.NonEurostatHandler.unzip_datapoint(datapoint, filename=None, only_open=False, delete_original=True)

Extract or open a downloaded zip file.

Helper method to unzip a downloaded zip file. If filename is not provided, all files are extracted to the folder. Otherwise, only the specified filename is retained in the root folder.

Parameters:

Name	Type	Description	Default
datapoint	str	The name of the datapoint (without .zip extension).	required
filename	str or int	The specific file to extract. If an integer is provided, extracts the file at that index in the zip archive. If None, extracts all files. Defaults to None.	None
only_open	bool	If True, opens the file in memory without extracting to disk. Defaults to False.	False
delete_original	bool	If True, deletes the original zip file after extraction. Defaults to True.	True

Returns:

Type	Description
	file or None: If only_open is True, returns the opened file object. Otherwise, returns None.

Raises:

Type	Description
AssertionError	If the datapoint is not a zip file type.

Example

Extract all files

handler.unzip_datapoint("population_data")

Extract specific file by name

handler.unzip_datapoint("population_data", filename="data.csv")

Extract file by index

handler.unzip_datapoint("population_data", filename=0)

synpop_tools.NonEurostatHandler.execute_synpop_extract(s3_base_path, logger=None)

Execute the synpop extract process for non-Eurostat data sources.

Retrieves the latest census data for all configured datapoints, checks for updates against the stored metadata, and uploads new data to S3 if available.

This method: 1. Reads the current metadata from S3 to determine last update dates 2. Iterates through all configured datapoints 3. Checks if new data is available (by comparing release dates) 4. Downloads, extracts, and uploads new data to S3 5. Updates the metadata file

Parameters:

Name	Type	Description	Default
s3_base_path	str	The base S3 path where census data should be stored (e.g., 's3://ap-synthetic-population/raw_data/census').	required
logger	Logger	A logger instance for tracking progress and reporting status. If not provided, logging is disabled. Defaults to None.	None

Raises:

Type	Description
Exception	If metadata cannot be read or if data retrieval fails.

Note

• Only datapoints with release dates newer than stored metadata are updated
• Sends Slack notifications on errors via send_slack_notification
• Creates reminders for manual downloads if needed via notify_manual_download

synpop_tools.NonEurostatHandler.notify_manual_download(datapoint, url)

Create a Slack reminder asking an operator to download a datapoint.

Parameters:

Name	Type	Description	Default
datapoint		Key into :attr:datapoints. Its next_release_date schedules the reminder, falling back to today when unset, and its optional instructions value is appended to the message.	required
url		Source URL included in the reminder for the operator to open.	required

Notes

Delegates to :class:jobs_tools.ReminderScheduler, so the reminder is persisted in Delta and delivered at-least-once rather than sent inline.

synpop_tools.NonEurostatHandler.__archive_datapoint(datapoint, update_date, s3_base_path)

Copy the current version of a datapoint into a timestamped archive folder.

Keeping the previous version lets an operator revert a bad refresh.

Parameters:

Name	Type	Description	Default
datapoint		Datapoint folder name below the country prefix.	required
update_date		Date used to build the archive folder name, formatted YYYYMMDD.	required
s3_base_path		Base S3 prefix containing the per-country datapoint folders.	required

Notes

Copies rather than moves, so the live datapoint remains in place for the caller to overwrite.

synpop_tools.GeneralDataSource

Bases: NonEurostatHandler

Retrieve global reference datasets that are not country-specific.

Attributes:

Name	Type	Description
country_code	str	Metadata key for global datasets, set to "general".
datapoints	dict	Dictionary mapping datapoint names to their source URLs, file types, and source release dates.

The UN source files are retained in their compressed CSV format. The ITU archive is reduced to its data CSV and saved as Parquet.

synpop_tools.GeneralDataSource.__init__(root_path='/Workspace/dame/Synthetic Population')

Initialize the global reference-source configuration.

Parameters:

Name	Type	Description	Default
root_path		Local directory used for staging downloaded files.	'/Workspace/dame/Synthetic Population'

synpop_tools.GeneralDataSource.get_datapoint_release_date(datapoint)

Return a source file's last-modified timestamp.

Parameters:

Name	Type	Description	Default
datapoint	str	Configured datapoint name.	required

Returns:

Type	Description
datetime	Timezone-aware source timestamp from the HTTP Last-Modified
datetime	header.

Raises:

Type	Description
RuntimeError	If the source does not provide a last-modified timestamp.
HTTPError	If the source cannot be reached successfully.

synpop_tools.GeneralDataSource.retrieve_datapoint(datapoint, verbose=False)

Download one configured global source file.

Parameters:

Name	Type	Description	Default
datapoint	str	Configured datapoint name.	required
verbose	bool	Whether to report the local staging path.	False

The ITU archive contains metadata CSVs in addition to the indicator data. Its configured data CSV is read and written as Parquet; the source archive is then removed. Other source files are retained in their original format.

Returns:

Type	Description
bool	True after a successful download and optional transformation.

synpop_tools.GeneralDataSource.process_zip_datapoint(datapoint)

Extract a configured ZIP CSV and write it as a Parquet file.

Parameters:

Name	Type	Description	Default
datapoint	str	Configured ZIP datapoint whose metadata specifies an archive_member_pattern and optional csv_skiprows.	required

Raises:

Type	Description
ValueError	If the archive does not contain exactly one matching data CSV.

Notes

The source ZIP is removed only after its selected CSV has been successfully converted to Parquet.

synpop_tools.AUDataSource

Bases: NonEurostatHandler

Australian Bureau of Statistics (ABS) data source handler.

This class handles data retrieval from the Australian Bureau of Statistics website. It inherits from NonEurostatHandler and provides country-specific configuration for Australian census and survey data.

Attributes:

Name	Type	Description
country_code	str	The ISO country code, set to "AU".
datapoints	dict	Dictionary mapping datapoint names to their metadata including URLs and download patterns.

Example

au_data = AUDataSource() au_data.execute_synpop_extract("s3://bucket/census")

synpop_tools.AUDataSource.__init__()

Initialize the AUDataSource with ABS data configuration.

Sets up the Australian data source with predefined datapoints for population, education, and internet usage data from ABS.

synpop_tools.AUDataSource.retrieve_datapoint(datapoint, verbose=False)

Retrieve a specific datapoint from the ABS website.

Downloads the data file for the specified datapoint from the ABS website and saves it locally.

Parameters:

Name	Type	Description	Default
datapoint		Name of the datapoint to retrieve; must be a key of :attr:datapoints.	required
verbose		Print progress messages while resolving the table URL.	False

Returns:

Type	Description
bool	Always True once the download completes. Failures surface as exceptions from the underlying request rather than False.

synpop_tools.AUDataSource.__get_html(url)

Fetch HTML content from a URL.

Parameters:

Name	Type	Description	Default
url	str	The URL to fetch content from.	required

Returns:

Name	Type	Description
str		The HTML content as text.

synpop_tools.AUDataSource.__get_soup(html)

Parse HTML content into a BeautifulSoup object.

Parameters:

Name	Type	Description	Default
html	str	The HTML content to parse.	required

Returns:

Name	Type	Description
BeautifulSoup		The parsed HTML object.

synpop_tools.AUDataSource.get_datapoint_reference_period(datapoint, verbose=False)

Extract the reference period (year) for a datapoint from the ABS website.

Parses the ABS webpage to find the current reference period/year for the specified datapoint.

Parameters:

Name	Type	Description	Default
datapoint	str	The name of the datapoint to look up.	required
verbose	bool	If True, prints the release year. Defaults to False.	False

Returns:

Name	Type	Description
str		The reference period year (e.g., "2023").

Raises:

Type	Description
ValueError	If the reference period block cannot be found.

synpop_tools.AUDataSource.get_table_url(datapoint, verbose=False)

Get the download URL for a specific datapoint.

Constructs the download URL by finding the appropriate download link on the ABS webpage based on the current reference period.

Parameters:

Name	Type	Description	Default
datapoint	str	The name of the datapoint to get URL for.	required
verbose	bool	If True, prints the download URL. Defaults to False.	False

Returns:

Name	Type	Description
str		The full URL to download the datapoint file.

synpop_tools.AUDataSource.get_datapoint_release_and_next_release_dates(datapoint, verbose=False)

Extract release and next release dates for a datapoint from the ABS website.

Parses the ABS webpage to find the release date and next scheduled release date for the specified datapoint.

Parameters:

Name	Type	Description	Default
datapoint	str	The name of the datapoint to look up.	required

verbose (bool, optional): If True, prints the dates found. Defaults to False.

Returns:

Name	Type	Description
tuple		A tuple of (released_date: str or None, next_release_date: str or None).

Raises:

Type	Description
ValueError	If the release date section cannot be found.

synpop_tools.CADataSource

Bases: NonEurostatHandler

Statistics Canada data source handler.

This class handles data retrieval from the Statistics Canada website (StatCan). It inherits from NonEurostatHandler and provides country-specific configuration for Canadian census data.

Attributes:

Name	Type	Description
country_code	str	The ISO country code, set to "CA".
datapoints	dict	Dictionary mapping datapoint names to their metadata including product IDs (pid) and data types.

Example

ca_data = CADataSource() ca_data.execute_synpop_extract("s3://bucket/census")

synpop_tools.CADataSource.__init__()

Initialize the CADataSource with Statistics Canada data configuration.

Sets up the Canadian data source with predefined datapoints for labour force, income, age, and internet usage data from StatCan.

synpop_tools.CADataSource.retrieve_datapoint(datapoint)

Retrieve a specific datapoint from Statistics Canada.

Downloads the data file for the specified datapoint from StatCan, extracts it if needed, and saves it locally.

Parameters:

Name	Type	Description	Default
datapoint		Name of the datapoint to retrieve; must be a key of :attr:datapoints.	required

Returns:

Type	Description
bool	Always True once the download completes. Failures surface as exceptions from the underlying request rather than False.

Notes

ZIP datapoints are read into pandas to convert them to Parquet, so this runs on the driver. The labour-force table is reduced by :meth:reduce_df_labour first because the raw file is roughly 5 GB.

synpop_tools.CADataSource.__get_pid_table_csv_link(pid)

Get the download link for a Statistics Canada product ID.

Parameters:

Name	Type	Description	Default
pid	str	The Statistics Canada product ID.	required

Returns:

Name	Type	Description
str		The download URL for the CSV file.

synpop_tools.CADataSource.__get_pid_release_date(pid)

Get the release date for a Statistics Canada product ID.

Parameters:

Name	Type	Description	Default
pid	str	The Statistics Canada product ID.	required

Returns:

Name	Type	Description
str		The release date in ISO format.

synpop_tools.CADataSource.reduce_df_labour(df)

Reduce the labour force DataFrame to essential columns.

Special method to reduce the labour force DataFrame to only the columns needed for synpop generation, following Villads filtering.

Parameters:

Name	Type	Description	Default
df	DataFrame	The full labour force DataFrame.	required

Returns:

Type	Description
	pd.DataFrame: The reduced DataFrame with only essential columns.

synpop_tools.MXDataSource

Bases: NonEurostatHandler

Retrieve and track Mexican SynPop source datapoints.

The handler combines manually downloaded INEGI census tables, an ITU internet-usage archive, and the Correos de México postal-code mapping. Release dates for manual census inputs are estimated on a ten-year cycle.

synpop_tools.MXDataSource.__init__()

Initialize the Mexican data source and its datapoint metadata.

Release and next-release dates are calculated for every configured datapoint during initialization.

synpop_tools.MXDataSource.retrieve_datapoint_release_date(datapoint)

Return estimated release dates for a Mexican datapoint.

Parameters:

Name	Type	Description	Default
datapoint		Key from self.datapoints.	required

Returns:

Type	Description
	Pair (release_date, next_release_date). Manual inputs use the
	surrounding decade boundaries; other inputs use today with no
	scheduled next release.

synpop_tools.MXDataSource.retrieve_datapoint(datapoint)

Ensure a Mexican source datapoint is available locally.

Parameters:

Name	Type	Description	Default
datapoint		Key from self.datapoints.	required

Returns:

Type	Description
	True when the datapoint is available or downloaded; False when a
	required manual file is missing and a notification was sent.

Notes

ZIP downloads are extracted and their original archives removed. The postal-code source uses special_download_postal_codes.

synpop_tools.MXDataSource.special_download_postal_codes()

Download the Correos de México all-state postal-code archive.

Returns:

Type	Description
	None. The ZIP archive is written beneath self.data_path.

Notes

The source requires ASP.NET form state and a simulated download button POST. On failure, the error is printed and the datapoint is changed to manual-download mode.

synpop_tools.GBDataSource

Bases: NonEurostatHandler

Retrieve and track Great Britain and Northern Ireland SynPop sources.

Supported inputs include ArcGIS GeoJSON services, downloadable spreadsheets and ZIP archives, manually supplied Ofcom data, and paginated Nomis census datasets.

synpop_tools.GBDataSource.__init__(verbose=False)

Initialize the UK data source and its datapoint metadata.

Parameters:

Name	Type	Description	Default
verbose		If True, print each datapoint while its release dates are resolved.	False

Notes

Release and next-release dates are calculated for every configured datapoint during initialization.

synpop_tools.GBDataSource.retrieve_datapoint(datapoint, verbose=False)

Retrieve one configured UK datapoint into the local data directory.

Parameters:

Name	Type	Description	Default
datapoint		Key from self.datapoints.	required
verbose		If True, print download or Nomis pagination progress.	False

Returns:

Type	Description
	True after a successful automated retrieval; False when the datapoint
	requires a manual download and a notification was sent.

Notes

ArcGIS GeoJSON is converted to a shapefile bundle, ZIP archives are extracted, and Nomis pages are concatenated into a CSV.

synpop_tools.GBDataSource.retrieve_datapoint_release_date(datapoint)

Discover the release schedule for a configured UK datapoint.

Parameters:

Name	Type	Description	Default
datapoint		Key from self.datapoints.	required

Returns:

Type	Description
	Pair (release_date, next_release_date) derived from the appropriate
	ArcGIS, NISRA, National Records of Scotland, education-statistics, Nomis,
	or annual Technology Tracker metadata source.

Raises:

Type	Description
Exception	If the metadata URL has no supported date handler.

Notes

Sources without exposed update metadata can fall back to the current date. Some metadata pages can return None when a release marker is unavailable.

synpop_tools.GBDataSource.retrieve_nomis_dataset(url, verbose=False)

Download every page of a Nomis CSV dataset.

Parameters:

Name	Type	Description	Default
url		Nomis URL template containing a {current_row} placeholder.	required
verbose		If True, print row-range progress.	False

Returns:

Type	Description
	pandas DataFrame containing the concatenated pages.

Raises:

Type	Description
HTTPError	If a page request is unsuccessful.
AssertionError	If Nomis returns a non-CSV or empty response.

Notes

Nomis is queried in blocks of 25,000 rows. RECORD_COUNT from the first returned row controls pagination.

synpop_tools.Geo2NUTSGetter

Class to get the mapping between geographical codes (postal codes, etc.) and NUTS codes for European countries, using the TERCET files from Eurostat.

The TERCET files are available for download at https://gisco-services.ec.europa.eu/tercet/NUTS-2024/, and contain the mapping between NUTS codes and various geographical codes for each country.

Author: Gino F. Fazzi (gino.franco.fazzi@audienceproject.com)

synpop_tools.Geo2NUTSGetter.__init__(base_path, logger=None, countries=None)

Configure a Geo-to-NUTS TERCET downloader.

Parameters:

Name	Type	Description	Default
base_path		Destination root for downloaded and processed TERCET files.	required
logger		Optional logger receiving download and processing messages.	None
countries		Optional iterable of country codes. If omitted, the supported default set of European countries is used.	None

synpop_tools.Geo2NUTSGetter._tercet_file_exists(url)

Return whether a TERCET file URL points to a downloadable file.

TERCET sometimes serves transient server/WAF errors for valid files, especially on HEAD requests. Retryable errors are treated as inconclusive instead of missing so the download step can use its curl fallback with retries.

Parameters:

Name	Type	Description	Default
url	str	Absolute TERCET file URL to probe.	required

Returns:

Type	Description
bool	True when the URL responds with 200 and a non-HTML content type, or when the status is retryable and therefore inconclusive. False when the server answers with a non-retryable, non-200 status or returns HTML, which TERCET serves for missing files.

Raises:

Type	Description
RuntimeError	If the probe request fails at the transport level.

Notes

Probes with HEAD first and retries with a streaming GET for statuses in :attr:TERCET_GET_FALLBACK_STATUS_CODES, since TERCET rejects HEAD for some valid files.

synpop_tools.Geo2NUTSGetter.download_tercet_files()

Download and process the newest available TERCET mapping for each country.

Raises:

Type	Description
FileNotFoundError	If none of the supported TERCET versions exists for a country.

synpop_tools.Geo2NUTSGetter.cleanup()

Remove the temporary TERCET download directory.

Returns:

Type	Description
	None.

Raises:

Type	Description
FileNotFoundError	If ./temp does not exist.

synpop_tools.Geo2NUTSGetter.update_Geo2NUTS(logger=None)

Download, process, and clean up all configured Geo-to-NUTS mappings.

Parameters:

Name	Type	Description	Default
logger		Optional logger receiving processing progress.	None

Returns:

Type	Description
	None.

synpop_tools.Approver

Class to get produce the approval requests for finished Synthetic Populations, and send them to the relevant stakeholders for review.

Author: Gino F. Fazzi (gino.franco.fazzi@audienceproject.com)

synpop_tools.Approver.__init__()

Initialize the Slack approval client.

Notes: The bot token is loaded from the Databricks dame secret scope, and approval requests are sent to the configured SynPop approval channel.

synpop_tools.Approver.__construct_payload(approval_id, synpop_id, report_url)

Build the interactive Slack payload for an approval request.

Parameters:

Name	Type	Description	Default
approval_id		Unique identifier associated with the approval request.	required
synpop_id		Synthetic-population identifier. Its first hyphen-delimited component is displayed as the country code.	required
report_url		URL of the report stakeholders should review.	required

Returns:

Type	Description
	Slack message payload containing Approve and Reject buttons. Each
	button carries the approval and synthetic-population identifiers as
	JSON metadata.

synpop_tools.Approver.generate_approval_request(synpop_id, report_url)

Post an interactive SynPop approval request to Slack.

Parameters:

Name	Type	Description	Default
synpop_id		Identifier of the synthetic population awaiting approval. Carried through the Slack action payload so the callback can resolve it.	required
report_url		Review-report URL included in the Slack message.	required

Returns:

Type	Description
None	The request is posted to Slack rather than returned.

Notes

The message contains Approve and Reject actions carrying a freshly generated approval ID together with synpop_id. Authentication uses the bot token loaded by the class constructor.

synpop_tools.SynPopAPIHandler

Class to handle the API calls for Synthetic Population ingestion and validation.

Author: Gino F. Fazzi (gino.franco.fazzi@audienceproject.com)

synpop_tools.SynPopAPIHandler.__init__(env, token=None, logger=None)

Initialize a SynPop API client for an environment.

Parameters:

Name	Type	Description	Default
env		API environment accepted by :func:get_environment_path_population_api (prod, dev, or qa).	required
token		Optional bearer token. When omitted, the token is read from the Databricks dame secret scope under the synpop_api_bearer key.	None
logger		Optional logger retained for callers that need to attach process logging.	None

Raises:

Type	Description
Exception	If env is not a recognized environment, or if no token is given and the Databricks secret cannot be read.

Notes

Authentication headers are built during initialization, so constructing this class requires either a token or secret-scope access.

synpop_tools.SynPopAPIHandler.whoami()

Return the identity associated with the configured API bearer token.

Returns:

Type	Description
	JSON-decoded identity response.

Raises:

Type	Description
Exception	If the API does not return HTTP 200.

synpop_tools.SynPopAPIHandler._post_ingest_population(payload)

Submit a population-ingestion payload to the SynPop API.

Parameters:

Name	Type	Description	Default
payload		JSON-serializable request body for the population ingestion endpoint.	required

Returns:

Type	Description
	JSON-decoded API response.

Raises:

Type	Description
Exception	If the ingestion endpoint does not return HTTP 201.

synpop_tools.SynPopAPIHandler._generate_ingestion_path(country_code, version, date, name)

Build the production S3 path for a formatted population.

Parameters:

Name	Type	Description	Default
country_code		Population country partition.	required
version		Population version partition.	required
date		Three-element [year, month, day] partition date.	required
name		Population name partition.	required

Returns:

Type	Description
	S3 URI below the production ingested prefix.

synpop_tools.SynPopAPIHandler._construct_payload_for_ingestion(country_code, version, date, name)

Build an ingestion payload and ensure its population is formatted.

Parameters:

Name	Type	Description	Default
country_code		Population country partition.	required
version		Population version partition.	required
date		Three-element [year, month, day] partition date.	required
name		Alias and S3 name partition for the ingested population.	required

Returns:

Type	Description
	SynPop API ingestion payload containing the S3 location and column
	metadata.

Notes

Geographic columns are read from the matching approved metadata. If the formatted ingestion path does not exist, the approved population is formatted and written there before the payload is returned.

synpop_tools.SynPopAPIHandler.ingest_population(country_code, version, date, name=None, force=False, verbose=False)

Register an approved synthetic population with the SynPop API.

Parameters:

Name	Type	Description	Default
country_code		Population country code.	required
version		Population version and API alias.	required
date		Three-element [year, month, day] partition date or a datetime.date instance.	required
name		Optional population name and S3 name partition. Defaults to "{country_code}_{version}".	None
force		Re-submit even when a response.json already exists for this country, version, and date. Without it, the existing response is returned unchanged.	False
verbose		Print the resulting population ID and response.	False

Returns:

Type	Description
dict	JSON-decoded API response for the created population, or the previously stored response when one exists and force is false.

Raises:

Type	Description
Exception	If the ingestion endpoint does not return HTTP 201, or if the population contains a column without known API metadata.

Notes

Writes payload.json and response.json beside the ingested population in S3. The duplicate check keys on country, version, and date but not name, so a given partition can only be ingested once unless force is set.

synpop_tools.SynPopAPIHandler.activate_population(population_id, activation_date)

Activate a synthetic population in the SynPop API.

Parameters:

Name	Type	Description	Default
population_id		Population ID returned by the ingestion endpoint.	required
activation_date		Activation date string formatted as YYYY-MM-DD.	required

Returns:

Type	Description
dict	JSON-decoded API response for the activated population.

Raises:

Type	Description
Exception	If the activation endpoint does not return HTTP 200.

synpop_tools.SynPopAPIHandler.get_population(country_code=None, date=None)

Query available synthetic populations.

Parameters:

Name	Type	Description	Default
country_code	str | None	Optional country code filter. Sent to the API as country.	None
date	str | None	Optional population date filter. String formatted date YYYY-MM-DD	None

Returns:

Type	Description
dict	JSON response from the population query endpoint.

Raises:

Type	Description
Exception	If the query endpoint does not return HTTP 200.

Notes

API reference: http://synthetic-population-prod.dk-prod.ap.priv/api-docs/index.html#/population/getQueriedPopulations

synpop_tools.SynPopAPIHandler.get_population_by_id(population_id)

Retrieve one synthetic population by its API ID.

Parameters:

Name	Type	Description	Default
population_id		Population ID produced when the population was ingested.	required

Returns:

Type	Description
dict	JSON response describing the requested population.

Raises:

Type	Description
Exception	If the endpoint does not return HTTP 200, including when no population matches population_id.

Notes

API reference: http://synthetic-population-prod.dk-prod.ap.priv/api-docs/index.html#/population/getPopulation

synpop_tools.SynPopAPIHandler._get_column_info(geo_columns, columns_to_include)

Convert population columns into SynPop API column metadata.

Parameters:

Name	Type	Description	Default
geo_columns		Geographic column names for this population. Each is declared categorical and marked as a target-granularity column.	required
columns_to_include		Population column names to describe. id is skipped; every other name must be either a known column or listed in geo_columns.	required

Returns:

Type	Description
list of dict	One entry per column with its label, columnType, and, for universe columns, the universeAliases the engine accepts.

Raises:

Type	Description
Exception	If a requested column has no known API metadata. Adding a new population column requires registering it here first.

synpop_tools.SynPopAPIHandler._get_headers(token)

Build authenticated JSON headers for the SynPop API.

Parameters:

Name	Type	Description	Default
token		Bearer token to use. When falsy, the token is read from the Databricks dame secret scope under the synpop_api_bearer key.	required

Returns:

Type	Description
dict	Headers containing the JSON content type and the bearer-token authorization value.

Raises:

Type	Description
Exception	Re-raises the underlying secrets error if no token is supplied and the Databricks secret cannot be read.

synpop_tools.get_population_by_id(population_id, env='prod', token=None)

Query available synthetic populations.

Parameters:

Name	Type	Description	Default
population_id			required
env			'prod'
token			None

Returns:

Name	Type	Description
	dict	JSON response from the population query endpoint.
NOTE	Documentation http://synthetic-population-prod.dk-prod.ap.priv/api-docs/index.html#/population/getPopulation

synpop_tools.get_environment_path_population_api(env)

Return the synthetic-population API base URL for an environment.

Author: Christian Starup (christian@audienceproject.com)

Parameters:

Name	Type	Description	Default
env		Target environment. One of dev, qa, or prod.	required

Returns:

Type	Description
str	Base URL for the requested environment, with a trailing slash.

Raises:

Type	Description
Exception	If env is not one of the three recognized environments.

synpop_tools.format_population_for_engine_upload(population_load_path, population_save_path, country_code, columns_to_include)

Rewrite a Data Science synthetic population into the Engine upload format.

Compared with the Data Science output, the Engine format drops the freq_ column prefix and exposes exactly one column per universe, computed as the sum of the hard-coded channel columns that make up that universe. The implementation tolerates any Data Science population version.

Parameters:

Name	Type	Description	Default
population_load_path		S3 path of the Parquet synthetic population to read.	required
population_save_path		S3 path to overwrite with the reformatted population.	required
country_code		Country code driving the minimum-age filter and the geographic column mapping. GB keeps ages 18 and above and maps nuts1-nuts3 to itl1-itl3; every other country keeps ages 16 and above.	required
columns_to_include		Engine column names to emit. Must be a subset of the allowed column list and must contain id.	required

Returns:

Type	Description
None	The reformatted population is written to population_save_path rather than returned.

Raises:

Type	Description
Exception	If columns_to_include contains a name outside the allowed list, if a required column is missing from the source, if more than one row exists per person, if a required column contains nulls, or if a panel or ownership column holds a value outside its permitted set.

Notes

Writes with mode("overwrite"), so any existing data at population_save_path is replaced. Validation runs several counting actions over the population, so this is deliberately not a lazy transformation.

synpop_tools.load_approvals_table(env='prod')

Load the synthetic-population approvals Delta table for an environment.

Parameters:

Name	Type	Description	Default
env		Environment prefix below the ap-synthetic-population bucket, for example prod.	'prod'

Returns:

Type	Description
DataFrame	Spark DataFrame over the approvals Delta table.

synpop_tools.get_attribute_from_synpop_table(country_code, version, attribute, stage='active', max_date=None)

Return one attribute from the latest eligible synthetic-population row.

Parameters:

Name	Type	Description	Default
country_code		Country code identifying the synthetic population.	required
version		Population version to select.	required
attribute		Approval-table column or derived s3_path to return.	required
stage		Required lifecycle stage: approved, ingested, or active.	'active'
max_date		Latest allowed release date as a date string or (year, month, day).	None

Returns:

Type	Description
Any	Value of attribute from the selected approval-table row.

Raises:

Type	Description
Exception	If an argument is invalid or no matching population row is available.

synpop_tools.write_to_approval_table(step, env, args)

Write an approval workflow event to the approval's Delta table.

Parameters

step: Workflow step to apply: "decision", "ingestion", or "activation". env: Environment partition used in the approvals-table S3 path. args: Values required by the selected step. Decision events require synpop_id, creation_date, approval_id, decision, and approved_by; ingestion updates require synpop_id, population_id, and status; activation updates require synpop_id and an activation_date formatted as YYYY-MM-DD.

Returns

None The event is persisted to Delta rather than returned.

Raises

ValueError If step is not one of the three supported workflow steps. KeyError If args is missing any argument required by step.

Notes

Decision events are appended to the Delta table. Ingestion and activation events update matching rows by synpop_id and do not insert unmatched rows, so an activation for an unknown synpop_id succeeds silently without changing anything. Requires Spark, Delta Lake, and write access to the target S3 location.

synpop_tools.push_synthethic_population(country_code, version, approver='', synpop_id='', overwrite=False, debug=False)

Promote the latest reviewed synthetic population to approved storage.

Parameters:

Name	Type	Description	Default
country_code	str	Country partition of the reviewed population.	required
version	str	Version partition of the reviewed population.	required
approver	str	Approver value recorded in the metadata manifest.	''
synpop_id	str	Synthetic-population identifier recorded in the manifest.	''
overwrite	bool	If True, remove existing approved targets before copying. If False, existing targets cause an exception.	False
debug		If True, perform discovery and validation without deleting, copying, or writing the manifest.	False

Returns:

Type	Description
	None.

Raises:

Type	Description
FileNotFoundError	If a required reviewed data or metadata source is missing.
Exception	If an approved destination exists and overwrite is False.

Notes

Only the artifact=synthetic_population data artifact is promoted. All items in the matching latest metadata partition are copied, and a manifest.json file is written alongside them. The operation uses dbutils.fs and production S3 paths.

Author: Christian Starup (christian@audienceproject.com) [Copied from synthetic-population-generation/synpop_general_functions and adapted to use dsr_tools instead of deprecated working_tools_edit]

synpop_tools.load_synpop_survey(country_code, max_date=None)

Load the synthetic-population survey for a country.

Parameters:

Name	Type	Description	Default
country_code		Country whose survey extracts should be searched.	required
max_date		Optional inclusive upper bound as an ISO YYYY-MM-DD string. When given, the newest extract-YYYYMMDD folder at or before this date is loaded. When omitted, the latest folder is loaded instead.	None

Returns:

Type	Description
DataFrame	Survey responses. Read as Parquet from the latest folder, or as header-bearing CSV when resolved through max_date.

Raises:

Type	Description
Exception	If the country has no survey directory, if latest is missing, if max_date cannot be parsed as an ISO date, or if no extract exists at or before max_date.

Notes

The two branches return different file formats, so callers that pass max_date receive all-string CSV columns rather than the typed Parquet schema returned by the default branch.

synpop_tools.load_demographic_weights(country_code, version, date=None)

Load the demographic weights produced by the latest approved SynPop flow.

Parameters:

Name	Type	Description	Default
country_code		Country whose weights should be loaded.	required
version		Population version to resolve.	required
date		Optional inclusive upper bound on the population creation date, accepted as a YYYY-MM-DD string or a (year, month, day) tuple. Defaults to today, so the most recent active population is used.	None

Returns:

Type	Description
DataFrame	Spark DataFrame of demographic weights for the resolved population.

Raises:

Type	Description
Exception	If no active population matches the country, version, and date bound, or if the resolved population has no demographic_weights artifact.