Debby
1675 lines
79 KiB
Python
1675 lines
79 KiB
Python
"""
|
|
BIGQUERY ANALYSIS LAYER - FOOD SECURITY AGGREGATION
|
|
|
|
PERUBAHAN ARSITEKTUR:
|
|
- ASEAN aggregate DIGABUNG ke dalam tabel yang sama dengan negara-negara,
|
|
menggunakan country_name = "ASEAN" dan country_id = 0.
|
|
Looker Studio dapat memfilter: semua negara, per negara, atau ASEAN saja.
|
|
- 3 tabel DIHAPUS (digantikan oleh filter di tabel gabungan):
|
|
* agg_pillar_composite -> cukup filter country_name = "ASEAN" di agg_pillar_by_country
|
|
* agg_framework_asean -> cukup filter country_name = "ASEAN" di agg_framework_by_country
|
|
* agg_narrative_overview -> cukup filter country_name = "ASEAN" di agg_narrative_pillar
|
|
- "Sustainability" diganti "Other" di seluruh mapping pilar.
|
|
|
|
Output 3 tabel ke fs_asean_gold:
|
|
1. agg_pillar_by_country (termasuk baris ASEAN per pillar per year)
|
|
2. agg_framework_by_country (termasuk baris ASEAN per framework per year)
|
|
3. agg_narrative_pillar (termasuk baris ASEAN per pillar per year)
|
|
|
|
Narrative style:
|
|
- Plain text, tanpa markdown bold (**)
|
|
- Interpretatif: membaca tren, gap, anomali, konsistensi dari data nyata
|
|
- Bilingual: narrative_en (Inggris) + narrative_id (Indonesia)
|
|
|
|
KONDISI PILAR (pillar_condition_en / pillar_condition_id):
|
|
Kolom tambahan di agg_pillar_by_country untuk mendeskripsikan kondisi
|
|
tiap pilar per negara per tahun secara kontekstual dan kuantitatif.
|
|
|
|
Landasan teori:
|
|
1. FAO & CFS (1996 World Food Summit; CFS Reform Document 2009):
|
|
Definisi 4 pilar ketahanan pangan dan makna substantif masing-masing.
|
|
Referensi: FAO (2009). "Declaration of the World Summit on Food Security."
|
|
CFS (2012). "Global Strategic Framework for Food Security & Nutrition."
|
|
2. GFSI — Economist Impact (2022):
|
|
Threshold klasifikasi skor 0-100:
|
|
>= 75 : "Good" environment -> label "Secure / Aman"
|
|
>= 60 : above threshold -> label "Adequate / Memadai"
|
|
>= 40 : "Moderate" env -> label "Moderate / Sedang"
|
|
>= 20 : below moderate -> label "At Risk / Berisiko"
|
|
< 20 : severe -> label "Critical / Kritis"
|
|
Referensi: Economist Impact (2022). "Global Food Security Index 2022."
|
|
3. IPC — Integrated Food Security Phase Classification (2019):
|
|
Klasifikasi bertingkat per pilar: dari "Moderate Risk" hingga "Critical".
|
|
Referensi: IPC (2019). "IPC Technical Manual Version 3.0."
|
|
4. FAO SOFI (2023/2024):
|
|
Konteks kondisi per pilar: availability (supply/stok), access (keterjangkauan),
|
|
utilization (nutrisi/sanitasi), stability (kerentanan terhadap guncangan).
|
|
Referensi: FAO et al. (2024). "The State of Food Security and Nutrition
|
|
in the World 2024."
|
|
"""
|
|
|
|
import pandas as pd
|
|
import numpy as np
|
|
from datetime import datetime
|
|
import logging
|
|
import json
|
|
import sys as _sys
|
|
|
|
from scripts.bigquery_config import get_bigquery_client
|
|
from scripts.bigquery_helpers import (
|
|
log_update,
|
|
load_to_bigquery,
|
|
read_from_bigquery,
|
|
setup_logging,
|
|
save_etl_metadata,
|
|
)
|
|
from google.cloud import bigquery
|
|
|
|
|
|
# =============================================================================
|
|
# KONSTANTA GLOBAL
|
|
# =============================================================================
|
|
|
|
DIRECTION_INVERT_KEYWORDS = frozenset({
|
|
"negative", "lower_better", "lower_is_better", "inverse", "neg",
|
|
})
|
|
DIRECTION_POSITIVE_KEYWORDS = frozenset({
|
|
"positive", "higher_better", "higher_is_better",
|
|
})
|
|
|
|
NORMALIZE_FRAMEWORKS_JOINTLY = False
|
|
PERFORMANCE_THRESHOLD = 60.0
|
|
|
|
# country_id fiktif untuk baris ASEAN aggregate
|
|
ASEAN_COUNTRY_ID = 0
|
|
ASEAN_COUNTRY_NAME = "ASEAN"
|
|
ASEAN_COUNTRY_NAME_ID = "ASEAN" # sama di kedua bahasa
|
|
|
|
SDG_ONLY_KEYWORDS: frozenset = frozenset([
|
|
"prevalence of undernourishment (percent) (3-year average)",
|
|
"number of people undernourished (million) (3-year average)",
|
|
"prevalence of severe food insecurity in the total population (percent) (3-year average)",
|
|
"prevalence of severe food insecurity in the male adult population (percent) (3-year average)",
|
|
"prevalence of severe food insecurity in the female adult population (percent) (3-year average)",
|
|
"prevalence of moderate or severe food insecurity in the total population (percent) (3-year average)",
|
|
"prevalence of moderate or severe food insecurity in the male adult population (percent) (3-year average)",
|
|
"prevalence of moderate or severe food insecurity in the female adult population (percent) (3-year average)",
|
|
"number of severely food insecure people (million) (3-year average)",
|
|
"number of severely food insecure male adults (million) (3-year average)",
|
|
"number of severely food insecure female adults (million) (3-year average)",
|
|
"number of moderately or severely food insecure people (million) (3-year average)",
|
|
"number of moderately or severely food insecure male adults (million) (3-year average)",
|
|
"number of moderately or severely food insecure female adults (million) (3-year average)",
|
|
"percentage of children under 5 years of age who are stunted (modelled estimates) (percent)",
|
|
"number of children under 5 years of age who are stunted (modeled estimates) (million)",
|
|
"percentage of children under 5 years affected by wasting (percent)",
|
|
"number of children under 5 years affected by wasting (million)",
|
|
"percentage of children under 5 years of age who are overweight (modelled estimates) (percent)",
|
|
"number of children under 5 years of age who are overweight (modeled estimates) (million)",
|
|
"prevalence of anemia among women of reproductive age (15-49 years) (percent)",
|
|
"number of women of reproductive age (15-49 years) affected by anemia (million)",
|
|
])
|
|
_SDG_ONLY_LOWER: frozenset = frozenset(k.lower() for k in SDG_ONLY_KEYWORDS)
|
|
|
|
_FIES_DETECTION_LOWER: frozenset = frozenset([
|
|
"prevalence of severe food insecurity in the total population (percent) (3-year average)",
|
|
"prevalence of moderate or severe food insecurity in the total population (percent) (3-year average)",
|
|
"number of severely food insecure people (million) (3-year average)",
|
|
"number of moderately or severely food insecure people (million) (3-year average)",
|
|
])
|
|
|
|
|
|
# =============================================================================
|
|
# TRANSLATION DICTIONARIES
|
|
# CHANGED: "Sustainability" / "sustainability" -> "Other" / "Lainnya"
|
|
# =============================================================================
|
|
|
|
COUNTRY_NAME_ID_MAP: dict = {
|
|
"Brunei Darussalam" : "Brunei Darussalam",
|
|
"Cambodia" : "Kamboja",
|
|
"Indonesia" : "Indonesia",
|
|
"Lao People's Democratic Republic" : "Laos",
|
|
"Lao PDR" : "Laos",
|
|
"Malaysia" : "Malaysia",
|
|
"Myanmar" : "Myanmar",
|
|
"Philippines" : "Filipina",
|
|
"Singapore" : "Singapura",
|
|
"Thailand" : "Thailand",
|
|
"Timor-Leste" : "Timor-Leste",
|
|
"Viet Nam" : "Vietnam",
|
|
"Vietnam" : "Vietnam",
|
|
"ASEAN" : "ASEAN",
|
|
}
|
|
|
|
PILLAR_TRANSLATION_ID: dict = {
|
|
# Mapping nama pilar (Inggris dengan prefix Food) -> Bahasa Indonesia
|
|
"Food Availability" : "Ketersediaan Pangan",
|
|
"Food Access" : "Akses Pangan",
|
|
"Food Utilization" : "Pemanfaatan Pangan",
|
|
"Food Stability" : "Stabilitas Pangan",
|
|
"Food Other" : "Indikator Tambahan",
|
|
# Variasi tanpa prefix Food
|
|
"Availability" : "Ketersediaan Pangan",
|
|
"Access" : "Akses Pangan",
|
|
"Utilization" : "Pemanfaatan Pangan",
|
|
"Stability" : "Stabilitas Pangan",
|
|
"Other" : "Indikator Tambahan",
|
|
# Legacy Sustainability
|
|
"Sustainability" : "Indikator Tambahan",
|
|
"sustainability" : "Indikator Tambahan",
|
|
# lowercase
|
|
"food availability" : "Ketersediaan Pangan",
|
|
"food access" : "Akses Pangan",
|
|
"food utilization" : "Pemanfaatan Pangan",
|
|
"food stability" : "Stabilitas Pangan",
|
|
"food other" : "Indikator Tambahan",
|
|
"availability" : "Ketersediaan Pangan",
|
|
"access" : "Akses Pangan",
|
|
"utilization" : "Pemanfaatan Pangan",
|
|
"stability" : "Stabilitas Pangan",
|
|
"other" : "Indikator Tambahan",
|
|
}
|
|
|
|
|
|
def translate_country(name: str) -> str:
|
|
if not name:
|
|
return name
|
|
return COUNTRY_NAME_ID_MAP.get(name.strip(), name)
|
|
|
|
|
|
def translate_pillar(name: str) -> str:
|
|
if not name:
|
|
return name
|
|
return PILLAR_TRANSLATION_ID.get(name, name)
|
|
|
|
|
|
# =============================================================================
|
|
# PILLAR CONDITION CLASSIFIER
|
|
# =============================================================================
|
|
#
|
|
# Landasan teori (lihat docstring modul di atas untuk referensi lengkap):
|
|
#
|
|
# Tier skor (skala 1-100, mengacu GFSI 2022 + IPC Phase Classification):
|
|
# >= 75 : Secure / Aman — performa tinggi, kondisi baik
|
|
# >= 60 : Adequate / Memadai — di atas threshold, masih ada ruang
|
|
# >= 40 : Moderate / Sedang — tantangan nyata, perlu perhatian
|
|
# >= 20 : At Risk / Berisiko — kondisi lemah, butuh intervensi
|
|
# < 20 : Critical / Kritis — sangat buruk, tindakan segera
|
|
#
|
|
# Label kontekstual per pilar mengacu definisi FAO/CFS empat pilar:
|
|
# Food Availability : ketersediaan pasokan (produksi, stok, impor)
|
|
# Food Access : keterjangkauan ekonomi & fisik terhadap pangan
|
|
# Food Utilization : pemanfaatan biologis (gizi, sanitasi, kesehatan)
|
|
# Food Stability : konsistensi tiga pilar di atas dari waktu ke waktu
|
|
# Food Other : indikator multidimensi / suplemen
|
|
#
|
|
# =============================================================================
|
|
|
|
# Tier thresholds (urut dari tertinggi)
|
|
_CONDITION_TIERS = [
|
|
# (min_score, base_label_en, base_label_id)
|
|
(75, "Secure", "Aman"),
|
|
(60, "Adequate", "Memadai"),
|
|
(40, "Moderate", "Sedang"),
|
|
(20, "At Risk", "Berisiko"),
|
|
( 0, "Critical", "Kritis"),
|
|
]
|
|
|
|
# Konteks kondisi per pilar per tier (EN, ID)
|
|
# Mengacu makna substantif pilar (FAO SOFI 2024; FSC Handbook 2020;
|
|
# IPC Technical Manual 2019).
|
|
_PILLAR_CONTEXT: dict = {
|
|
# ---- Food Availability ----
|
|
"Food Availability": {
|
|
"Secure" : ("Food supply is abundant and well-distributed",
|
|
"Pasokan pangan berlimpah dan terdistribusi merata"),
|
|
"Adequate" : ("Food supply is sufficient with minor gaps",
|
|
"Pasokan pangan cukup dengan kesenjangan minor"),
|
|
"Moderate" : ("Food supply shows signs of strain",
|
|
"Pasokan pangan menunjukkan tanda-tanda tekanan"),
|
|
"At Risk" : ("Food supply is insufficient; stocks are dwindling",
|
|
"Pasokan pangan tidak mencukupi; stok mulai menipis"),
|
|
"Critical" : ("Severe food supply deficit; stocks critically low",
|
|
"Defisit pasokan pangan parah; stok dalam kondisi kritis"),
|
|
},
|
|
# ---- Food Access ----
|
|
"Food Access": {
|
|
"Secure" : ("Food is economically and physically accessible to all",
|
|
"Pangan terjangkau secara ekonomi dan fisik bagi semua"),
|
|
"Adequate" : ("Food access is generally good with limited barriers",
|
|
"Akses pangan umumnya baik dengan hambatan terbatas"),
|
|
"Moderate" : ("Portions of the population face access constraints",
|
|
"Sebagian penduduk menghadapi kendala akses pangan"),
|
|
"At Risk" : ("Significant affordability or physical access barriers",
|
|
"Hambatan keterjangkauan atau akses fisik yang signifikan"),
|
|
"Critical" : ("Widespread inability to access sufficient food",
|
|
"Ketidakmampuan meluas dalam mengakses pangan yang cukup"),
|
|
},
|
|
# ---- Food Utilization ----
|
|
"Food Utilization": {
|
|
"Secure" : ("Dietary quality, nutrition, and sanitation are strong",
|
|
"Kualitas gizi, nutrisi, dan sanitasi dalam kondisi baik"),
|
|
"Adequate" : ("Nutrition and sanitation are adequate; minor deficiencies",
|
|
"Gizi dan sanitasi memadai; kekurangan minor masih ada"),
|
|
"Moderate" : ("Nutritional gaps or sanitation issues are evident",
|
|
"Kesenjangan gizi atau masalah sanitasi mulai terlihat"),
|
|
"At Risk" : ("Significant nutritional deficiencies or poor sanitation",
|
|
"Kekurangan gizi atau sanitasi buruk yang signifikan"),
|
|
"Critical" : ("Severe malnutrition and/or critical sanitation deficits",
|
|
"Malnutrisi parah dan/atau defisit sanitasi yang kritis"),
|
|
},
|
|
# ---- Food Stability ----
|
|
"Food Stability": {
|
|
"Secure" : ("Food security is consistently maintained over time",
|
|
"Ketahanan pangan terjaga konsisten dari waktu ke waktu"),
|
|
"Adequate" : ("Stability is generally good with manageable risks",
|
|
"Stabilitas umumnya baik dengan risiko yang masih terkelola"),
|
|
"Moderate" : ("Periodic shocks or vulnerabilities affect stability",
|
|
"Guncangan periodik atau kerentanan memengaruhi stabilitas"),
|
|
"At Risk" : ("Frequent disruptions threaten food security continuity",
|
|
"Gangguan berulang mengancam kesinambungan ketahanan pangan"),
|
|
"Critical" : ("Sustained instability; food security is highly fragile",
|
|
"Ketidakstabilan berkelanjutan; ketahanan pangan sangat rapuh"),
|
|
},
|
|
# ---- Food Other / Indikator Tambahan ----
|
|
"Food Other": {
|
|
"Secure" : ("Supplementary indicators reflect strong food system",
|
|
"Indikator tambahan mencerminkan sistem pangan yang kuat"),
|
|
"Adequate" : ("Supplementary indicators are at acceptable levels",
|
|
"Indikator tambahan berada pada level yang dapat diterima"),
|
|
"Moderate" : ("Supplementary indicators signal emerging challenges",
|
|
"Indikator tambahan memberi sinyal tantangan yang muncul"),
|
|
"At Risk" : ("Supplementary indicators show concerning levels",
|
|
"Indikator tambahan menunjukkan level yang mengkhawatirkan"),
|
|
"Critical" : ("Supplementary indicators reflect systemic food system failure",
|
|
"Indikator tambahan mencerminkan kegagalan sistemik pangan"),
|
|
},
|
|
}
|
|
|
|
# Fallback jika pillar_name tidak dikenali
|
|
_PILLAR_CONTEXT_FALLBACK: dict = {
|
|
"Secure" : ("Performance is high across food security indicators",
|
|
"Performa tinggi pada indikator ketahanan pangan"),
|
|
"Adequate" : ("Performance is adequate across food security indicators",
|
|
"Performa memadai pada indikator ketahanan pangan"),
|
|
"Moderate" : ("Performance shows moderate challenges",
|
|
"Performa menunjukkan tantangan yang moderat"),
|
|
"At Risk" : ("Performance indicates vulnerability in food security",
|
|
"Performa mengindikasikan kerentanan ketahanan pangan"),
|
|
"Critical" : ("Performance is critically low; urgent action needed",
|
|
"Performa sangat rendah; tindakan segera diperlukan"),
|
|
}
|
|
|
|
|
|
# =============================================================================
|
|
# FRAMEWORK CONDITION CLASSIFIER
|
|
# =============================================================================
|
|
#
|
|
# Konteks kondisi per framework per tier.
|
|
# Framework mencerminkan era kebijakan global:
|
|
# MDGs : Millennium Development Goals (2000-2015)
|
|
# SDGs : Sustainable Development Goals (2016-sekarang)
|
|
# Total : Gabungan semua indikator (lintas framework)
|
|
#
|
|
# Label tier sama dengan pillar condition (GFSI 2022 + IPC 2019),
|
|
# namun deskripsi kontekstual mengacu pada capaian indikator dalam
|
|
# kerangka MDGs/SDGs/Total secara keseluruhan.
|
|
#
|
|
# =============================================================================
|
|
|
|
_FRAMEWORK_CONTEXT: dict = {
|
|
# ---- MDGs ----
|
|
"MDGs": {
|
|
"Secure" : ("MDG food security targets are consistently met",
|
|
"Target ketahanan pangan MDGs terpenuhi secara konsisten"),
|
|
"Adequate" : ("MDG food security targets are largely on track",
|
|
"Target ketahanan pangan MDGs sebagian besar berjalan sesuai rencana"),
|
|
"Moderate" : ("Progress toward MDG food security targets shows mixed results",
|
|
"Kemajuan menuju target MDGs menunjukkan hasil yang beragam"),
|
|
"At Risk" : ("MDG food security targets are at risk of not being met",
|
|
"Target ketahanan pangan MDGs berisiko tidak tercapai"),
|
|
"Critical" : ("MDG food security targets are severely off track",
|
|
"Target ketahanan pangan MDGs sangat jauh dari jalur pencapaian"),
|
|
},
|
|
# ---- SDGs ----
|
|
"SDGs": {
|
|
"Secure" : ("SDG food security commitments are well on track",
|
|
"Komitmen ketahanan pangan SDGs berjalan dengan baik"),
|
|
"Adequate" : ("SDG food security indicators show adequate progress",
|
|
"Indikator ketahanan pangan SDGs menunjukkan kemajuan yang memadai"),
|
|
"Moderate" : ("SDG food security progress is uneven and requires attention",
|
|
"Kemajuan ketahanan pangan SDGs tidak merata dan perlu perhatian"),
|
|
"At Risk" : ("SDG food security commitments face significant implementation gaps",
|
|
"Komitmen ketahanan pangan SDGs menghadapi kesenjangan implementasi yang signifikan"),
|
|
"Critical" : ("SDG food security goals are critically off track",
|
|
"Tujuan ketahanan pangan SDGs sangat jauh dari pencapaian yang diharapkan"),
|
|
},
|
|
# ---- Total ----
|
|
"Total": {
|
|
"Secure" : ("Overall food security performance is strong across all indicators",
|
|
"Kinerja ketahanan pangan secara keseluruhan kuat di semua indikator"),
|
|
"Adequate" : ("Overall food security performance is adequate with room for improvement",
|
|
"Kinerja ketahanan pangan secara keseluruhan memadai dengan ruang perbaikan"),
|
|
"Moderate" : ("Overall food security performance shows moderate challenges",
|
|
"Kinerja ketahanan pangan secara keseluruhan menunjukkan tantangan moderat"),
|
|
"At Risk" : ("Overall food security performance indicates systemic vulnerabilities",
|
|
"Kinerja ketahanan pangan secara keseluruhan mengindikasikan kerentanan sistemik"),
|
|
"Critical" : ("Overall food security performance is critically low across indicators",
|
|
"Kinerja ketahanan pangan secara keseluruhan sangat rendah di seluruh indikator"),
|
|
},
|
|
}
|
|
|
|
_FRAMEWORK_CONTEXT_FALLBACK: dict = {
|
|
"Secure" : ("Food security performance is strong under this framework",
|
|
"Kinerja ketahanan pangan kuat dalam kerangka ini"),
|
|
"Adequate" : ("Food security performance is adequate under this framework",
|
|
"Kinerja ketahanan pangan memadai dalam kerangka ini"),
|
|
"Moderate" : ("Food security performance shows moderate challenges under this framework",
|
|
"Kinerja ketahanan pangan menunjukkan tantangan moderat dalam kerangka ini"),
|
|
"At Risk" : ("Food security performance is at risk under this framework",
|
|
"Kinerja ketahanan pangan berisiko dalam kerangka ini"),
|
|
"Critical" : ("Food security performance is critically low under this framework",
|
|
"Kinerja ketahanan pangan sangat rendah dalam kerangka ini"),
|
|
}
|
|
|
|
|
|
def get_framework_condition(framework: str, score: float) -> tuple:
|
|
"""
|
|
Mengembalikan 4-tuple kondisi framework berdasarkan skor dan nama framework.
|
|
|
|
Tier mengacu GFSI 2022 (Economist Impact) + IPC Phase Classification (2019):
|
|
>= 75 -> Secure / Aman
|
|
>= 60 -> Adequate / Memadai
|
|
>= 40 -> Moderate / Sedang
|
|
>= 20 -> At Risk / Berisiko
|
|
< 20 -> Critical / Kritis
|
|
|
|
Deskripsi kontekstual mengacu capaian MDGs/SDGs per FAO SOFI 2024
|
|
dan UN SDG Progress Report 2023.
|
|
|
|
Args:
|
|
framework : Nama framework ("MDGs", "SDGs", atau "Total").
|
|
score : Skor ternormalisasi skala 1-100.
|
|
|
|
Returns:
|
|
Tuple (
|
|
condition_en : str — label tier EN, e.g. "At Risk"
|
|
condition_id : str — label tier ID, e.g. "Berisiko"
|
|
condition_desc_en : str — deskripsi kontekstual EN
|
|
condition_desc_id : str — deskripsi kontekstual ID
|
|
)
|
|
"""
|
|
if score is None or (isinstance(score, float) and np.isnan(score)):
|
|
return ("N/A", "N/A", "N/A", "N/A")
|
|
|
|
# Tentukan tier
|
|
tier_label_en = _CONDITION_TIERS[-1][1] # default: Critical
|
|
tier_label_id = _CONDITION_TIERS[-1][2]
|
|
for min_score, lbl_en, lbl_id in _CONDITION_TIERS:
|
|
if score >= min_score:
|
|
tier_label_en = lbl_en
|
|
tier_label_id = lbl_id
|
|
break
|
|
|
|
# Ambil deskripsi kontekstual per framework
|
|
ctx = _FRAMEWORK_CONTEXT.get(framework, None)
|
|
if ctx:
|
|
ctx_en, ctx_id = ctx.get(
|
|
tier_label_en,
|
|
_FRAMEWORK_CONTEXT_FALLBACK.get(tier_label_en, ("", ""))
|
|
)
|
|
else:
|
|
ctx_en, ctx_id = _FRAMEWORK_CONTEXT_FALLBACK.get(tier_label_en, ("", ""))
|
|
|
|
return tier_label_en, tier_label_id, ctx_en, ctx_id
|
|
|
|
|
|
def get_pillar_condition(pillar_name: str, score: float) -> tuple:
|
|
"""
|
|
Mengembalikan 4-tuple kondisi pilar berdasarkan skor dan nama pilar.
|
|
|
|
Tier mengacu GFSI 2022 (Economist Impact) + IPC Phase Classification (2019):
|
|
>= 75 -> Secure / Aman
|
|
>= 60 -> Adequate / Memadai
|
|
>= 40 -> Moderate / Sedang
|
|
>= 20 -> At Risk / Berisiko
|
|
< 20 -> Critical / Kritis
|
|
|
|
Deskripsi kontekstual mengacu FAO/CFS definisi 4 pilar (World Food Summit
|
|
1996; CFS 2009) dan FAO SOFI 2024.
|
|
|
|
Args:
|
|
pillar_name : Nama pilar dalam bahasa Inggris (e.g. "Food Availability").
|
|
score : Skor ternormalisasi skala 1-100.
|
|
|
|
Returns:
|
|
Tuple (
|
|
condition_en : str — label tier bahasa Inggris, e.g. "At Risk"
|
|
condition_id : str — label tier bahasa Indonesia, e.g. "Berisiko"
|
|
condition_desc_en : str — deskripsi kontekstual bahasa Inggris
|
|
condition_desc_id : str — deskripsi kontekstual bahasa Indonesia
|
|
)
|
|
"""
|
|
if score is None or (isinstance(score, float) and np.isnan(score)):
|
|
return ("N/A", "N/A", "N/A", "N/A")
|
|
|
|
# Tentukan tier
|
|
tier_label_en = _CONDITION_TIERS[-1][1] # default: Critical
|
|
tier_label_id = _CONDITION_TIERS[-1][2]
|
|
for min_score, lbl_en, lbl_id in _CONDITION_TIERS:
|
|
if score >= min_score:
|
|
tier_label_en = lbl_en
|
|
tier_label_id = lbl_id
|
|
break
|
|
|
|
# Ambil deskripsi kontekstual per pilar
|
|
ctx = _PILLAR_CONTEXT.get(pillar_name, None)
|
|
if ctx:
|
|
ctx_en, ctx_id = ctx.get(
|
|
tier_label_en,
|
|
_PILLAR_CONTEXT_FALLBACK.get(tier_label_en, ("", ""))
|
|
)
|
|
else:
|
|
ctx_en, ctx_id = _PILLAR_CONTEXT_FALLBACK.get(tier_label_en, ("", ""))
|
|
|
|
return tier_label_en, tier_label_id, ctx_en, ctx_id
|
|
|
|
|
|
# =============================================================================
|
|
# WINDOWS CP1252 SAFE LOGGING
|
|
# =============================================================================
|
|
|
|
class _SafeStreamHandler(logging.StreamHandler):
|
|
def emit(self, record):
|
|
try:
|
|
super().emit(record)
|
|
except UnicodeEncodeError:
|
|
try:
|
|
msg = self.format(record)
|
|
self.stream.write(
|
|
msg.encode("utf-8", errors="replace").decode("ascii", errors="replace")
|
|
+ self.terminator
|
|
)
|
|
self.flush()
|
|
except Exception:
|
|
self.handleError(record)
|
|
|
|
|
|
# =============================================================================
|
|
# HELPERS
|
|
# =============================================================================
|
|
|
|
def _should_invert(direction: str, logger=None, context: str = "") -> bool:
|
|
d = str(direction).lower().strip()
|
|
if d in DIRECTION_INVERT_KEYWORDS:
|
|
return True
|
|
if d in DIRECTION_POSITIVE_KEYWORDS:
|
|
return False
|
|
if logger:
|
|
logger.warning(
|
|
f" [DIRECTION WARNING] Unknown direction '{direction}' "
|
|
f"{'(' + context + ')' if context else ''}. Defaulting to positive (no invert)."
|
|
)
|
|
return False
|
|
|
|
|
|
def global_minmax(series: pd.Series, lo: float = 1.0, hi: float = 100.0) -> pd.Series:
|
|
values = series.dropna().values
|
|
if len(values) == 0:
|
|
return pd.Series(np.nan, index=series.index)
|
|
v_min, v_max = values.min(), values.max()
|
|
if v_min == v_max:
|
|
return pd.Series((lo + hi) / 2.0, index=series.index)
|
|
result = np.full(len(series), np.nan)
|
|
not_nan = series.notna()
|
|
raw = series[not_nan].values
|
|
result[not_nan.values] = lo + (raw - v_min) / (v_max - v_min) * (hi - lo)
|
|
return pd.Series(result, index=series.index)
|
|
|
|
|
|
def add_yoy(df: pd.DataFrame, group_cols: list, score_col: str) -> pd.DataFrame:
|
|
df = df.sort_values(group_cols + ["year"]).reset_index(drop=True)
|
|
if group_cols:
|
|
df["year_over_year_change"] = df.groupby(group_cols)[score_col].diff()
|
|
else:
|
|
df["year_over_year_change"] = df[score_col].diff()
|
|
return df
|
|
|
|
|
|
def safe_int(series: pd.Series, fill: int = 0, col_name: str = "", logger=None) -> pd.Series:
|
|
n_nan = series.isna().sum()
|
|
if n_nan > 0 and logger:
|
|
logger.warning(
|
|
f" [NaN WARNING] Kolom '{col_name}' punya {n_nan} NaN -> di-fill dengan {fill}"
|
|
)
|
|
return series.fillna(fill).astype(int)
|
|
|
|
|
|
def check_and_dedup(df: pd.DataFrame, key_cols: list, context: str = "", logger=None) -> pd.DataFrame:
|
|
dupes = df.duplicated(subset=key_cols, keep=False)
|
|
if dupes.any():
|
|
n_dupes = dupes.sum()
|
|
if logger:
|
|
logger.warning(
|
|
f" [DEDUP WARNING] {context}: {n_dupes} duplikat rows pada {key_cols}. "
|
|
f"Di-aggregate dengan mean."
|
|
)
|
|
numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()
|
|
agg_dict = {
|
|
c: ("mean" if c in numeric_cols else "first")
|
|
for c in df.columns if c not in key_cols
|
|
}
|
|
df = df.groupby(key_cols, as_index=False).agg(agg_dict)
|
|
return df
|
|
|
|
|
|
def _performance_status(score) -> str:
|
|
if score is None or (isinstance(score, float) and np.isnan(score)):
|
|
return "N/A"
|
|
return "Good" if score >= PERFORMANCE_THRESHOLD else "Bad"
|
|
|
|
|
|
def _fmt_score(score) -> str:
|
|
if score is None or (isinstance(score, float) and np.isnan(score)):
|
|
return "N/A"
|
|
return f"{score:.2f}"
|
|
|
|
|
|
def _fmt_delta(delta) -> str:
|
|
if delta is None or (isinstance(delta, float) and np.isnan(delta)):
|
|
return "N/A"
|
|
sign = "+" if delta >= 0 else ""
|
|
return f"{sign}{delta:.2f}"
|
|
|
|
|
|
# =============================================================================
|
|
# NARRATIVE CONDITION DETECTORS (shared)
|
|
# =============================================================================
|
|
|
|
def _detect_series_trend(scores: list) -> str:
|
|
if len(scores) < 3:
|
|
return "insufficient"
|
|
|
|
x = np.arange(len(scores))
|
|
slope = np.polyfit(x, scores, 1)[0]
|
|
cv = np.std(scores) / (np.mean(scores) + 1e-9)
|
|
|
|
if cv > 0.20:
|
|
return "fluctuating"
|
|
|
|
mid = len(scores) // 2
|
|
slope1 = np.polyfit(np.arange(mid), scores[:mid], 1)[0] if mid > 1 else slope
|
|
slope2 = np.polyfit(np.arange(len(scores) - mid), scores[mid:], 1)[0] if (len(scores) - mid) > 1 else slope
|
|
|
|
if slope > 0:
|
|
slowing = slope2 < slope1
|
|
return "improving_slowing" if slowing else "improving_consistent"
|
|
else:
|
|
return "deteriorating"
|
|
|
|
|
|
def _detect_country_gap(scores_by_country_year: pd.DataFrame, score_col: str) -> str:
|
|
std_by_year = (
|
|
scores_by_country_year[scores_by_country_year["country_name"] != ASEAN_COUNTRY_NAME]
|
|
.groupby("year")[score_col]
|
|
.std().dropna()
|
|
)
|
|
if len(std_by_year) < 3:
|
|
return "unknown"
|
|
|
|
years = sorted(std_by_year.index)
|
|
stds = [std_by_year[y] for y in years]
|
|
slope = np.polyfit(np.arange(len(stds)), stds, 1)[0]
|
|
mean_s = np.mean(stds)
|
|
|
|
if abs(slope) < 0.02 * mean_s:
|
|
return "stable"
|
|
return "widening" if slope > 0 else "narrowing"
|
|
|
|
|
|
def _find_anomaly_year(values_by_year: dict) -> tuple:
|
|
years = sorted(values_by_year.keys())
|
|
deltas = {}
|
|
for i in range(1, len(years)):
|
|
y0, y1 = years[i-1], years[i]
|
|
v0, v1 = values_by_year.get(y0), values_by_year.get(y1)
|
|
if v0 is not None and v1 is not None and not (pd.isna(v0) or pd.isna(v1)):
|
|
deltas[y1] = v1 - v0
|
|
|
|
if not deltas:
|
|
return None, None
|
|
|
|
threshold = 1.5 * np.std(list(deltas.values()))
|
|
min_y = min(deltas, key=deltas.get)
|
|
max_y = max(deltas, key=deltas.get)
|
|
|
|
if abs(deltas[min_y]) > threshold and deltas[min_y] < 0:
|
|
return min_y, "drop"
|
|
if abs(deltas[max_y]) > threshold and deltas[max_y] > 0:
|
|
return max_y, "rise"
|
|
return None, None
|
|
|
|
|
|
# =============================================================================
|
|
# NARRATIVE BUILDER — PILLAR
|
|
# =============================================================================
|
|
|
|
def _build_pillar_narrative(
|
|
year: int,
|
|
pillar_name: str,
|
|
pillar_score: float,
|
|
rank_in_year: int,
|
|
n_pillars: int,
|
|
yoy_val,
|
|
top_country: str,
|
|
top_country_score,
|
|
bot_country: str,
|
|
bot_country_score,
|
|
pillar_scores_history: dict,
|
|
all_pillar_scores_year: pd.DataFrame,
|
|
country_pillar_all: pd.DataFrame,
|
|
is_asean: bool = False,
|
|
) -> tuple:
|
|
sentences_en = []
|
|
sentences_id = []
|
|
|
|
pillar_name_id = translate_pillar(pillar_name)
|
|
|
|
rank_suffix = {1: "st", 2: "nd", 3: "rd"}.get(rank_in_year, "th")
|
|
perf_word_en = "good" if pillar_score >= PERFORMANCE_THRESHOLD else "below target"
|
|
perf_word_id = "baik" if pillar_score >= PERFORMANCE_THRESHOLD else "di bawah target"
|
|
|
|
subject_en = "ASEAN region" if is_asean else "this region"
|
|
subject_id = "kawasan ASEAN" if is_asean else "kawasan ini"
|
|
|
|
s1_en = (
|
|
f"In {year}, the {pillar_name} pillar ranked {rank_in_year}{rank_suffix} out of "
|
|
f"{n_pillars} pillars with a score of {_fmt_score(pillar_score)} ({perf_word_en})."
|
|
)
|
|
s1_id = (
|
|
f"Pada tahun {year}, pilar {pillar_name_id} menempati peringkat {rank_in_year} dari "
|
|
f"{n_pillars} pilar dengan skor {_fmt_score(pillar_score)} ({perf_word_id})."
|
|
)
|
|
sentences_en.append(s1_en)
|
|
sentences_id.append(s1_id)
|
|
|
|
if yoy_val is not None and not pd.isna(yoy_val):
|
|
if abs(yoy_val) < 0.5:
|
|
s2_en = "Performance was relatively stable compared to the previous year."
|
|
s2_id = "Performa relatif stabil dibandingkan tahun sebelumnya."
|
|
elif yoy_val > 0:
|
|
s2_en = f"This is an improvement of {abs(yoy_val):.2f} points from the previous year."
|
|
s2_id = f"Ini merupakan peningkatan {abs(yoy_val):.2f} poin dari tahun sebelumnya."
|
|
else:
|
|
s2_en = f"This marks a decline of {abs(yoy_val):.2f} points from the previous year."
|
|
s2_id = f"Ini menandai penurunan {abs(yoy_val):.2f} poin dari tahun sebelumnya."
|
|
sentences_en.append(s2_en)
|
|
sentences_id.append(s2_id)
|
|
|
|
hist_years = sorted(pillar_scores_history.keys())
|
|
hist_scores = [
|
|
pillar_scores_history[y]
|
|
for y in hist_years
|
|
if not pd.isna(pillar_scores_history.get(y, np.nan))
|
|
]
|
|
|
|
if len(hist_scores) >= 3:
|
|
trend = _detect_series_trend(hist_scores)
|
|
if trend == "improving_consistent":
|
|
s3_en = f"This pillar has shown consistent improvement since {hist_years[0]}."
|
|
s3_id = f"Pilar {pillar_name_id} menunjukkan perbaikan yang konsisten sejak {hist_years[0]}."
|
|
elif trend == "improving_slowing":
|
|
s3_en = f"While the pillar improved since {hist_years[0]}, the pace has slowed in recent years."
|
|
s3_id = f"Meskipun pilar {pillar_name_id} membaik sejak {hist_years[0]}, lajunya melambat dalam beberapa tahun terakhir."
|
|
elif trend == "deteriorating":
|
|
s3_en = f"This pillar has shown a declining trend since {hist_years[0]}, requiring targeted intervention."
|
|
s3_id = f"Pilar {pillar_name_id} menunjukkan tren penurunan sejak {hist_years[0]}, memerlukan intervensi yang terarah."
|
|
elif trend == "fluctuating":
|
|
s3_en = f"Performance in this pillar has been inconsistent since {hist_years[0]}, with no clear trend."
|
|
s3_id = f"Performa pilar {pillar_name_id} tidak konsisten sejak {hist_years[0]}, tanpa tren yang jelas."
|
|
else:
|
|
s3_en = ""
|
|
s3_id = ""
|
|
if s3_en:
|
|
sentences_en.append(s3_en)
|
|
sentences_id.append(s3_id)
|
|
|
|
if is_asean and not country_pillar_all.empty:
|
|
gap_trend = _detect_country_gap(
|
|
country_pillar_all[country_pillar_all["year"] <= year],
|
|
"pillar_country_score_1_100"
|
|
)
|
|
if gap_trend == "widening":
|
|
s4_en = "Country disparities within this pillar have widened over time."
|
|
s4_id = f"Kesenjangan antar negara dalam pilar {pillar_name_id} semakin melebar seiring waktu."
|
|
elif gap_trend == "narrowing":
|
|
s4_en = "Country disparities within this pillar have narrowed, indicating more balanced progress."
|
|
s4_id = f"Kesenjangan antar negara dalam pilar {pillar_name_id} menyempit, mengindikasikan kemajuan yang lebih merata."
|
|
else:
|
|
s4_en = ""
|
|
s4_id = ""
|
|
if s4_en:
|
|
sentences_en.append(s4_en)
|
|
sentences_id.append(s4_id)
|
|
|
|
if is_asean and top_country and bot_country and top_country != bot_country:
|
|
top_country_id = translate_country(top_country)
|
|
bot_country_id = translate_country(bot_country)
|
|
s5_en = (
|
|
f"{top_country} performed best in this pillar ({_fmt_score(top_country_score)}), "
|
|
f"while {bot_country} had the lowest score ({_fmt_score(bot_country_score)})."
|
|
)
|
|
s5_id = (
|
|
f"{top_country_id} memiliki performa terbaik dalam pilar {pillar_name_id} "
|
|
f"({_fmt_score(top_country_score)}), "
|
|
f"sementara {bot_country_id} memiliki skor terendah ({_fmt_score(bot_country_score)})."
|
|
)
|
|
sentences_en.append(s5_en)
|
|
sentences_id.append(s5_id)
|
|
|
|
if not all_pillar_scores_year.empty and len(all_pillar_scores_year) > 1:
|
|
sorted_pillars = all_pillar_scores_year.sort_values("pillar_country_score_1_100", ascending=False)
|
|
strongest = sorted_pillars.iloc[0]
|
|
weakest = sorted_pillars.iloc[-1]
|
|
|
|
if strongest["pillar_name"] != pillar_name and weakest["pillar_name"] != pillar_name:
|
|
strongest_id = translate_pillar(strongest["pillar_name"])
|
|
weakest_id = translate_pillar(weakest["pillar_name"])
|
|
s6_en = (
|
|
f"Across all pillars in {year}, {strongest['pillar_name']} scored highest "
|
|
f"({_fmt_score(strongest['pillar_country_score_1_100'])}) and {weakest['pillar_name']} "
|
|
f"scored lowest ({_fmt_score(weakest['pillar_country_score_1_100'])})."
|
|
)
|
|
s6_id = (
|
|
f"Di antara semua pilar pada tahun {year}, {strongest_id} mendapat skor "
|
|
f"tertinggi ({_fmt_score(strongest['pillar_country_score_1_100'])}) dan {weakest_id} "
|
|
f"mendapat skor terendah ({_fmt_score(weakest['pillar_country_score_1_100'])})."
|
|
)
|
|
sentences_en.append(s6_en)
|
|
sentences_id.append(s6_id)
|
|
|
|
narrative_en = " ".join(s for s in sentences_en if s)
|
|
narrative_id = " ".join(s for s in sentences_id if s)
|
|
return narrative_en, narrative_id
|
|
|
|
|
|
# =============================================================================
|
|
# MAIN CLASS
|
|
# =============================================================================
|
|
|
|
class FoodSecurityAggregator:
|
|
|
|
def __init__(self, client: bigquery.Client):
|
|
self.client = client
|
|
self.logger = logging.getLogger(self.__class__.__name__)
|
|
self.logger.propagate = False
|
|
|
|
self.load_metadata = {
|
|
"agg_pillar_by_country": {"rows_loaded": 0, "status": "pending", "start_time": None, "end_time": None},
|
|
"agg_framework_by_country": {"rows_loaded": 0, "status": "pending", "start_time": None, "end_time": None},
|
|
"agg_narrative_pillar": {"rows_loaded": 0, "status": "pending", "start_time": None, "end_time": None},
|
|
}
|
|
|
|
self.df = None
|
|
self.sdgs_start_year = None
|
|
self._ind_year_framework: pd.DataFrame = None
|
|
|
|
# =========================================================================
|
|
# STEP 1: Load data
|
|
# =========================================================================
|
|
|
|
def load_data(self):
|
|
self.logger.info("=" * 70)
|
|
self.logger.info("STEP 1: LOAD DATA from fs_asean_gold")
|
|
self.logger.info("=" * 70)
|
|
|
|
self.df = read_from_bigquery(
|
|
self.client, "fact_asean_food_security_selected", layer='gold'
|
|
)
|
|
self.logger.info(f" fact_asean_food_security_selected : {len(self.df):,} rows")
|
|
|
|
required_cols = {
|
|
"country_id", "country_name",
|
|
"indicator_id", "indicator_name", "direction",
|
|
"pillar_id", "pillar_name",
|
|
"time_id", "year", "value",
|
|
}
|
|
missing_cols = required_cols - set(self.df.columns)
|
|
if missing_cols:
|
|
raise ValueError(f"Kolom berikut tidak ditemukan: {missing_cols}")
|
|
|
|
n_null_dir = self.df["direction"].isna().sum()
|
|
if n_null_dir > 0:
|
|
self.logger.warning(f" [DIRECTION] {n_null_dir} rows NULL -> diisi 'positive'")
|
|
self.df["direction"] = self.df["direction"].fillna("positive")
|
|
|
|
PILLAR_RENAME_MAP = {
|
|
'Availability' : 'Food Availability',
|
|
'Access' : 'Food Access',
|
|
'Utilization' : 'Food Utilization',
|
|
'Stability' : 'Food Stability',
|
|
'Other' : 'Food Other',
|
|
'Sustainability': 'Food Other',
|
|
'sustainability': 'Food Other',
|
|
}
|
|
self.df["pillar_name"] = self.df["pillar_name"].replace(PILLAR_RENAME_MAP)
|
|
|
|
if "country_name_id" not in self.df.columns:
|
|
self.df["country_name_id"] = self.df["country_name"].apply(translate_country)
|
|
if "pillar_name_id" not in self.df.columns:
|
|
self.df["pillar_name_id"] = self.df["pillar_name"].apply(translate_pillar)
|
|
|
|
self.logger.info(f"\n Rows : {len(self.df):,}")
|
|
self.logger.info(f" Countries : {self.df['country_id'].nunique()}")
|
|
self.logger.info(f" Indicators: {self.df['indicator_id'].nunique()}")
|
|
self.logger.info(
|
|
f" Years : {int(self.df['year'].min())} - {int(self.df['year'].max())}"
|
|
)
|
|
|
|
# =========================================================================
|
|
# STEP 1b: Detect sdgs_start_year + assign framework
|
|
# =========================================================================
|
|
|
|
def _detect_sdgs_start_year(self) -> int:
|
|
fies_rows = self.df[
|
|
self.df["indicator_name"].str.lower().str.strip().isin(_FIES_DETECTION_LOWER)
|
|
]
|
|
if not fies_rows.empty:
|
|
sdgs_start = int(fies_rows["year"].min())
|
|
self.logger.info(f" [FIES explicit] sdgs_start_year = {sdgs_start}")
|
|
return sdgs_start
|
|
|
|
ind_min_year = (
|
|
self.df.groupby("indicator_id")["year"]
|
|
.min().reset_index().rename(columns={"year": "min_year"})
|
|
)
|
|
unique_years = sorted(ind_min_year["min_year"].unique())
|
|
if len(unique_years) == 1:
|
|
return int(unique_years[0]) + 9999
|
|
|
|
gaps = [
|
|
(unique_years[i+1] - unique_years[i], unique_years[i], unique_years[i+1])
|
|
for i in range(len(unique_years) - 1)
|
|
]
|
|
gaps.sort(reverse=True)
|
|
_, y_before, y_after = gaps[0]
|
|
self.logger.info(f" [Fallback gap] sdgs_start_year = {y_after}")
|
|
return int(y_after)
|
|
|
|
def _assign_framework_labels(self):
|
|
self.logger.info("\n" + "=" * 70)
|
|
self.logger.info("STEP 1b: ASSIGN FRAMEWORK LABELS")
|
|
self.logger.info(f" sdgs_start_year = {self.sdgs_start_year}")
|
|
self.logger.info("=" * 70)
|
|
|
|
df = self.df.copy()
|
|
df["_is_sdg_kw"] = df["indicator_name"].str.lower().str.strip().isin(_SDG_ONLY_LOWER)
|
|
df["framework"] = "MDGs"
|
|
mask_sdgs = df["_is_sdg_kw"] & (df["year"] >= self.sdgs_start_year)
|
|
df.loc[mask_sdgs, "framework"] = "SDGs"
|
|
df = df.drop(columns=["_is_sdg_kw"])
|
|
self.df = df
|
|
|
|
self._ind_year_framework = (
|
|
self.df[["indicator_id", "year", "framework"]]
|
|
.drop_duplicates()
|
|
.reset_index(drop=True)
|
|
)
|
|
|
|
fw_dist = self.df["framework"].value_counts()
|
|
self.logger.info("\n Framework distribution (rows):")
|
|
for fw, cnt in fw_dist.items():
|
|
self.logger.info(f" {fw:<6}: {cnt:,} rows")
|
|
|
|
def _count_framework_indicators(self, year: int, framework: str) -> int:
|
|
mask = (
|
|
(self._ind_year_framework["year"] == year) &
|
|
(self._ind_year_framework["framework"] == framework)
|
|
)
|
|
return int(self._ind_year_framework.loc[mask, "indicator_id"].nunique())
|
|
|
|
# =========================================================================
|
|
# CORE HELPER: normalisasi raw value per indikator
|
|
# =========================================================================
|
|
|
|
def _get_norm_value_df(self) -> pd.DataFrame:
|
|
if "framework" not in self.df.columns:
|
|
raise ValueError("Kolom 'framework' tidak ada.")
|
|
|
|
norm_parts = []
|
|
for ind_id, grp in self.df.groupby("indicator_id"):
|
|
grp = grp.copy()
|
|
direction = str(grp["direction"].iloc[0])
|
|
do_invert = _should_invert(direction, self.logger, context=f"indicator_id={ind_id}")
|
|
valid_mask = grp["value"].notna()
|
|
n_valid = valid_mask.sum()
|
|
|
|
if n_valid < 2:
|
|
grp["norm_value"] = np.nan
|
|
norm_parts.append(grp)
|
|
continue
|
|
|
|
raw = grp.loc[valid_mask, "value"].values
|
|
v_min, v_max = raw.min(), raw.max()
|
|
normed = np.full(len(grp), np.nan)
|
|
if v_min == v_max:
|
|
normed[valid_mask.values] = 0.5
|
|
else:
|
|
normed[valid_mask.values] = (raw - v_min) / (v_max - v_min)
|
|
|
|
if do_invert:
|
|
normed = np.where(np.isnan(normed), np.nan, 1.0 - normed)
|
|
|
|
grp["norm_value"] = normed
|
|
norm_parts.append(grp)
|
|
|
|
return pd.concat(norm_parts, ignore_index=True)
|
|
|
|
# =========================================================================
|
|
# METADATA BUILDER
|
|
# =========================================================================
|
|
|
|
def _build_etl_metadata(
|
|
self,
|
|
table_name: str,
|
|
rows_loaded: int,
|
|
start_time: datetime,
|
|
end_time: datetime,
|
|
status: str,
|
|
error_msg: str = None,
|
|
) -> dict:
|
|
duration = (end_time - start_time).total_seconds() if (start_time and end_time) else 0.0
|
|
return {
|
|
"source_class" : "FoodSecurityAggregator",
|
|
"table_name" : table_name,
|
|
"execution_timestamp": start_time or end_time,
|
|
"duration_seconds" : round(duration, 4),
|
|
"rows_fetched" : rows_loaded,
|
|
"rows_transformed" : rows_loaded,
|
|
"rows_loaded" : rows_loaded,
|
|
"completeness_pct" : 100.0 if status == "success" else 0.0,
|
|
"config_snapshot" : json.dumps({
|
|
"layer" : "gold",
|
|
"write_disposition" : "WRITE_TRUNCATE",
|
|
"normalize_frameworks_jointly": NORMALIZE_FRAMEWORKS_JOINTLY,
|
|
"performance_threshold" : PERFORMANCE_THRESHOLD,
|
|
"status" : status,
|
|
"asean_country_id" : ASEAN_COUNTRY_ID,
|
|
"pillar_change" : "Sustainability renamed to Food Other, all pillars prefixed with Food",
|
|
"architecture" : "ASEAN merged into country tables (country_id=0)",
|
|
"condition_column" : "pillar_condition_en/id added to agg_pillar_by_country",
|
|
"condition_reference" : (
|
|
"GFSI 2022 (Economist Impact) score tiers >= 75/60/40/20; "
|
|
"IPC Technical Manual 2019; FAO/CFS 4-pillar framework 1996/2009; "
|
|
"FAO SOFI 2024"
|
|
),
|
|
}),
|
|
"validation_metrics" : json.dumps({
|
|
"status" : status,
|
|
"error_msg": error_msg or "",
|
|
}),
|
|
}
|
|
|
|
# =========================================================================
|
|
# HELPER: build ASEAN rows untuk tabel pillar_by_country
|
|
# =========================================================================
|
|
|
|
def _build_asean_pillar_rows(self, df_normed: pd.DataFrame) -> pd.DataFrame:
|
|
asean_agg = (
|
|
df_normed
|
|
.groupby(["pillar_id", "pillar_name", "year"])
|
|
.agg(pillar_country_norm=("norm_value", "mean"))
|
|
.reset_index()
|
|
)
|
|
asean_agg["country_id"] = ASEAN_COUNTRY_ID
|
|
asean_agg["country_name"] = ASEAN_COUNTRY_NAME
|
|
asean_agg["country_name_id"] = ASEAN_COUNTRY_NAME_ID
|
|
asean_agg["pillar_name_id"] = asean_agg["pillar_name"].apply(translate_pillar)
|
|
return asean_agg
|
|
|
|
# =========================================================================
|
|
# STEP 2: agg_pillar_by_country (termasuk ASEAN + kolom kondisi)
|
|
# =========================================================================
|
|
|
|
def calc_pillar_by_country(self) -> pd.DataFrame:
|
|
table_name = "agg_pillar_by_country"
|
|
self.load_metadata[table_name]["start_time"] = datetime.now()
|
|
self.logger.info("\n" + "=" * 70)
|
|
self.logger.info(f"STEP 2: {table_name} -> [Gold] fs_asean_gold")
|
|
self.logger.info(" Termasuk baris ASEAN (country_id=0) untuk filter Looker Studio")
|
|
self.logger.info(" Kolom baru: pillar_condition_en, pillar_condition_id")
|
|
self.logger.info("=" * 70)
|
|
|
|
try:
|
|
df_normed = self._get_norm_value_df()
|
|
|
|
# Baris per negara
|
|
df_countries = (
|
|
df_normed
|
|
.groupby(["country_id", "country_name", "pillar_id", "pillar_name", "year"])
|
|
.agg(pillar_country_norm=("norm_value", "mean"))
|
|
.reset_index()
|
|
)
|
|
df_countries["pillar_name_id"] = df_countries["pillar_name"].apply(translate_pillar)
|
|
df_countries["country_name_id"] = df_countries["country_name"].apply(translate_country)
|
|
|
|
# Baris ASEAN aggregate
|
|
df_asean = self._build_asean_pillar_rows(df_normed)
|
|
|
|
# Gabung
|
|
df = pd.concat([df_countries, df_asean], ignore_index=True)
|
|
|
|
# Scale 1-100 secara BERSAMA
|
|
df["pillar_country_score_1_100"] = global_minmax(df["pillar_country_norm"])
|
|
|
|
# ---------------------------------------------------------------
|
|
# TAMBAHAN: kolom kondisi pilar (tier + deskripsi dipisah)
|
|
# Dibangkitkan SETELAH score_1_100 tersedia, sehingga tier
|
|
# langsung mencerminkan skor dalam skala akhir 1-100.
|
|
# Referensi tier: GFSI 2022 (Economist Impact); IPC 2019;
|
|
# FAO/CFS 1996/2009; FAO SOFI 2024.
|
|
#
|
|
# Kolom yang dihasilkan:
|
|
# pillar_condition_en — label tier EN, e.g. "At Risk"
|
|
# pillar_condition_id — label tier ID, e.g. "Berisiko"
|
|
# pillar_condition_desc_en — deskripsi kontekstual EN
|
|
# pillar_condition_desc_id — deskripsi kontekstual ID
|
|
# ---------------------------------------------------------------
|
|
conditions = df.apply(
|
|
lambda row: get_pillar_condition(
|
|
row["pillar_name"],
|
|
row["pillar_country_score_1_100"]
|
|
),
|
|
axis=1
|
|
)
|
|
df["pillar_condition_en"] = conditions.apply(lambda x: x[0])
|
|
df["pillar_condition_id"] = conditions.apply(lambda x: x[1])
|
|
df["pillar_condition_desc_en"] = conditions.apply(lambda x: x[2])
|
|
df["pillar_condition_desc_id"] = conditions.apply(lambda x: x[3])
|
|
|
|
# Rank hanya di antara negara asli
|
|
country_only = df[df["country_id"] != ASEAN_COUNTRY_ID].copy()
|
|
country_only["rank_in_pillar_year"] = (
|
|
country_only.groupby(["pillar_id", "year"])["pillar_country_score_1_100"]
|
|
.rank(method="min", ascending=False)
|
|
.astype(int)
|
|
)
|
|
asean_only = df[df["country_id"] == ASEAN_COUNTRY_ID].copy()
|
|
asean_only["rank_in_pillar_year"] = 0 # 0 = ASEAN aggregate
|
|
|
|
df = pd.concat([country_only, asean_only], ignore_index=True)
|
|
df = add_yoy(df, ["country_id", "pillar_id"], "pillar_country_score_1_100")
|
|
|
|
# Tipe data
|
|
df["country_id"] = df["country_id"].astype(int)
|
|
df["pillar_id"] = df["pillar_id"].astype(int)
|
|
df["year"] = df["year"].astype(int)
|
|
df["rank_in_pillar_year"] = df["rank_in_pillar_year"].astype(int)
|
|
df["pillar_country_norm"] = df["pillar_country_norm"].astype(float)
|
|
df["pillar_country_score_1_100"] = df["pillar_country_score_1_100"].astype(float)
|
|
df["pillar_name_id"] = df["pillar_name_id"].astype(str)
|
|
df["country_name_id"] = df["country_name_id"].astype(str)
|
|
df["pillar_condition_en"] = df["pillar_condition_en"].astype(str)
|
|
df["pillar_condition_id"] = df["pillar_condition_id"].astype(str)
|
|
df["pillar_condition_desc_en"] = df["pillar_condition_desc_en"].astype(str)
|
|
df["pillar_condition_desc_id"] = df["pillar_condition_desc_id"].astype(str)
|
|
|
|
self.logger.info(
|
|
f" Total rows: {len(df):,} "
|
|
f"({len(df_countries):,} country + {len(asean_only):,} ASEAN)"
|
|
)
|
|
|
|
# Log distribusi kondisi untuk QA
|
|
self.logger.info("\n Distribusi pillar_condition_en (sample):")
|
|
cond_dist = df["pillar_condition_en"].value_counts().head(10)
|
|
for cond, cnt in cond_dist.items():
|
|
self.logger.info(f" {cnt:>6,} {cond}")
|
|
|
|
schema = [
|
|
bigquery.SchemaField("country_id", "INTEGER", mode="REQUIRED"),
|
|
bigquery.SchemaField("country_name", "STRING", mode="REQUIRED"),
|
|
bigquery.SchemaField("country_name_id", "STRING", mode="REQUIRED"),
|
|
bigquery.SchemaField("pillar_id", "INTEGER", mode="REQUIRED"),
|
|
bigquery.SchemaField("pillar_name", "STRING", mode="REQUIRED"),
|
|
bigquery.SchemaField("pillar_name_id", "STRING", mode="REQUIRED"),
|
|
bigquery.SchemaField("year", "INTEGER", mode="REQUIRED"),
|
|
bigquery.SchemaField("pillar_country_norm", "FLOAT", mode="REQUIRED"),
|
|
bigquery.SchemaField("pillar_country_score_1_100", "FLOAT", mode="REQUIRED"),
|
|
bigquery.SchemaField("rank_in_pillar_year", "INTEGER", mode="REQUIRED"),
|
|
bigquery.SchemaField("year_over_year_change", "FLOAT", mode="NULLABLE"),
|
|
# --- KOLOM KONDISI ---
|
|
# Tier label (GFSI 2022): Secure / Adequate / Moderate / At Risk / Critical
|
|
bigquery.SchemaField("pillar_condition_en", "STRING", mode="REQUIRED"),
|
|
bigquery.SchemaField("pillar_condition_id", "STRING", mode="REQUIRED"),
|
|
# Deskripsi kontekstual per pilar (FAO/CFS 4-pillar; IPC 2019; FAO SOFI 2024)
|
|
bigquery.SchemaField("pillar_condition_desc_en", "STRING", mode="REQUIRED"),
|
|
bigquery.SchemaField("pillar_condition_desc_id", "STRING", mode="REQUIRED"),
|
|
]
|
|
rows = load_to_bigquery(
|
|
self.client, df, table_name, layer='gold',
|
|
write_disposition="WRITE_TRUNCATE", schema=schema
|
|
)
|
|
self._finalize(table_name, rows)
|
|
return df
|
|
|
|
except Exception as e:
|
|
self._fail(table_name, e)
|
|
raise
|
|
|
|
# =========================================================================
|
|
# HELPER: composite per country (untuk framework_by_country)
|
|
# =========================================================================
|
|
|
|
def _calc_country_composite_inmemory(self) -> pd.DataFrame:
|
|
df_normed = self._get_norm_value_df()
|
|
df = (
|
|
df_normed
|
|
.groupby(["country_id", "country_name", "year"])
|
|
.agg(
|
|
composite_score=("norm_value", "mean"),
|
|
n_indicators =("indicator_id", "nunique"),
|
|
)
|
|
.reset_index()
|
|
)
|
|
df["score_1_100"] = global_minmax(df["composite_score"])
|
|
df["rank_in_asean"] = (
|
|
df.groupby("year")["score_1_100"]
|
|
.rank(method="min", ascending=False)
|
|
.astype(int)
|
|
)
|
|
df = add_yoy(df, ["country_id"], "score_1_100")
|
|
df["country_id"] = df["country_id"].astype(int)
|
|
df["year"] = df["year"].astype(int)
|
|
df["n_indicators"] = safe_int(df["n_indicators"], col_name="n_indicators", logger=self.logger)
|
|
df["composite_score"] = df["composite_score"].astype(float)
|
|
df["score_1_100"] = df["score_1_100"].astype(float)
|
|
df["rank_in_asean"] = df["rank_in_asean"].astype(int)
|
|
return df
|
|
|
|
# =========================================================================
|
|
# STEP 3: agg_framework_by_country (termasuk ASEAN)
|
|
# =========================================================================
|
|
|
|
def calc_framework_by_country(self) -> pd.DataFrame:
|
|
table_name = "agg_framework_by_country"
|
|
self.load_metadata[table_name]["start_time"] = datetime.now()
|
|
self.logger.info("\n" + "=" * 70)
|
|
self.logger.info(f"STEP 3: {table_name} -> [Gold] fs_asean_gold")
|
|
self.logger.info(" Termasuk baris ASEAN (country_id=0)")
|
|
self.logger.info("=" * 70)
|
|
|
|
try:
|
|
country_composite = self._calc_country_composite_inmemory()
|
|
df_normed = self._get_norm_value_df()
|
|
parts = []
|
|
|
|
# ---- Per negara (Total) ----
|
|
agg_total = (
|
|
country_composite[[
|
|
"country_id", "country_name", "year",
|
|
"score_1_100", "n_indicators", "composite_score"
|
|
]]
|
|
.copy()
|
|
.rename(columns={
|
|
"score_1_100" : "framework_score_1_100",
|
|
"composite_score": "framework_norm",
|
|
})
|
|
)
|
|
agg_total["framework"] = "Total"
|
|
parts.append(agg_total)
|
|
|
|
pre_sdgs_rows = country_composite[country_composite["year"] < self.sdgs_start_year].copy()
|
|
if not pre_sdgs_rows.empty:
|
|
mdgs_pre = (
|
|
pre_sdgs_rows[[
|
|
"country_id", "country_name", "year",
|
|
"score_1_100", "n_indicators", "composite_score"
|
|
]]
|
|
.copy()
|
|
.rename(columns={
|
|
"score_1_100" : "framework_score_1_100",
|
|
"composite_score": "framework_norm",
|
|
})
|
|
)
|
|
mdgs_pre["framework"] = "MDGs"
|
|
parts.append(mdgs_pre)
|
|
|
|
mdgs_indicator_ids = set(
|
|
self._ind_year_framework[self._ind_year_framework["framework"] == "MDGs"]["indicator_id"]
|
|
)
|
|
if mdgs_indicator_ids:
|
|
df_mdgs_mixed = df_normed[
|
|
(df_normed["indicator_id"].isin(mdgs_indicator_ids)) &
|
|
(df_normed["year"] >= self.sdgs_start_year)
|
|
].copy()
|
|
if not df_mdgs_mixed.empty:
|
|
agg_mdgs_mixed = (
|
|
df_mdgs_mixed
|
|
.groupby(["country_id", "country_name", "year"])
|
|
.agg(
|
|
framework_norm=("norm_value", "mean"),
|
|
n_indicators =("indicator_id", "nunique"),
|
|
)
|
|
.reset_index()
|
|
)
|
|
if not NORMALIZE_FRAMEWORKS_JOINTLY:
|
|
agg_mdgs_mixed["framework_score_1_100"] = global_minmax(agg_mdgs_mixed["framework_norm"])
|
|
agg_mdgs_mixed["framework"] = "MDGs"
|
|
parts.append(agg_mdgs_mixed)
|
|
|
|
sdgs_indicator_ids = set(
|
|
self._ind_year_framework[self._ind_year_framework["framework"] == "SDGs"]["indicator_id"]
|
|
)
|
|
if sdgs_indicator_ids:
|
|
df_sdgs = df_normed[
|
|
(df_normed["indicator_id"].isin(sdgs_indicator_ids)) &
|
|
(df_normed["year"] >= self.sdgs_start_year)
|
|
].copy()
|
|
if not df_sdgs.empty:
|
|
agg_sdgs = (
|
|
df_sdgs
|
|
.groupby(["country_id", "country_name", "year"])
|
|
.agg(
|
|
framework_norm=("norm_value", "mean"),
|
|
n_indicators =("indicator_id", "nunique"),
|
|
)
|
|
.reset_index()
|
|
)
|
|
if not NORMALIZE_FRAMEWORKS_JOINTLY:
|
|
agg_sdgs["framework_score_1_100"] = global_minmax(agg_sdgs["framework_norm"])
|
|
agg_sdgs["framework"] = "SDGs"
|
|
parts.append(agg_sdgs)
|
|
|
|
df_countries = pd.concat(parts, ignore_index=True)
|
|
|
|
# ---- ASEAN aggregate ----
|
|
asean_parts = []
|
|
for fw in df_countries["framework"].unique():
|
|
fw_df = df_countries[
|
|
(df_countries["framework"] == fw) &
|
|
(df_countries["country_id"] != ASEAN_COUNTRY_ID)
|
|
]
|
|
asean_fw = (
|
|
fw_df.groupby(["year", "framework"])
|
|
.agg(
|
|
framework_norm =("framework_norm", "mean"),
|
|
framework_score_1_100 =("framework_score_1_100", "mean"),
|
|
n_indicators =("n_indicators", "mean"),
|
|
)
|
|
.reset_index()
|
|
)
|
|
asean_fw["country_id"] = ASEAN_COUNTRY_ID
|
|
asean_fw["country_name"] = ASEAN_COUNTRY_NAME
|
|
asean_parts.append(asean_fw)
|
|
|
|
df_asean_fw = pd.concat(asean_parts, ignore_index=True)
|
|
|
|
df = pd.concat([df_countries, df_asean_fw], ignore_index=True)
|
|
|
|
if NORMALIZE_FRAMEWORKS_JOINTLY:
|
|
mixed_mask = (df["framework"].isin(["MDGs", "SDGs"])) & (df["year"] >= self.sdgs_start_year)
|
|
if mixed_mask.any():
|
|
df.loc[mixed_mask, "framework_score_1_100"] = global_minmax(df.loc[mixed_mask, "framework_norm"])
|
|
|
|
df = check_and_dedup(df, ["country_id", "framework", "year"], context=table_name, logger=self.logger)
|
|
|
|
country_mask = df["country_id"] != ASEAN_COUNTRY_ID
|
|
df.loc[country_mask, "rank_in_framework_year"] = (
|
|
df[country_mask]
|
|
.groupby(["framework", "year"])["framework_score_1_100"]
|
|
.rank(method="min", ascending=False)
|
|
)
|
|
df.loc[~country_mask, "rank_in_framework_year"] = 0
|
|
|
|
df = add_yoy(df, ["country_id", "framework"], "framework_score_1_100")
|
|
df["country_name_id"] = df["country_name"].apply(translate_country)
|
|
|
|
df["country_id"] = df["country_id"].astype(int)
|
|
df["year"] = df["year"].astype(int)
|
|
df["n_indicators"] = safe_int(df["n_indicators"], col_name="n_indicators", logger=self.logger)
|
|
df["rank_in_framework_year"] = safe_int(df["rank_in_framework_year"], col_name="rank_in_framework_year", logger=self.logger)
|
|
df["framework_norm"] = df["framework_norm"].astype(float)
|
|
df["framework_score_1_100"] = df["framework_score_1_100"].astype(float)
|
|
df["country_name_id"] = df["country_name_id"].astype(str)
|
|
|
|
schema = [
|
|
bigquery.SchemaField("country_id", "INTEGER", mode="REQUIRED"),
|
|
bigquery.SchemaField("country_name", "STRING", mode="REQUIRED"),
|
|
bigquery.SchemaField("country_name_id", "STRING", mode="REQUIRED"),
|
|
bigquery.SchemaField("year", "INTEGER", mode="REQUIRED"),
|
|
bigquery.SchemaField("framework", "STRING", mode="REQUIRED"),
|
|
bigquery.SchemaField("n_indicators", "INTEGER", mode="REQUIRED"),
|
|
bigquery.SchemaField("framework_norm", "FLOAT", mode="REQUIRED"),
|
|
bigquery.SchemaField("framework_score_1_100", "FLOAT", mode="REQUIRED"),
|
|
bigquery.SchemaField("rank_in_framework_year", "INTEGER", mode="REQUIRED"),
|
|
bigquery.SchemaField("year_over_year_change", "FLOAT", mode="NULLABLE"),
|
|
]
|
|
rows = load_to_bigquery(
|
|
self.client, df, table_name, layer='gold',
|
|
write_disposition="WRITE_TRUNCATE", schema=schema
|
|
)
|
|
self._finalize(table_name, rows)
|
|
return df
|
|
|
|
except Exception as e:
|
|
self._fail(table_name, e)
|
|
raise
|
|
|
|
# =========================================================================
|
|
# STEP 4: agg_narrative_pillar (termasuk baris ASEAN)
|
|
# =========================================================================
|
|
|
|
def calc_narrative_pillar(
|
|
self,
|
|
df_pillar_by_country: pd.DataFrame,
|
|
) -> pd.DataFrame:
|
|
table_name = "agg_narrative_pillar"
|
|
self.load_metadata[table_name]["start_time"] = datetime.now()
|
|
self.logger.info("\n" + "=" * 70)
|
|
self.logger.info(f"STEP 4: {table_name} -> [Gold] fs_asean_gold")
|
|
self.logger.info(" Termasuk baris ASEAN (country_id=0)")
|
|
self.logger.info("=" * 70)
|
|
|
|
try:
|
|
records = []
|
|
years = sorted(df_pillar_by_country["year"].unique())
|
|
pillars = df_pillar_by_country["pillar_id"].unique()
|
|
|
|
history = {}
|
|
for (c_id, p_id), grp in df_pillar_by_country.groupby(["country_id", "pillar_id"]):
|
|
history[(c_id, p_id)] = dict(
|
|
zip(grp["year"].astype(int), grp["pillar_country_score_1_100"].astype(float))
|
|
)
|
|
|
|
for yr in years:
|
|
yr_df = df_pillar_by_country[df_pillar_by_country["year"] == yr]
|
|
country_only_yr = yr_df[yr_df["country_id"] != ASEAN_COUNTRY_ID]
|
|
|
|
for p_id in pillars:
|
|
yr_pillar_all = yr_df[yr_df["pillar_id"] == p_id]
|
|
if yr_pillar_all.empty:
|
|
continue
|
|
|
|
p_name_row = yr_pillar_all.iloc[0]
|
|
p_name = str(p_name_row["pillar_name"])
|
|
n_pillars = len(pillars)
|
|
|
|
asean_yr_all_pillars = yr_df[yr_df["country_id"] == ASEAN_COUNTRY_ID]
|
|
asean_sorted = asean_yr_all_pillars.sort_values("pillar_country_score_1_100", ascending=False).reset_index(drop=True)
|
|
|
|
country_pillar_yr = country_only_yr[country_only_yr["pillar_id"] == p_id]
|
|
if not country_pillar_yr.empty:
|
|
top_row = country_pillar_yr.loc[country_pillar_yr["pillar_country_score_1_100"].idxmax()]
|
|
bot_row = country_pillar_yr.loc[country_pillar_yr["pillar_country_score_1_100"].idxmin()]
|
|
top_country = str(top_row["country_name"])
|
|
top_score = round(float(top_row["pillar_country_score_1_100"]), 2)
|
|
bot_country = str(bot_row["country_name"])
|
|
bot_score = round(float(bot_row["pillar_country_score_1_100"]), 2)
|
|
else:
|
|
top_country = bot_country = None
|
|
top_score = bot_score = None
|
|
|
|
for _, row in yr_pillar_all.iterrows():
|
|
c_id = int(row["country_id"])
|
|
c_name = str(row["country_name"])
|
|
c_name_id = translate_country(c_name)
|
|
p_score = float(row["pillar_country_score_1_100"])
|
|
p_yoy = row.get("year_over_year_change", None)
|
|
p_yoy_val = float(p_yoy) if pd.notna(p_yoy) else None
|
|
p_name_id = translate_pillar(p_name)
|
|
is_asean = (c_id == ASEAN_COUNTRY_ID)
|
|
|
|
if is_asean:
|
|
rank_sorted = asean_sorted.reset_index(drop=True)
|
|
p_rank = int(rank_sorted[rank_sorted["pillar_id"] == p_id].index[0]) + 1 if p_id in rank_sorted["pillar_id"].values else 0
|
|
else:
|
|
country_all_pillars = yr_df[yr_df["country_id"] == c_id].sort_values("pillar_country_score_1_100", ascending=False).reset_index(drop=True)
|
|
p_rank = int(country_all_pillars[country_all_pillars["pillar_id"] == p_id].index[0]) + 1 if p_id in country_all_pillars["pillar_id"].values else 0
|
|
|
|
hist_up = {y: s for y, s in history.get((c_id, p_id), {}).items() if y <= yr}
|
|
all_pillar_yr = yr_df[yr_df["country_id"] == c_id][["pillar_name", "pillar_country_score_1_100"]].copy()
|
|
cpa = df_pillar_by_country[
|
|
(df_pillar_by_country["pillar_id"] == p_id) &
|
|
(df_pillar_by_country["country_id"] != ASEAN_COUNTRY_ID)
|
|
][["year", "country_id", "country_name", "pillar_country_score_1_100"]].copy()
|
|
|
|
narrative_en, narrative_id = _build_pillar_narrative(
|
|
year = yr,
|
|
pillar_name = p_name,
|
|
pillar_score = p_score,
|
|
rank_in_year = p_rank,
|
|
n_pillars = n_pillars,
|
|
yoy_val = p_yoy_val,
|
|
top_country = top_country if is_asean else None,
|
|
top_country_score = top_score if is_asean else None,
|
|
bot_country = bot_country if is_asean else None,
|
|
bot_country_score = bot_score if is_asean else None,
|
|
pillar_scores_history = hist_up,
|
|
all_pillar_scores_year= all_pillar_yr,
|
|
country_pillar_all = cpa,
|
|
is_asean = is_asean,
|
|
)
|
|
|
|
# Ambil kondisi dari kolom yang sudah dihitung di df_pillar_by_country
|
|
cond_en = str(row.get("pillar_condition_en", "N/A"))
|
|
cond_id = str(row.get("pillar_condition_id", "N/A"))
|
|
cond_desc_en = str(row.get("pillar_condition_desc_en", "N/A"))
|
|
cond_desc_id = str(row.get("pillar_condition_desc_id", "N/A"))
|
|
|
|
records.append({
|
|
"year": yr,
|
|
"country_id": c_id,
|
|
"country_name": c_name,
|
|
"country_name_id": c_name_id,
|
|
"pillar_id": int(row["pillar_id"]),
|
|
"pillar_name": p_name,
|
|
"pillar_name_id": p_name_id,
|
|
"pillar_score": round(p_score, 2),
|
|
"rank_in_year": p_rank,
|
|
"yoy_change": p_yoy_val,
|
|
"top_country": top_country if is_asean else None,
|
|
"top_country_id": translate_country(top_country) if (is_asean and top_country) else None,
|
|
"top_country_score": top_score if is_asean else None,
|
|
"bottom_country": bot_country if is_asean else None,
|
|
"bottom_country_id": translate_country(bot_country) if (is_asean and bot_country) else None,
|
|
"bottom_country_score": bot_score if is_asean else None,
|
|
"is_asean_aggregate": is_asean,
|
|
"pillar_condition_en": cond_en,
|
|
"pillar_condition_id": cond_id,
|
|
"pillar_condition_desc_en": cond_desc_en,
|
|
"pillar_condition_desc_id": cond_desc_id,
|
|
"narrative_en": narrative_en,
|
|
"narrative_id": narrative_id,
|
|
})
|
|
|
|
df = pd.DataFrame(records)
|
|
df["year"] = df["year"].astype(int)
|
|
df["country_id"] = df["country_id"].astype(int)
|
|
df["pillar_id"] = df["pillar_id"].astype(int)
|
|
df["rank_in_year"] = df["rank_in_year"].astype(int)
|
|
df["is_asean_aggregate"] = df["is_asean_aggregate"].astype(bool)
|
|
df["pillar_name_id"] = df["pillar_name_id"].astype(str)
|
|
df["country_name_id"] = df["country_name_id"].astype(str)
|
|
df["pillar_condition_en"] = df["pillar_condition_en"].astype(str)
|
|
df["pillar_condition_id"] = df["pillar_condition_id"].astype(str)
|
|
df["pillar_condition_desc_en"] = df["pillar_condition_desc_en"].astype(str)
|
|
df["pillar_condition_desc_id"] = df["pillar_condition_desc_id"].astype(str)
|
|
df["narrative_en"] = df["narrative_en"].astype(str)
|
|
df["narrative_id"] = df["narrative_id"].astype(str)
|
|
for col in ["pillar_score", "yoy_change", "top_country_score", "bottom_country_score"]:
|
|
df[col] = pd.to_numeric(df[col], errors="coerce").astype(float)
|
|
|
|
self.logger.info(f"\n Total rows: {len(df):,}")
|
|
self.logger.info(f" ASEAN rows: {df['is_asean_aggregate'].sum():,}")
|
|
self.logger.info(f" Country rows: {(~df['is_asean_aggregate']).sum():,}")
|
|
|
|
schema = [
|
|
bigquery.SchemaField("year", "INTEGER", mode="REQUIRED"),
|
|
bigquery.SchemaField("country_id", "INTEGER", mode="REQUIRED"),
|
|
bigquery.SchemaField("country_name", "STRING", mode="REQUIRED"),
|
|
bigquery.SchemaField("country_name_id", "STRING", mode="REQUIRED"),
|
|
bigquery.SchemaField("pillar_id", "INTEGER", mode="REQUIRED"),
|
|
bigquery.SchemaField("pillar_name", "STRING", mode="REQUIRED"),
|
|
bigquery.SchemaField("pillar_name_id", "STRING", mode="REQUIRED"),
|
|
bigquery.SchemaField("pillar_score", "FLOAT", mode="REQUIRED"),
|
|
bigquery.SchemaField("rank_in_year", "INTEGER", mode="REQUIRED"),
|
|
bigquery.SchemaField("yoy_change", "FLOAT", mode="NULLABLE"),
|
|
bigquery.SchemaField("top_country", "STRING", mode="NULLABLE"),
|
|
bigquery.SchemaField("top_country_id", "STRING", mode="NULLABLE"),
|
|
bigquery.SchemaField("top_country_score", "FLOAT", mode="NULLABLE"),
|
|
bigquery.SchemaField("bottom_country", "STRING", mode="NULLABLE"),
|
|
bigquery.SchemaField("bottom_country_id", "STRING", mode="NULLABLE"),
|
|
bigquery.SchemaField("bottom_country_score", "FLOAT", mode="NULLABLE"),
|
|
bigquery.SchemaField("is_asean_aggregate", "BOOL", mode="REQUIRED"),
|
|
# Tier label: Secure / Adequate / Moderate / At Risk / Critical
|
|
bigquery.SchemaField("pillar_condition_en", "STRING", mode="REQUIRED"),
|
|
bigquery.SchemaField("pillar_condition_id", "STRING", mode="REQUIRED"),
|
|
# Deskripsi kontekstual per pilar
|
|
bigquery.SchemaField("pillar_condition_desc_en", "STRING", mode="REQUIRED"),
|
|
bigquery.SchemaField("pillar_condition_desc_id", "STRING", mode="REQUIRED"),
|
|
bigquery.SchemaField("narrative_en", "STRING", mode="REQUIRED"),
|
|
bigquery.SchemaField("narrative_id", "STRING", mode="REQUIRED"),
|
|
]
|
|
rows = load_to_bigquery(
|
|
self.client, df, table_name, layer='gold',
|
|
write_disposition="WRITE_TRUNCATE", schema=schema,
|
|
)
|
|
self._finalize(table_name, rows)
|
|
return df
|
|
|
|
except Exception as e:
|
|
self._fail(table_name, e)
|
|
raise
|
|
|
|
# =========================================================================
|
|
# HELPERS
|
|
# =========================================================================
|
|
|
|
def _finalize(self, table_name: str, rows_loaded: int):
|
|
end_time = datetime.now()
|
|
start_time = self.load_metadata[table_name].get("start_time")
|
|
self.load_metadata[table_name].update({
|
|
"rows_loaded": rows_loaded,
|
|
"status" : "success",
|
|
"end_time" : end_time,
|
|
})
|
|
log_update(self.client, "DW", table_name, "full_load", rows_loaded)
|
|
try:
|
|
save_etl_metadata(
|
|
self.client,
|
|
self._build_etl_metadata(
|
|
table_name = table_name,
|
|
rows_loaded = rows_loaded,
|
|
start_time = start_time,
|
|
end_time = end_time,
|
|
status = "success",
|
|
)
|
|
)
|
|
except Exception as meta_err:
|
|
self.logger.warning(f" [METADATA WARNING] {table_name}: {meta_err}")
|
|
self.logger.info(f" [OK] {table_name}: {rows_loaded:,} rows -> [Gold] fs_asean_gold")
|
|
|
|
def _fail(self, table_name: str, error: Exception):
|
|
end_time = datetime.now()
|
|
start_time = self.load_metadata[table_name].get("start_time")
|
|
error_msg = str(error)
|
|
self.load_metadata[table_name].update({"status": "failed", "end_time": end_time})
|
|
log_update(self.client, "DW", table_name, "full_load", 0, "failed", error_msg)
|
|
try:
|
|
save_etl_metadata(
|
|
self.client,
|
|
self._build_etl_metadata(
|
|
table_name = table_name,
|
|
rows_loaded = 0,
|
|
start_time = start_time,
|
|
end_time = end_time,
|
|
status = "failed",
|
|
error_msg = error_msg,
|
|
)
|
|
)
|
|
except Exception as meta_err:
|
|
self.logger.warning(f" [METADATA WARNING] {table_name}: {meta_err}")
|
|
self.logger.error(f" [FAIL] {table_name}: {error_msg}")
|
|
|
|
# =========================================================================
|
|
# RUN
|
|
# =========================================================================
|
|
|
|
def run(self):
|
|
start = datetime.now()
|
|
self.logger.info("\n" + "=" * 70)
|
|
self.logger.info("FOOD SECURITY AGGREGATION — 3 TABLES -> fs_asean_gold")
|
|
self.logger.info(" ASEAN aggregate DIGABUNG ke tabel yang sama (country_id=0)")
|
|
self.logger.info(" Kolom baru : pillar_condition_en, pillar_condition_id")
|
|
self.logger.info(f" Performance threshold: {PERFORMANCE_THRESHOLD}")
|
|
self.logger.info(f" Condition tiers (GFSI 2022): >=75 Secure | >=60 Adequate |")
|
|
self.logger.info(f" >=40 Moderate | >=20 At Risk | <20 Critical")
|
|
self.logger.info(f" Narrative style : interpretive, plain text, bilingual EN/ID")
|
|
self.logger.info(f" Sustainability : renamed to 'Food Other' / 'Indikator Tambahan'")
|
|
self.logger.info("=" * 70)
|
|
|
|
self.load_data()
|
|
self.sdgs_start_year = self._detect_sdgs_start_year()
|
|
self._assign_framework_labels()
|
|
|
|
df_pillar_by_country = self.calc_pillar_by_country()
|
|
df_framework_by_country = self.calc_framework_by_country()
|
|
self.calc_narrative_pillar(df_pillar_by_country=df_pillar_by_country)
|
|
|
|
duration = (datetime.now() - start).total_seconds()
|
|
total_rows = sum(m["rows_loaded"] for m in self.load_metadata.values())
|
|
|
|
self.logger.info("\n" + "=" * 70)
|
|
self.logger.info("SELESAI")
|
|
self.logger.info("=" * 70)
|
|
self.logger.info(f" Durasi : {duration:.2f}s")
|
|
self.logger.info(f" Total rows : {total_rows:,}")
|
|
for tbl, meta in self.load_metadata.items():
|
|
icon = "[OK]" if meta["status"] == "success" else "[FAIL]"
|
|
self.logger.info(f" {icon} {tbl:<35} {meta['rows_loaded']:>10,}")
|
|
|
|
|
|
# =============================================================================
|
|
# AIRFLOW TASK
|
|
# =============================================================================
|
|
|
|
def run_aggregation():
|
|
from scripts.bigquery_config import get_bigquery_client
|
|
client = get_bigquery_client()
|
|
agg = FoodSecurityAggregator(client)
|
|
agg.run()
|
|
total = sum(m["rows_loaded"] for m in agg.load_metadata.values())
|
|
print(f"Aggregation completed: {total:,} total rows loaded")
|
|
|
|
|
|
# =============================================================================
|
|
# MAIN
|
|
# =============================================================================
|
|
|
|
if __name__ == "__main__":
|
|
import io
|
|
|
|
if _sys.stdout.encoding and _sys.stdout.encoding.lower() not in ("utf-8", "utf8"):
|
|
_sys.stdout = io.TextIOWrapper(_sys.stdout.buffer, encoding="utf-8", errors="replace")
|
|
if _sys.stderr.encoding and _sys.stderr.encoding.lower() not in ("utf-8", "utf8"):
|
|
_sys.stderr = io.TextIOWrapper(_sys.stderr.buffer, encoding="utf-8", errors="replace")
|
|
|
|
print("=" * 70)
|
|
print("FOOD SECURITY AGGREGATION -> fs_asean_gold")
|
|
print(f" NORMALIZE_FRAMEWORKS_JOINTLY : {NORMALIZE_FRAMEWORKS_JOINTLY}")
|
|
print(f" PERFORMANCE_THRESHOLD : {PERFORMANCE_THRESHOLD}")
|
|
print(f" ASEAN_COUNTRY_ID : {ASEAN_COUNTRY_ID}")
|
|
print(f" Condition tiers (GFSI 2022) : >=75 Secure | >=60 Adequate | >=40 Moderate | >=20 At Risk | <20 Critical")
|
|
print("=" * 70)
|
|
|
|
logger = setup_logging()
|
|
for handler in logger.handlers:
|
|
handler.__class__ = _SafeStreamHandler
|
|
|
|
client = get_bigquery_client()
|
|
agg = FoodSecurityAggregator(client)
|
|
agg.run()
|
|
|
|
print("\n" + "=" * 70)
|
|
print("[OK] SELESAI")
|
|
print("=" * 70) |