Fix/socio economic/provision

blocksnet/analysis/indicators/socio_economic/social_provision/core.py

@@ -1,38 +1,43 @@
 import pandas as pd
+from tqdm import tqdm
 from loguru import logger
 from .indicator import SocialProvisionIndicator
-from .schemas import BlocksSchema
-from blocksnet.config import service_types_config
+from blocksnet.config import service_types_config, log_config
+from blocksnet.analysis.provision import competitive_provision, provision_strong_total
 
 
 def calculate_social_provision_indicators(
-    blocks_df: pd.DataFrame,
+    blocks_df: pd.DataFrame, acc_mx: pd.DataFrame
 ) -> tuple[dict[SocialProvisionIndicator, float], list[SocialProvisionIndicator]]:
 
     result = {}
     missing = []
-    for indicator in SocialProvisionIndicator:
+
+    disable_tqdm = log_config.disable_tqdm
+    logger_level = log_config.logger_level
+    log_config.set_disable_tqdm(True)
+    log_config.set_logger_level("ERROR")
+
+    for indicator in tqdm(list(SocialProvisionIndicator), disable=disable_tqdm):
 
         name = indicator.meta.name
         if not name in service_types_config:
             logger.warning(f"{name} not found in config. The indicator is skipped")
             missing.append(indicator)
             continue
 
-        _, demand, accessibility = service_types_config[name].values()
-
         column = f"capacity_{indicator.meta.name}"
         if not column in blocks_df.columns:
             logger.warning(f"{column} is missing. The indicator is skipped")
             missing.append(indicator)
             continue
 
-        df = BlocksSchema(blocks_df.rename(columns={column: "capacity"}))
-
-        population = df["population"].sum()
-        capacity = df["capacity"].sum()
-        provision = capacity / (population / 1000 * demand)
+        _, demand, accessibility = service_types_config[name].values()
+        df = blocks_df.rename(columns={column: "capacity"})
+        prov_df, _ = competitive_provision(df, acc_mx, accessibility, demand)
+        result[indicator] = provision_strong_total(prov_df)
 
-        result[indicator] = min(1.0, float(provision))
+    log_config.set_disable_tqdm(disable_tqdm)
+    log_config.set_logger_level(logger_level)
 
     return result, missing

blocksnet/analysis/provision/competivive/core.py

@@ -149,11 +149,16 @@ def _distribute_demand(
 
 
 def provision_strong_total(blocks_df: pd.DataFrame):
-    return blocks_df[DEMAND_WITHIN_COLUMN].sum() / blocks_df.demand.sum()
+    demand_within = blocks_df[DEMAND_WITHIN_COLUMN].sum()
+    demand = blocks_df.demand.sum()
+    return float(demand_within / demand)
 
 
 def provision_weak_total(blocks_df: pd.DataFrame):
-    return (blocks_df[DEMAND_WITHIN_COLUMN].sum() + blocks_df[DEMAND_WITHOUT_COLUMN].sum()) / blocks_df.demand.sum()
+    demand_within = blocks_df[DEMAND_WITHIN_COLUMN].sum()
+    demand_without = blocks_df[DEMAND_WITHOUT_COLUMN].sum()
+    demand = blocks_df.demand.sum()
+    return float((demand_within + demand_without) / demand)
 
 
 def competitive_provision(
@@ -173,7 +178,7 @@ def competitive_provision(
 
     links = []
     logger.info("Setting and solving LP problems until max depth or break condition reached")
-    for depth in tqdm(range(1, max_depth + 1), disable=log_config.disable_tqdm):
+    for depth in range(1, max_depth + 1):
         blocks_df, depth_links = _distribute_demand(blocks_df, accessibility_matrix, accessibility, depth)
         links.extend(depth_links)
         break_condition = blocks_df[DEMAND_LEFT_COLUMN].sum() == 0 or blocks_df[CAPACITY_LEFT_COLUMN].sum() == 0

blocksnet/preprocessing/imputing/__init__.py

@@ -1,3 +1,4 @@
 from .buildings import *
 from .services import *
 from .development import *
+from .population import *

blocksnet/preprocessing/imputing/population/__init__.py

@@ -0,0 +1 @@
+from .core import impute_population

blocksnet/preprocessing/imputing/population/core.py

@@ -0,0 +1,51 @@
+import pandas as pd
+import numpy as np
+from loguru import logger
+from .schemas import Schema
+
+
+def _impute_population(df: pd.DataFrame, total_population: int) -> pd.DataFrame:
+    df = df.copy()
+    weights = df["living_area"] / df["living_area"].sum()
+    ideal = weights * total_population
+
+    alloc = np.floor(ideal).astype(int)
+    remainder = int(total_population - alloc.sum())
+    fractions = ideal - alloc
+
+    sorted_fractions = fractions.sort_values(ascending=False).head(remainder)
+    fractions_idx = sorted_fractions.index
+
+    alloc.loc[fractions_idx] += 1
+
+    df["population"] = alloc
+    return df
+
+
+def impute_population(df: pd.DataFrame, total_population: int) -> pd.DataFrame:
+    if not isinstance(total_population, int):
+        raise TypeError("Total population must be int")
+    if total_population <= 0:
+        raise ValueError("Total population must be greater than 0")
+    df = Schema(df)
+
+    sum_population = int(df["population"].sum())
+    delta_population = total_population - sum_population
+    if delta_population < 0:
+        raise ValueError(
+            f"Total population must be greater than population sum, got:\n{total_population} - {sum_population} = {delta_population}"
+        )
+    if delta_population == 0:
+        logger.warning("0 population to distribute")
+        return df.fillna(0)
+
+    idx = df[(df["population"].isna()) | ((df["population"] == 0) & (df["living_area"] > 0))].index
+    if len(idx) == 0:
+        logger.warning("No unknown population found")
+        return df
+
+    logger.info(f"Distributing {delta_population} population between {len(idx)} rows")
+
+    sub_df = _impute_population(df.loc[idx], delta_population)
+    df.loc[idx, df.columns] = sub_df
+    return df

blocksnet/preprocessing/imputing/population/schemas.py

@@ -0,0 +1,8 @@
+from pandera import Field
+from pandera.typing import Series
+from blocksnet.utils.validation import DfSchema
+
+
+class Schema(DfSchema):
+    living_area: Series[float] = Field(ge=0)
+    population: Series[float] = Field(ge=0, nullable=True)