app.py

import gradio as gr
import pandas as pd
import numpy as np
import json
from io import StringIO
from collections import OrderedDict

def test(input_json):
    print("Received input")
    # Parse the input JSON string
    try:
        inputs = json.loads(input_json)
    except json.JSONDecodeError:
        inputs = json.loads(input_json.replace("'", '"'))

    # Accessing the lists 
    ids_index = inputs['input']['ids_list']
    weightsNames = inputs['input']["weights_names"]
    
    # Extract the datatree part which is a list of dictionaries
    matrix = inputs['input']["matrix"]
    weights = inputs['input']["weights"]
    

    alpha = inputs['input']["alpha"]
    alpha = float(alpha)
    threshold = inputs['input']["threshold"]
    threshold = float(threshold)
    
    df_matrix = pd.DataFrame(matrix).T
    df_weights = pd.DataFrame(weights).T
    df_matrix = df_matrix.round(0).astype(int)
    df_weights = df_weights.round(0).astype(int)


    
    def computeAccessibility (DistanceMatrix,weightsNames, destinationWeights=None,alpha = 0.0038, threshold = 600):
    
        decay_factors = np.exp(-alpha * DistanceMatrix) * (DistanceMatrix <= threshold)
        subdomainsAccessibility = pd.DataFrame(index=DistanceMatrix.index, columns=weightsNames) #destinationWeights.columns)
        # for weighted accessibility (e. g. areas)
        if not destinationWeights.empty:
            for col,columnName in zip(destinationWeights.columns, weightsNames):
                subdomainsAccessibility[columnName] = (decay_factors * destinationWeights[col].values).sum(axis=1)
        # for unweighted accessibility (e. g. points of interest)
        else:
            for columnName in weightsNames:
                subdomainsAccessibility[columnName] = (decay_factors * 1).sum(axis=1)
        
        return subdomainsAccessibility
    
    subdomainsAccessibility = computeAccessibility(df_matrix,weightsNames,df_weights,alpha,threshold)

    # make a dictionary to output in grasshopper / etc
    subdomainsAccessibility_dictionary = subdomainsAccessibility.to_dict('index')


    def remap(value, B_min, B_max, C_min, C_max):
        return C_min + (((value - B_min) / (B_max - B_min))* (C_max - C_min))    


    
    def accessibilityToLivability (DistanceMatrix,subdomainsAccessibility, SubdomainAttributeDict):
    
    """
    Converts accessibility measures into livability scores for various urban subdomains
    using a specified scaling mechanism based on predefined thresholds and maximum points.
    
    This function takes a DataFrame of total accessibility per subdomain and remaps these values
    into livability scores based on thresholds and maximum scores provided in a dictionary.
    The output DataFrame retains the original order of indices from a reference distance matrix.
    New columns for combined values such as 'social infrastructure' and 'transportation' are added,
    aggregating scores from relevant subdomains.
    
    Parameters:
    - DistanceMatrix (pd.DataFrame): DataFrame used to maintain the order of indices.
    - totalAccessibility (pd.DataFrame): DataFrame containing accessibility scores for various subdomains.
    - SubdomainAttributeDict (dict): Dictionary where each key is a subdomain and each value is a list
    where the first element is the minimum threshold for good accessibility, and the second element is
    the maximum livability score for that threshold.
    
    Returns:
    - pd.DataFrame: A new DataFrame with the same indices as DistanceMatrix and columns corresponding to
    totalAccessibility, enhanced with additional columns for combined livability metrics.
    
    The function processes each subdomain defined in SubdomainAttributeDict. If the accessibility in a
    subdomain exceeds the threshold, the maximum livability score is assigned. Otherwise, a livability
    score is calculated based on linear interpolation between 0 and the threshold. Combined metrics
    for broader categories like 'social infrastructure' are computed by summing up relevant subdomain
    scores.
    
    Example:
    --------
    # Define the DistanceMatrix and totalAccessibility with example data
    DistanceMatrix = pd.DataFrame(index=[0, 1, 2])
    totalAccessibility = pd.DataFrame({'jobs': [100, 150, 200], 'health': [80, 90, 95]}, index=[0, 1, 2])
    SubdomainAttributeDict = {'jobs': [100, 50], 'health': [80, 40]}
    
    # Call the function
    livability_scores = accessibilityToLivability(DistanceMatrix, totalAccessibility, SubdomainAttributeDict)
    print(livability_scores)
    
    Notes:
    ------
    - The function assumes all columns in totalAccessibility are represented in SubdomainAttributeDict unless
    explicitly handled otherwise within the function.
    """
    
    livability = pd.DataFrame(index=DistanceMatrix.index, columns=subdomainsAccessibility.columns)
    #      livability["Workplaces"] = 0
    livability.fillna(0, inplace=True)
    
    
    # find a set of unique domains, to which subdomains are aggregated
    
    temp = []

    for key, values in SubdomainAttributeDict.items():
      domain = SubdomainAttributeDict[key]['domain']
      for item in domain:
        if ',' in item:
          domain_list = item.split(',')
          SubdomainAttributeDict[key]['domain'] = domain_list
          for domain in domain_list:
            temp.append(domain) 
        else:
          if item != 0: 
              temp.append(item)  
    
    domainsUnique = list(set(temp))
         
    for domain in domainsUnique:
        livability[domain] = 0


    # remap accessibility to livability points
    
    for key, values in SubdomainAttributeDict.items():
        threshold = float(SubdomainAttributeDict[key]['thresholds'])
        max_livability = float(SubdomainAttributeDict[key]['max_points'])
        domain = SubdomainAttributeDict[key]['domain']
        sqm_per_employee = str(SubdomainAttributeDict[key]['sqmPerEmpl'])

        if key in subdomainsAccessibility.columns:
            livability_score = remap(subdomainsAccessibility[key], 0, threshold, 0, max_livability)
            livability.loc[subdomainsAccessibility[key] >= threshold, key] = max_livability
            livability.loc[subdomainsAccessibility[key] < threshold, key] = livability_score
            if any(domain):
                for item in domain:
                    livability.loc[subdomainsAccessibility[key] >= threshold, domain] += max_livability
                    livability.loc[subdomainsAccessibility[key] < threshold, domain] += livability_score
        
    return livability
    
    
    
    




        
    # Prepare the output
    output = {
        "subdomainsAccessibility_dictionary": subdomainsAccessibility_dictionary
    }


    
    return json.dumps(output)

    # Define the Gradio interface with a single JSON input
iface = gr.Interface(
    fn=test,
    inputs=gr.Textbox(label="Input JSON", lines=20, placeholder="Enter JSON with all parameters here..."),
    outputs=gr.JSON(label="Output JSON"),
    title="testspace"
)

iface.launch()