education service delivery indicators

education/completion_rates.py

@@ -0,0 +1,24 @@
+# Databricks notebook source
+# MAGIC %run ../utils
+
+# COMMAND ----------
+
+# Completion rate by education level (UIS CR.MOD modeled series). Pulled directly
+# from the UNESCO Institute for Statistics (UIS) API. Values are percentages (0-100).
+# Levels: .1 = primary, .2 = lower secondary, .3 = upper secondary.
+series_to_col_name = {
+    'CR.MOD.1': 'completion_rate_primary',
+    'CR.MOD.2': 'completion_rate_lower_secondary',
+    'CR.MOD.3': 'completion_rate_upper_secondary',
+}
+
+data_source = 'UNESCO Institute for Statistics (UIS)'
+
+# outer join so a country-year is kept even if it reports only some levels
+df = uis_fetch(series_to_col_name, data_source, how='outer')
+df
+
+# COMMAND ----------
+
+sdf = spark.createDataFrame(df)
+sdf.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable("prd_mega.indicator.completion_rates")

education/pupil_teacher_ratio.py

@@ -0,0 +1,33 @@
+# Databricks notebook source
+# MAGIC %run ../utils
+
+# COMMAND ----------
+
+# Pupil-teacher ratio by education level (World Bank / UNESCO Institute for Statistics)
+indicators = [
+    'SE.PRE.ENRL.TC.ZS',
+    'SE.PRM.ENRL.TC.ZS',
+    'SE.SEC.ENRL.TC.ZS',
+    'SE.SEC.ENRL.LO.TC.ZS',
+    'SE.SEC.ENRL.UP.TC.ZS',
+    'SE.TER.ENRL.TC.ZS',
+]
+col_names = [
+    'pupil_teacher_ratio_pre_primary',
+    'pupil_teacher_ratio_primary',
+    'pupil_teacher_ratio_secondary',
+    'pupil_teacher_ratio_lower_secondary',
+    'pupil_teacher_ratio_upper_secondary',
+    'pupil_teacher_ratio_tertiary',
+]
+
+data_source = 'UNESCO Institute for Statistics (UIS)'
+
+# outer join so a country-year is kept even if it reports only some education levels
+df = wbgapi_fetch(indicators, col_names, data_source, how='outer')
+df
+
+# COMMAND ----------
+
+sdf = spark.createDataFrame(df)
+sdf.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable("prd_mega.indicator.pupil_teacher_ratio")

education/school_basic_services.py

@@ -0,0 +1,34 @@
+# Databricks notebook source
+# MAGIC %run ../utils
+
+# COMMAND ----------
+
+# Proportion of schools with access to basic services / infrastructure, by
+# education level (UIS "school basic services", SDG 4.a.1). Pulled directly from
+# the UNESCO Institute for Statistics (UIS) API. Values are percentages (0-100).
+# Levels: .1 = primary, .2 = lower secondary, .3 = upper secondary.
+series_to_col_name = {
+    'SCHBSP.1.WELEC': 'schools_with_electricity_primary',
+    'SCHBSP.2.WELEC': 'schools_with_electricity_lower_secondary',
+    'SCHBSP.3.WELEC': 'schools_with_electricity_upper_secondary',
+    'SCHBSP.1.WINTERN': 'schools_with_internet_primary',
+    'SCHBSP.2.WINTERN': 'schools_with_internet_lower_secondary',
+    'SCHBSP.3.WINTERN': 'schools_with_internet_upper_secondary',
+    'SCHBSP.1.WCOMPUT': 'schools_with_computers_primary',
+    'SCHBSP.2.WCOMPUT': 'schools_with_computers_lower_secondary',
+    'SCHBSP.3.WCOMPUT': 'schools_with_computers_upper_secondary',
+    'SCHBSP.1.WWATA': 'schools_with_basic_water_primary',
+    'SCHBSP.2.WWATA': 'schools_with_basic_water_lower_secondary',
+    'SCHBSP.3.WWATA': 'schools_with_basic_water_upper_secondary',
+}
+
+data_source = 'UNESCO Institute for Statistics (UIS)'
+
+# outer join so a country-year is kept even if it reports only some indicators
+df = uis_fetch(series_to_col_name, data_source, how='outer')
+df
+
+# COMMAND ----------
+
+sdf = spark.createDataFrame(df)
+sdf.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable("prd_mega.indicator.school_basic_services")

education/teacher_salaries.py

@@ -0,0 +1,25 @@
+# Databricks notebook source
+# MAGIC %run ../utils
+
+# COMMAND ----------
+
+# Teachers' salaries by education level (UIS TSALARY series). Pulled directly
+# from the UNESCO Institute for Statistics (UIS) API.
+# Levels: .0 = pre-primary, .1 = primary, .2 = lower secondary, .3 = upper secondary.
+series_to_col_name = {
+    'TSALARY.0': 'teacher_salary_pre_primary',
+    'TSALARY.1': 'teacher_salary_primary',
+    'TSALARY.2': 'teacher_salary_lower_secondary',
+    'TSALARY.3': 'teacher_salary_upper_secondary',
+}
+
+data_source = 'UNESCO Institute for Statistics (UIS)'
+
+# outer join so a country-year is kept even if it reports only some levels
+df = uis_fetch(series_to_col_name, data_source, how='outer')
+df
+
+# COMMAND ----------
+
+sdf = spark.createDataFrame(df)
+sdf.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable("prd_mega.indicator.teacher_salaries")

indicator_data_availability_dlt.sql

@@ -102,6 +102,68 @@ OR REFRESH LIVE TABLE indicator_data_availability USING DELTA AS (
     GROUP BY
       1
   ),
+  pupil_teacher_ratio AS (
+    SELECT
+      country_name,
+      'pupil_teacher_ratio' AS indicator_key,
+      CAST(min(year) AS INT) AS earliest_year,
+      CAST(max(year) AS INT) AS latest_year
+    FROM
+      prd_mega.indicator.pupil_teacher_ratio
+    WHERE
+      COALESCE(pupil_teacher_ratio_pre_primary, pupil_teacher_ratio_primary, pupil_teacher_ratio_secondary, pupil_teacher_ratio_lower_secondary, pupil_teacher_ratio_upper_secondary, pupil_teacher_ratio_tertiary) IS NOT NULL
+    GROUP BY
+      1
+  ),
+  school_basic_services AS (
+    SELECT
+      country_name,
+      'school_basic_services' AS indicator_key,
+      CAST(min(year) AS INT) AS earliest_year,
+      CAST(max(year) AS INT) AS latest_year
+    FROM
+      prd_mega.indicator.school_basic_services
+    WHERE
+      COALESCE(
+        schools_with_electricity_primary, schools_with_electricity_lower_secondary, schools_with_electricity_upper_secondary,
+        schools_with_internet_primary, schools_with_internet_lower_secondary, schools_with_internet_upper_secondary,
+        schools_with_computers_primary, schools_with_computers_lower_secondary, schools_with_computers_upper_secondary,
+        schools_with_basic_water_primary, schools_with_basic_water_lower_secondary, schools_with_basic_water_upper_secondary
+      ) IS NOT NULL
+    GROUP BY
+      1
+  ),
+  teacher_salaries AS (
+    SELECT
+      country_name,
+      'teacher_salaries' AS indicator_key,
+      CAST(min(year) AS INT) AS earliest_year,
+      CAST(max(year) AS INT) AS latest_year
+    FROM
+      prd_mega.indicator.teacher_salaries
+    WHERE
+      COALESCE(
+        teacher_salary_pre_primary, teacher_salary_primary,
+        teacher_salary_lower_secondary, teacher_salary_upper_secondary
+      ) IS NOT NULL
+    GROUP BY
+      1
+  ),
+  completion_rates AS (
+    SELECT
+      country_name,
+      'completion_rates' AS indicator_key,
+      CAST(min(year) AS INT) AS earliest_year,
+      CAST(max(year) AS INT) AS latest_year
+    FROM
+      prd_mega.indicator.completion_rates
+    WHERE
+      COALESCE(
+        completion_rate_primary, completion_rate_lower_secondary, completion_rate_upper_secondary
+      ) IS NOT NULL
+    GROUP BY
+      1
+  ),
   all_indicators AS (
     SELECT * FROM hd_index
     UNION ALL
@@ -118,6 +180,14 @@ OR REFRESH LIVE TABLE indicator_data_availability USING DELTA AS (
     SELECT * FROM national_poverty
     UNION ALL
     SELECT * FROM edu_attendance
+    UNION ALL
+    SELECT * FROM pupil_teacher_ratio
+    UNION ALL
+    SELECT * FROM school_basic_services
+    UNION ALL
+    SELECT * FROM teacher_salaries
+    UNION ALL
+    SELECT * FROM completion_rates
   ),
   source_urls AS (
     SELECT * FROM (
@@ -129,7 +199,11 @@ OR REFRESH LIVE TABLE indicator_data_availability USING DELTA AS (
         ('pefa_by_pillar', 'https://www.pefa.org/assessments/batch-downloads'),
         ('health_private_expenditure', 'https://www.who.int/data/gho/data/indicators/indicator-details/GHO/out-of-pocket-expenditure-(oop)-per-capita-in-us'),
         ('poverty_rate', 'https://data360.worldbank.org/en/dataset/WB_PIP'),
-        ('global_data_lab_attendance', 'https://globaldatalab.org/education/about/')
+        ('global_data_lab_attendance', 'https://globaldatalab.org/education/about/'),
+        ('pupil_teacher_ratio', 'https://databrowser.uis.unesco.org/resources/glossary/3189'),
+        ('school_basic_services', 'https://databrowser.uis.unesco.org/resources/glossary/3145'),
+        ('teacher_salaries', 'https://databrowser.uis.unesco.org/resources/glossary/3218'),
+        ('completion_rates', 'https://databrowser.uis.unesco.org/resources/glossary/3201')
     ) AS t(indicator_key, source_url)
   )
   SELECT

utils.py

@@ -13,7 +13,11 @@ def get_volume_root_path():
 import pandas as pd
 from databricks.sdk.runtime import spark
 
-def wbgapi_fetch(indicators, col_names, data_source, extra_col_names_from_country_table=[]):
+def wbgapi_fetch(indicators, col_names, data_source, extra_col_names_from_country_table=None, how: str = 'inner'):
+    if extra_col_names_from_country_table is None:
+        extra_col_names_from_country_table = []
+    if how not in {'inner', 'outer', 'left', 'right'}:
+        raise ValueError(f"Unsupported merge how='{how}'")
     long_dfs = []
     for series, col_name in zip(indicators, col_names):
         df = wb.data.DataFrame(series, skipBlanks=True).reset_index()
@@ -25,7 +29,7 @@ def wbgapi_fetch(indicators, col_names, data_source, extra_col_names_from_countr
 
     merged_df = long_dfs[0]
     for df in long_dfs[1:]:
-        merged_df = pd.merge(merged_df, df, on=['economy', 'year'])
+        merged_df = pd.merge(merged_df, df, on=['economy', 'year'], how=how)
 
     merged_df['data_source'] = data_source
 
@@ -35,3 +39,57 @@ def wbgapi_fetch(indicators, col_names, data_source, extra_col_names_from_countr
 
     return df
 
+# COMMAND ----------
+
+import requests
+
+UIS_API_URL = 'https://api.uis.unesco.org/api/public/data/indicators'
+
+def uis_fetch(series_to_col_name, data_source, extra_col_names_from_country_table=None, how: str = 'inner', start: int = None, stop: int = None):
+    """Fetch indicators straight from the UNESCO Institute for Statistics (UIS) API.
+
+    Mirrors wbgapi_fetch's output (one row per country-year, one column per
+    indicator) so downstream scripts are interchangeable. The UIS geoUnit is an
+    ISO3 code, so it joins directly onto country.country_code; the join is inner,
+    which drops UIS regional aggregates (e.g. ECOWAS, World) and keeps only
+    countries.
+    """
+    if extra_col_names_from_country_table is None:
+        extra_col_names_from_country_table = []
+    if how not in {'inner', 'outer', 'left', 'right'}:
+        raise ValueError(f"Unsupported merge how='{how}'")
+    if not series_to_col_name:
+        raise ValueError("series_to_col_name must contain at least one indicator")
+    col_names = list(series_to_col_name.values())
+
+    params = [('indicator', ind) for ind in series_to_col_name]
+    if start is not None:
+        params.append(('start', start))
+    if stop is not None:
+        params.append(('stop', stop))
+    resp = requests.get(UIS_API_URL, params=params, timeout=60)
+    resp.raise_for_status()
+    payload = resp.json()
+    raw_df = pd.DataFrame.from_records(payload.get('records', []))
+    if raw_df.empty:
+        cols = ['country_name', 'country_code', 'region', *extra_col_names_from_country_table, 'year', *col_names, 'data_source']
+        return pd.DataFrame(columns=cols)
+
+    long_dfs = []
+    for series, col_name in series_to_col_name.items():
+        long_df = raw_df.loc[raw_df['indicatorId'] == series, ['geoUnit', 'year', 'value']]
+        long_df = long_df.rename(columns={'value': col_name}).dropna(subset=col_name)
+        long_dfs.append(long_df)
+
+    merged_df = long_dfs[0]
+    for df in long_dfs[1:]:
+        merged_df = pd.merge(merged_df, df, on=['geoUnit', 'year'], how=how)
+
+    merged_df = merged_df.astype({'year': 'int'})
+    merged_df['data_source'] = data_source
+
+    country_df = spark.table('indicator.country').select('country_name', 'country_code', 'region', *extra_col_names_from_country_table).toPandas()
+    df = pd.merge(merged_df, country_df, left_on='geoUnit', right_on='country_code', how='inner')[['country_name', 'country_code', 'region', *extra_col_names_from_country_table, 'year', *col_names, 'data_source']]
+
+    return df
+