Initial parameters

from blocksnet.optimization.services import (
    TPEOptimizer,
    WeightedObjective,
    WeightedConstraints,
    Facade,
    BlockSolution,
    GradientChooser,
    AreaSolution,
)
from blocksnet.enums import LandUse
import numpy as np
import pandas as pd
import geopandas as gpd
import ast
import csv

Services

BASIC_SERVICES = {
    "kindergarten": 0.2,
    "school": 0.2,
    "pharmacy": 0.2,
    "polyclinics": 0.3,
    "convenience": 0.2,
    "cafe": 0.15,
    "playground": 0.2,
    "post": 0.1,
    "hairdresser": 0.1
}
ADVANCED_SERVICES = {
    "hospital": 0.5,
    "supermarket": 0.4,
    "restaurant": 0.3,
    "park": 0.8,
    "bank": 0.3,
    "fuel": 0.25,
    "police": 0.2,
    "cinema": 0.2,
    "market": 0.2,
    "religion": 0.5,
    "multifunctional_center": 0.1
}
COMFORT_SERVICES = {
    "mall": 0.3,
    "swimming_pool": 0.4,
    "theatre": 0.2,
    "museum": 0.2,
    "university": 0.2,
    "sanatorium": 0.2,
}

Block id for profiling

PROFILING_ID = 1536

City Initialization

blocks = gpd.read_parquet('./../../data/blocks.parquet')
columns = [c for c in blocks.columns if 'capacity_' in c]
blocks = blocks[['geometry', 'population', 'site_area', *columns]].copy()
blocks.head(3)

accessibility_matrix = pd.read_pickle('./../../data/accessibility_matrix.pickle')
accessibility_matrix.head()

Service weights

def get_service_weight(service_name):
    if service_name in BASIC_SERVICES:
        return 0.5714 * BASIC_SERVICES[service_name] / sum(BASIC_SERVICES.values())
    elif service_name in ADVANCED_SERVICES:
        return 0.2857 * ADVANCED_SERVICES[service_name] / sum(ADVANCED_SERVICES.values())
    elif service_name in COMFORT_SERVICES:
        return 0.1429 * COMFORT_SERVICES[service_name] / sum(COMFORT_SERVICES.values())
    return 0

chosen_service_types = set(BASIC_SERVICES) | set(ADVANCED_SERVICES) | set(COMFORT_SERVICES)
service_weights = {service_type: get_service_weight(service_type) for service_type in chosen_service_types if f'capacity_{service_type}' in blocks.columns}

Save service to csv

def save_max_capacities(land_use: LandUse, facade: Facade):
    with open(f'{land_use.name}_max_capacities.csv', 'w', newline='') as csvfile:
        capacities = facade.get_max_capacities(PROFILING_ID)

        writer = csv.writer(csvfile)
        writer.writerow(["Service type", "Capacity required"])  # Write header row

        for service, capacity in capacities.items():
            writer.writerow([service, capacity])

import os
import shutil
from pathlib import Path

def move_csv_to_results():
    results_dir = Path("./results")
    results_dir.mkdir(exist_ok=True)

    csv_files = [f for f in os.listdir() if f.endswith('.csv')]

    for csv_file in csv_files:
        src = Path(csv_file)
        dest = results_dir / csv_file
        shutil.move(str(src), str(dest))

Land use provision with selected services

from tqdm import tqdm

provision_data = {land_use: dict() for land_use in LandUse}

for land_use in LandUse:
    blocks_lu = dict()

    blocks_lu[PROFILING_ID] = land_use

    var_adapter = AreaSolution(blocks_lu)

    facade = Facade(
        blocks_lu=blocks_lu,
        blocks_df=blocks,
        accessibility_matrix=accessibility_matrix,
        var_adapter=var_adapter,
    )

    for service_type, weight in tqdm(service_weights.items()):
        facade.add_service_type(service_type, weight, blocks.rename(columns={f'capacity_{service_type}': 'capacity'})[['capacity']])

    start_prov = facade.start_provisions

    objective = WeightedObjective(num_params=facade.num_params, facade=facade, weights=service_weights, max_evals=50)

    constraints = WeightedConstraints(num_params=facade.num_params, facade=facade)

    tpe_optimizer = TPEOptimizer(
        objective=objective, constraints=constraints, vars_chooser=SimpleChooser(facade, facade.num_params, 5)
    )

    best_x, best_val, perc, func_evals = tpe_optimizer.run(max_runs=1000, timeout=60000, initial_runs_num=1)

    solution_df = facade.solution_to_services_df(best_x)
    solution_df.to_csv(f'{land_use.name}_services.csv')

    save_max_capacities(land_use, facade)

    df = pd.read_csv('tpe_provisions.csv')
    best_provisions = None
    max_val = 0
    for index, row in df.iterrows():
        if max_val < row['best_val']:
            max_val = row['best_val']
            best_provisions = row['provisions']
    best_provisions = ast.literal_eval(best_provisions)
    for service in facade._chosen_service_types:
        if service in best_provisions.keys():
            opt = best_provisions[service]
        else:
            opt = start_prov[service]
        start = start_prov[service]
        provision_data[land_use][service] = (start, opt)

    move_csv_to_results()
    break

with open('./results/provisions_data.csv', 'w', newline='') as csvfile:
    writer = csv.writer(csvfile)
    writer.writerow(["Land use", "Service type", "Start provisions", "Opt. provisions", "Delta provisions"])  # Write header row

    for land_use, services in provision_data.items():
        for service_type, (start_prov, opt_prov) in services.items():
            writer.writerow([land_use, service_type, start_prov, opt_prov, opt_prov - start_prov])