From 10f4796be24e1c093be7649549b93ee74468b4b8 Mon Sep 17 00:00:00 2001 From: atul_bhardwaj Date: Fri, 20 Feb 2026 17:21:43 +0530 Subject: [PATCH] Apply custom thresholds and weights to Streamlit IQB score computation --- library/src/iqb/calculator.py | 32 +++++--- library/tests/iqb/calculator_test.py | 64 +++++++++++++++ prototype/Home.py | 9 ++- prototype/pages/IQB_Map.py | 95 +++++++++++++++-------- prototype/tests/calculation_utils_test.py | 48 ++++++++++++ prototype/utils/calculation_utils.py | 11 ++- 6 files changed, 212 insertions(+), 47 deletions(-) create mode 100644 prototype/tests/calculation_utils_test.py diff --git a/library/src/iqb/calculator.py b/library/src/iqb/calculator.py index e9f5611..0b58b0d 100644 --- a/library/src/iqb/calculator.py +++ b/library/src/iqb/calculator.py @@ -1,5 +1,6 @@ """Module that implements calculating IQB scores.""" +import copy from pprint import pprint from .config import IQB_CONFIG @@ -13,21 +14,30 @@ def __init__(self, config=None, name=None): Initialize a new instance of IQBCalculator. Parameters: - config (str): The file with the configuration of the IQB formula parameters. If "None" (default), it gets the parameters from the IQB_CONFIG dict. + config (dict | str): If "None" (default), use IQB_CONFIG. A dict is used + as in-memory configuration. A string is currently treated as a + configuration file path and is not implemented. name (str): [Optional] name for the IQBCalculator instance. """ self.set_config(config) self.name = name def set_config(self, config): - """Sets up configuration parameters. If "None" (default), it gets the parameters from the IQB_CONFIG dict.""" + """Set up configuration parameters.""" if config is None: self.config = IQB_CONFIG - else: + elif isinstance(config, dict): + # Keep calculator behavior isolated from external mutations. + self.config = copy.deepcopy(config) + elif isinstance(config, str): # TODO: load config data from file (json, yaml, or other format) as a dict raise NotImplementedError( "method for reading from configuration file other than the default not implemented" ) + else: + raise TypeError( + f"config must be None, dict, or str, got: {type(config).__name__}" + ) def print_config(self): """ @@ -36,27 +46,27 @@ def print_config(self): """ # TODO: to be updated print("### IQB formula weights and thresholds") - pprint(IQB_CONFIG) + pprint(self.config) print() print("### Use cases") - for uc in IQB_CONFIG["use cases"]: + for uc in self.config["use cases"]: print(f"\t{uc}") print() print("### Network requirements") - for uc in IQB_CONFIG["use cases"]: - for nr in IQB_CONFIG["use cases"][uc]["network requirements"]: + for uc in self.config["use cases"]: + for nr in self.config["use cases"][uc]["network requirements"]: print(f"\t{nr}") break print() print("### Weights & Thresholds") print("\tUse case\t \tNetwork requirement \tWeight \tThreshold min") - for uc in IQB_CONFIG["use cases"]: - for nr in IQB_CONFIG["use cases"][uc]["network requirements"]: - nr_w = IQB_CONFIG["use cases"][uc]["network requirements"][nr]["w"] - nr_th = IQB_CONFIG["use cases"][uc]["network requirements"][nr]["threshold min"] + for uc in self.config["use cases"]: + for nr in self.config["use cases"][uc]["network requirements"]: + nr_w = self.config["use cases"][uc]["network requirements"][nr]["w"] + nr_th = self.config["use cases"][uc]["network requirements"][nr]["threshold min"] print(f"\t{uc:20} \t{nr:20} \t{nr_w} \t{nr_th}") print() diff --git a/library/tests/iqb/calculator_test.py b/library/tests/iqb/calculator_test.py index 9a11762..88b6f0d 100644 --- a/library/tests/iqb/calculator_test.py +++ b/library/tests/iqb/calculator_test.py @@ -1,5 +1,7 @@ """Tests for the IQBCalculator score calculation module.""" +import copy + import pytest from iqb import IQB_CONFIG, IQBCalculator @@ -23,6 +25,22 @@ def test_init_uses_default_config(self): iqb = IQBCalculator() assert iqb.config == IQB_CONFIG + def test_init_accepts_dict_config(self): + """Test that IQBCalculator accepts an in-memory config dict.""" + custom_config = copy.deepcopy(IQB_CONFIG) + custom_config["use cases"]["web browsing"]["network requirements"][ + "download_throughput_mbps" + ]["threshold min"] = 500 + + iqb = IQBCalculator(config=custom_config) + + assert ( + iqb.config["use cases"]["web browsing"]["network requirements"][ + "download_throughput_mbps" + ]["threshold min"] + == 500 + ) + def test_init_with_invalid_config_raises_error(self): """Test that providing a non-existent config file raises NotImplementedError.""" with pytest.raises(NotImplementedError): @@ -127,6 +145,36 @@ def test_calculate_iqb_score_consistency(self): score2 = iqb.calculate_iqb_score() assert score1 == score2 + def test_calculate_iqb_score_changes_with_custom_thresholds(self): + """Test that stricter thresholds lower the IQB score for the same data.""" + sample_data = { + "m-lab": { + "download_throughput_mbps": 15, + "upload_throughput_mbps": 20, + "latency_ms": 75, + "packet_loss": 0.007, + } + } + + default_score = IQBCalculator().calculate_iqb_score(data=sample_data) + + strict_config = copy.deepcopy(IQB_CONFIG) + for use_case in strict_config["use cases"].values(): + use_case["network requirements"]["download_throughput_mbps"]["threshold min"] = ( + 500 + ) + use_case["network requirements"]["upload_throughput_mbps"]["threshold min"] = ( + 500 + ) + use_case["network requirements"]["latency_ms"]["threshold min"] = 5 + use_case["network requirements"]["packet_loss"]["threshold min"] = 0.001 + + strict_score = IQBCalculator(config=strict_config).calculate_iqb_score( + data=sample_data + ) + + assert strict_score < default_score + class TestIQBCalculatorConfig: """Tests for IQBCalculator configuration.""" @@ -179,8 +227,24 @@ def test_set_config_with_none(self): iqb.set_config(None) assert iqb.config == IQB_CONFIG + def test_set_config_with_dict(self): + """Test that set_config accepts a dict.""" + iqb = IQBCalculator() + custom_config = copy.deepcopy(IQB_CONFIG) + custom_config["use cases"]["gaming"]["w"] = 2 + + iqb.set_config(custom_config) + + assert iqb.config["use cases"]["gaming"]["w"] == 2 + def test_set_config_with_file_raises_error(self): """Test that set_config raises error for file paths.""" iqb = IQBCalculator() with pytest.raises(NotImplementedError): iqb.set_config("some_file.json") + + def test_set_config_with_invalid_type_raises_type_error(self): + """Test that set_config rejects unsupported config types.""" + iqb = IQBCalculator() + with pytest.raises(TypeError): + iqb.set_config(123) diff --git a/prototype/Home.py b/prototype/Home.py index b2a806c..15e4b51 100644 --- a/prototype/Home.py +++ b/prototype/Home.py @@ -2,7 +2,10 @@ import streamlit as st from session_state import initialize_app_state -from utils.calculation_utils import build_data_for_calculate +from utils.calculation_utils import ( + build_data_for_calculate, + calculate_iqb_score_with_custom_settings, +) from visualizations.sunburst_data import ( prepare_complete_hierarchy_sunburst_data, prepare_requirements_sunburst_data, @@ -69,8 +72,8 @@ def main(): with right_col: try: data_for_calculation = build_data_for_calculate(state) - iqb_score = state.iqb.calculate_iqb_score( - data=data_for_calculation, print_details=False + iqb_score = calculate_iqb_score_with_custom_settings( + state, data=data_for_calculation, print_details=False ) tab1, tab2, tab3 = st.tabs(["Requirements", "Use Cases", "Full Hierarchy"]) diff --git a/prototype/pages/IQB_Map.py b/prototype/pages/IQB_Map.py index 0748333..9e6edda 100644 --- a/prototype/pages/IQB_Map.py +++ b/prototype/pages/IQB_Map.py @@ -20,10 +20,14 @@ import pycountry import streamlit as st from dacite import from_dict -from iqb import IQBCache, IQBDatasetGranularity +from iqb import IQB, IQBCache, IQBDatasetGranularity from iqb.ghremote.cache import IQBRemoteCache, Manifest, data_dir_or_default from plotly.subplots import make_subplots from session_state import initialize_app_state +from utils.calculation_utils import ( + calculate_iqb_score_with_custom_settings, + get_config_with_custom_settings, +) from visualizations.sunburst_data import ( prepare_complete_hierarchy_sunburst_data, prepare_requirements_sunburst_data, @@ -175,6 +179,7 @@ def _load_manifest_from_url(self, url: str) -> Manifest: st.session_state.selected_percentile = "p95" state = st.session_state.app_state +custom_config = get_config_with_custom_settings(state) # --- Helper Functions --- @@ -250,27 +255,15 @@ def build_iqb_data_from_cache(metrics: dict, percentile: str = "p95") -> dict: def calculate_iqb_score_from_metrics( - metrics: dict, percentile: str = "p95" + metrics: dict, percentile: str = "p95", custom_config: dict | None = None ) -> float | None: """Calculate actual IQB score from cache metrics.""" if not metrics: return None try: - temp_state = initialize_app_state() - temp_state.manual_entry["m-lab"]["download_throughput_mbps"] = metrics[ - "download_throughput_mbps" - ][percentile] - temp_state.manual_entry["m-lab"]["upload_throughput_mbps"] = metrics[ - "upload_throughput_mbps" - ][percentile] - temp_state.manual_entry["m-lab"]["latency_ms"] = metrics["latency_ms"][ - percentile - ] - temp_state.manual_entry["m-lab"]["packet_loss"] = metrics["packet_loss"][ - percentile - ] iqb_data = build_iqb_data_from_cache(metrics, percentile) - return temp_state.iqb.calculate_iqb_score(data=iqb_data, print_details=False) + calculator = IQB(config=custom_config) if custom_config is not None else IQB() + return calculator.calculate_iqb_score(data=iqb_data, print_details=False) except Exception: return None @@ -479,6 +472,7 @@ def load_historical_data( available_periods: list[tuple[str, str, str]], percentile: str = "p95", subdivision_code: str | None = None, + custom_config: dict | None = None, ) -> pd.DataFrame: """Load historical data for a country or subdivision across all available periods.""" rows = [] @@ -558,7 +552,7 @@ def load_historical_data( "latency_ms": {percentile: float(latency)}, "packet_loss": {percentile: float(packet_loss)}, } - score = calculate_iqb_score_from_metrics(metrics, percentile) + score = calculate_iqb_score_from_metrics(metrics, percentile, custom_config) rows.append( { @@ -583,13 +577,17 @@ def load_historical_data( @st.cache_data def load_country_data_for_date( - _cache: IQBCache, start_date: str, end_date: str, percentile: str = "p95" + _cache: IQBCache, + start_date: str, + end_date: str, + percentile: str = "p95", + custom_config: dict | None = None, ) -> dict[str, dict]: """Load and enrich country data with IQB scores for a specific date range.""" raw_data = fetch_map_data(_cache, start_date, end_date) for iso_a3, data in raw_data.items(): metrics = data["metrics"] - score = calculate_iqb_score_from_metrics(metrics, percentile) + score = calculate_iqb_score_from_metrics(metrics, percentile, custom_config) data["score"] = score data["download"] = metrics["download_throughput_mbps"].get(percentile) data["upload"] = metrics["upload_throughput_mbps"].get(percentile) @@ -1085,6 +1083,7 @@ def load_historical_data_city( city_name: str, available_periods: list[tuple[str, str, str]], percentile: str = "p95", + custom_config: dict | None = None, ) -> pd.DataFrame: """Load historical data for a specific city across all available periods.""" rows = [] @@ -1134,7 +1133,7 @@ def load_historical_data_city( "packet_loss": {percentile: float(dl_row[f"loss_p{p_num}"])}, } - score = calculate_iqb_score_from_metrics(metrics, percentile) + score = calculate_iqb_score_from_metrics(metrics, percentile, custom_config) rows.append( { @@ -1168,6 +1167,7 @@ def create_trend_charts( subdivision_data: dict | None = None, start_date: str | None = None, end_date: str | None = None, + custom_config: dict | None = None, ): """Create historical trend charts for a country or subdivision with optional comparison.""" @@ -1190,7 +1190,12 @@ def create_trend_charts( # Load primary data df = load_historical_data( - cache, country_code, periods_to_load, percentile, subdivision_code + cache, + country_code, + periods_to_load, + percentile, + subdivision_code, + custom_config, ) if df.empty: @@ -1209,16 +1214,27 @@ def create_trend_charts( compare_city, periods_to_load, percentile, + custom_config, ) elif compare_subdiv: # Subdivision-level comparison compare_df = load_historical_data( - cache, compare_country, periods_to_load, percentile, compare_subdiv + cache, + compare_country, + periods_to_load, + percentile, + compare_subdiv, + custom_config, ) else: # Country-level comparison compare_df = load_historical_data( - cache, compare_country, periods_to_load, percentile, None + cache, + compare_country, + periods_to_load, + percentile, + None, + custom_config, ) if compare_df is not None and compare_df.empty: @@ -1317,6 +1333,7 @@ def _build_subdivision_map_data( geojson: dict, subdivision_data: dict[str, dict], percentile: str, + custom_config: dict | None = None, ) -> tuple[list, list, list, list, dict]: """Build data arrays for subdivision choropleth map.""" name_to_data = { @@ -1345,7 +1362,7 @@ def _build_subdivision_map_data( if matched_data: metrics = matched_data["metrics"] - score = calculate_iqb_score_from_metrics(metrics, percentile) + score = calculate_iqb_score_from_metrics(metrics, percentile, custom_config) locations.append(location_id) z_values.append(score if score is not None else 0) customdata.append(region_code or iso_code or location_id) @@ -1378,13 +1395,14 @@ def create_subdivision_map( selected_subdivision: str | None = None, city_coords: dict[str, tuple[float, float]] | None = None, city_data: dict | None = None, + custom_config: dict | None = None, ) -> go.Figure | None: """Create choropleth map of subdivisions, optionally with city overlay.""" if not geojson or not subdivision_data: return None locations, z_values, hover_texts, customdata, _ = _build_subdivision_map_data( - geojson, subdivision_data, percentile + geojson, subdivision_data, percentile, custom_config ) # Always show the choropleth layer @@ -1423,7 +1441,9 @@ def create_subdivision_map( for city, (lat, lon) in city_coords.items(): if data := city_name_to_data.get(city): - score = calculate_iqb_score_from_metrics(data["metrics"], percentile) + score = calculate_iqb_score_from_metrics( + data["metrics"], percentile, custom_config + ) samples = data["sample_counts"].get("downloads", 0) lats.append(lat) lons.append(lon) @@ -1507,7 +1527,11 @@ def create_subdivision_map( # Main content with st.spinner("Loading data..."): country_data = load_country_data_for_date( - cache, START_DATE, END_DATE, st.session_state.selected_percentile + cache, + START_DATE, + END_DATE, + st.session_state.selected_percentile, + custom_config, ) if not country_data: @@ -1587,6 +1611,7 @@ def create_subdivision_map( selected_code, city_coords, filtered_city_data, + custom_config, ) if fig: event = st.plotly_chart( @@ -1629,8 +1654,8 @@ def create_subdivision_map( tab1, tab2, tab3 = st.tabs(["Requirements", "Use Cases", "Full Hierarchy"]) try: iqb_data = build_iqb_data_from_cache(metrics, percentile) - iqb_score = state.iqb.calculate_iqb_score( - data=iqb_data, print_details=False + iqb_score = calculate_iqb_score_with_custom_settings( + state, data=iqb_data, print_details=False ) with tab1: render_sunburst( @@ -1703,6 +1728,7 @@ def highlight_selected(row): subdivision_data=subdivision_data, start_date=START_DATE, end_date=END_DATE, + custom_config=custom_config, ) # LEVEL 2: Subdivision view @@ -1719,7 +1745,11 @@ def highlight_selected(row): ] fig = create_subdivision_map( - admin1_geojson, subdivision_data, iso_a3, percentile + admin1_geojson, + subdivision_data, + iso_a3, + percentile, + custom_config=custom_config, ) if fig: event = st.plotly_chart( @@ -1763,8 +1793,8 @@ def highlight_selected(row): tab1, tab2, tab3 = st.tabs(["Requirements", "Use Cases", "Full Hierarchy"]) try: iqb_data = build_iqb_data_from_cache(metrics, percentile) - iqb_score = state.iqb.calculate_iqb_score( - data=iqb_data, print_details=False + iqb_score = calculate_iqb_score_with_custom_settings( + state, data=iqb_data, print_details=False ) with tab1: render_sunburst( @@ -1836,6 +1866,7 @@ def highlight_selected(row): subdivision_data=subdivision_data, start_date=START_DATE, end_date=END_DATE, + custom_config=custom_config, ) # LEVEL 1: World map diff --git a/prototype/tests/calculation_utils_test.py b/prototype/tests/calculation_utils_test.py new file mode 100644 index 0000000..da2e52e --- /dev/null +++ b/prototype/tests/calculation_utils_test.py @@ -0,0 +1,48 @@ +"""Tests for prototype calculation utilities.""" + +from session_state import initialize_app_state +from utils.calculation_utils import ( + build_data_for_calculate, + calculate_iqb_score_with_custom_settings, + get_config_with_custom_settings, +) + + +def test_get_config_with_custom_settings_applies_state_values(): + """Config builder should apply threshold and weight overrides from state.""" + state = initialize_app_state() + + state.thresholds["web browsing"]["download_throughput_mbps"] = 500 + state.requirement_weights["web browsing"]["latency_ms"] = 1 + state.use_case_weights["gaming"] = 2 + + config = get_config_with_custom_settings(state) + + assert ( + config["use cases"]["web browsing"]["network requirements"][ + "download_throughput_mbps" + ]["threshold min"] + == 500 + ) + assert ( + config["use cases"]["web browsing"]["network requirements"]["latency_ms"]["w"] == 1 + ) + assert config["use cases"]["gaming"]["w"] == 2 + + +def test_calculate_iqb_score_with_custom_settings_changes_result(): + """Applying strict thresholds in state should lower score for same input data.""" + state = initialize_app_state() + data = build_data_for_calculate(state) + + baseline_score = calculate_iqb_score_with_custom_settings(state, data) + + for use_case in state.thresholds.values(): + use_case["download_throughput_mbps"] = 500 + use_case["upload_throughput_mbps"] = 500 + use_case["latency_ms"] = 5 + use_case["packet_loss"] = 0.001 + + strict_score = calculate_iqb_score_with_custom_settings(state, data) + + assert strict_score < baseline_score diff --git a/prototype/utils/calculation_utils.py b/prototype/utils/calculation_utils.py index 2b3c87b..30a564d 100644 --- a/prototype/utils/calculation_utils.py +++ b/prototype/utils/calculation_utils.py @@ -4,7 +4,7 @@ from typing import Dict, Tuple from app_state import IQBAppState -from iqb import IQB_CONFIG +from iqb import IQB, IQB_CONFIG from utils.data_utils import get_available_datasets, get_available_requirements @@ -68,6 +68,15 @@ def get_config_with_custom_settings(state: IQBAppState) -> Dict: return modified_config +def calculate_iqb_score_with_custom_settings( + state: IQBAppState, data: Dict[str, Dict[str, float]], print_details: bool = False +) -> float: + """Calculate IQB score using current UI-customized thresholds and weights.""" + custom_config = get_config_with_custom_settings(state) + calculator = IQB(config=custom_config) + return calculator.calculate_iqb_score(data=data, print_details=print_details) + + def calculate_component_importance() -> Dict[str, float]: """Calculate the importance of each network component for visualization.""" importance = {}