Added the update JSON heuristics

app.py

@@ -3,60 +3,136 @@ import streamlit as st
 import json
 import clipboard
 
-from main import genetic_algorithm, polish_errors, calculate_errors
+from main import genetic_algorithm, polish_errors, calculate_errors, update_genetic_algorithm, calculate_diff
 
 # Initialize session state
 if 'services' not in st.session_state:
     st.session_state.services = {}
 if 'users' not in st.session_state:
     st.session_state.users = {}
+if 'solution' not in st.session_state:
+    st.session_state.solution = {}
 
 # App title
-st.title('Services and Users JSON Builder')
+st.title('Services and Users Assignment Center')
 
 # Add sliders for population_size, num_generations, and mutation_rate
 st.subheader('Genetic Algorithm Parameters')
-population_size = st.slider('Population Size', min_value=500, max_value=5000, value=2500, step=100)
-num_generations = st.slider('Number of Generations', min_value=1000, max_value=10000, value=5000, step=500)
-mutation_rate = st.slider('Mutation Rate', min_value=0.0, max_value=1.0, value=0.01, step=0.01)
+population_size = st.slider('Population Size', min_value=500, max_value=5000, value=1500, step=100)
+num_generations = st.slider('Number of Generations', min_value=1000, max_value=10000, value=2500, step=250)
+mutation_rate = st.slider('Mutation Rate', min_value=0.0, max_value=1.0, value=0.01, step=0.05)
 
 # Button to run the genetic algorithm
-if st.button('Run Genetic Algorithm'):
+new_generation_run = st.button('Run new solution')
+update_generation_run = st.button('Update solution')
+
+if new_generation_run:
     # Call the genetic_algorithm function and get the best_solution
-    best_solution = genetic_algorithm(st.session_state.services, st.session_state.users, population_size,
-                                      num_generations, mutation_rate)
+    best_solution = genetic_algorithm(
+        services=st.session_state.services,
+        users=st.session_state.users,
+        population_size=population_size,
+        num_generations=num_generations,
+        mutation_rate=mutation_rate
+    )
+
+    # Save the state of the current best solution
+    st.session_state.solution = best_solution
 
     # Convert the best_solution to JSON
-    best_solution_json = json.dumps(best_solution, indent=4)
-    best_solution_errors = calculate_errors(best_solution, st.session_state.services, st.session_state.users)
+    best_solution_json = json.dumps(st.session_state.solution, indent=4)
+    best_solution_errors = calculate_errors(st.session_state.solution, st.session_state.services,
+                                            st.session_state.users)
     best_solution_errors = polish_errors(best_solution_errors)
-    best_solution_errors = json.dumps(best_solution_errors, indent=4)
+    best_solution_errors_json = json.dumps(best_solution_errors, indent=4)
 
     # Display the output JSON in a read-only form
-    st.subheader('Best Solution JSON')
-    st.text_area('Best Solution', value=best_solution_json, height=400, max_chars=None, key=None, disabled=True)
-    st.text_area('Unmet constraints', value=best_solution_errors, height=200, max_chars=None, key=None, disabled=True)
+    st.subheader('Best solution JSON')
+    st.text_area('Best solution',
+                 value=best_solution_json, height=400, max_chars=None, key=None, disabled=True)
+    st.text_area('Unmet constraints',
+                 value=best_solution_errors_json, height=200, max_chars=None, key=None, disabled=True)
 
     if st.button('Copy solution to Clipboard'):
         clipboard.copy(best_solution_json)
-        st.success('JSON copied to clipboard!')
     if st.button('Copy unmet constraints to Clipboard'):
-        clipboard.copy(best_solution_errors)
-        st.success('JSON copied to clipboard!')
+        clipboard.copy(best_solution_errors_json)
+
+if update_generation_run:
+    # Call the genetic_algorithm function and get the best_solution
+    best_updated_solution = update_genetic_algorithm(
+        prev_solution=st.session_state.solution,
+        updated_services=st.session_state.services,
+        updated_users=st.session_state.users,
+        population_size=population_size,
+        num_generations=num_generations,
+        mutation_rate=mutation_rate
+    )
+
+    change_report = calculate_diff(best_updated_solution, st.session_state.solution)
+    change_report_json = json.dumps(change_report, indent=4)
+
+    # Convert the best_solution to JSON
+    best_updated_solution_json = json.dumps(st.session_state.solution, indent=4)
+    best_updated_solution_errors = calculate_errors(
+        st.session_state.solution, st.session_state.services, st.session_state.users
+    )
+    best_updated_solution_errors = polish_errors(best_updated_solution_errors)
+    best_updated_solution_errors_json = json.dumps(best_updated_solution_errors, indent=4)
+
+    # Display the output JSON in a read-only form
+    st.subheader('Best updated solution JSON')
+    st.text_area('Best updated solution',
+                 value=best_updated_solution_json, height=600, max_chars=None, key=None, disabled=True)
+    st.text_area('Updated unmet constraints',
+                 value=best_updated_solution_errors_json, height=300, max_chars=None, key=None, disabled=True)
+    st.text_area('Change report',
+                 value=change_report_json, height=200, max_chars=None, key=None, disabled=True)
+
+    if st.button('Copy updated solution to Clipboard'):
+        clipboard.copy(best_updated_solution_json)
+    if st.button('Copy updated unmet constraints to Clipboard'):
+        clipboard.copy(best_updated_solution_errors_json)
+    if st.button('Save this updated solution over the last fully generated one'):
+        # Save the state of the current best solution
+        st.session_state.solution = best_updated_solution
 
 # Sidebar for uploading previously generated JSON
-with st.sidebar.expander('Upload previously generated JSON'):
-    uploaded_json = st.text_area('Paste your JSON here')
-    merge_json = st.button('Merge with JSON')
-    reset_json = st.button('Reset JSON')
+with st.sidebar.expander('Previously generated solution JSON'):
+    uploaded_solution_json = st.text_area('Paste your previously generated JSON here',
+                                          value=json.dumps(st.session_state.get('solution', ''), indent=4))
+    merge_json = st.button('Upload previously generated JSON')
+    reset_json = st.button('Reset previously generated JSON')
+
+    if reset_json:
+        st.session_state.solution = {}
+
+    if merge_json and uploaded_solution_json:
+        try:
+            st.session_state.solution = json.loads(uploaded_solution_json)
+            st.success('JSON loaded successfully')
+        except json.JSONDecodeError:
+            st.error('Invalid JSON format')
+
+# Sidebar for uploading previously user and service description JSON
+with st.sidebar.expander('Previously generated user and service description JSON'):
+    previously_generated_user_service_json = {
+        'services': st.session_state.services,
+        'users': st.session_state.users
+    }
+    uploaded_user_service_json = st.text_area('Paste your user and service JSON here',
+                                              value=json.dumps(previously_generated_user_service_json, indent=4)
+                                              )
+    merge_json = st.button('Merge user and service description JSON')
+    reset_json = st.button('Reset user and service description JSON')
 
     if reset_json:
         st.session_state.services = {}
         st.session_state.users = {}
 
-    if merge_json and uploaded_json:
+    if merge_json and uploaded_user_service_json:
         try:
-            loaded_data = json.loads(uploaded_json)
+            loaded_data = json.loads(uploaded_user_service_json)
             st.session_state.services.update(loaded_data.get('services', {}))
             st.session_state.users.update(loaded_data.get('users', {}))
             st.success('JSON loaded successfully')
@@ -69,20 +145,26 @@ with st.sidebar.expander('Update existing user or service'):
 
     if object_type == 'Service':
         service_key = st.selectbox('Select a service', list(st.session_state.services.keys()), key='update_service_key')
-        if service_key and st.button('Load Service'):
-            st.session_state.service_name = service_key
-            st.session_state.min_val = st.session_state.services[service_key]['min']
-            st.session_state.rec_val = st.session_state.services[service_key]['rec']
-            st.session_state.max_val = st.session_state.services[service_key]['max']
-            st.session_state.priority = st.session_state.services[service_key]['priority']
+        if service_key:
+            if st.button('Load Service'):
+                st.session_state.service_name = service_key
+                st.session_state.min_val = st.session_state.services[service_key]['min']
+                st.session_state.rec_val = st.session_state.services[service_key]['rec']
+                st.session_state.max_val = st.session_state.services[service_key]['max']
+                st.session_state.priority = st.session_state.services[service_key]['priority']
+            if st.button('Drop Service'):
+                del st.session_state.services[service_key]
 
     elif object_type == 'User':
         user_key = st.selectbox('Select a user', list(st.session_state.users.keys()), key='update_user_key')
-        if user_key and st.button('Load User'):
-            st.session_state.user_name = user_key
-            st.session_state.max_assignments = st.session_state.users[user_key]['max_assignments']
-            st.session_state.preferences = st.session_state.users[user_key]['preferences']
-            st.session_state.cannot_assign = st.session_state.users[user_key]['cannot_assign']
+        if user_key:
+            if st.button('Load User'):
+                st.session_state.user_name = user_key
+                st.session_state.max_assignments = st.session_state.users[user_key]['max_assignments']
+                st.session_state.preferences = st.session_state.users[user_key]['preferences']
+                st.session_state.cannot_assign = st.session_state.users[user_key]['cannot_assign']
+            if st.button('Drop User'):
+                del st.session_state.users[user_key]
 
 # Add a service form
 with st.form(key='service_form'):
@@ -133,7 +215,7 @@ combined_data = {
 json_data = json.dumps(combined_data, indent=4)
 
 # Display the generated JSON
-st.subheader('Generated JSON')
+st.subheader('Generated user and services JSON')
 st.code(json_data, language='json')
 
 # Button to copy JSON to clipboard

main.py

@@ -1,6 +1,7 @@
 import copy
 import random
-from typing import Callable, Optional, Tuple
+from functools import partial
+from typing import Callable, Optional, Tuple, List
 
 
 def initialize_population(services: dict, users: dict, population_size: int) -> list:
@@ -83,35 +84,101 @@ def default_fitness_function(assignment_solution: dict, services: dict, users: d
     Returns:
         float: The fitness score of the given assignment solution.
     """
-    fitness = 0
+    fitness = -100
 
     for service, assigned_users in assignment_solution.items():
         service_info = services[service]
         num_assigned_users = len(assigned_users)
 
-        # Bonus for solutions that assign users near the recommended value
+        # Prefers for solutions that assign users near the recommended value
+        # (positive or negative value is punishment, score 0 for the best fit)
         if service_info["min"] <= num_assigned_users <= service_info["max"]:
-            fitness += abs(num_assigned_users - service_info["rec"])
+            fitness += abs(num_assigned_users - service_info["rec"]) * service_info["priority"]
 
         # Punish solutions that assign users below the minimum value
+        # (positive value is punishment)
         elif num_assigned_users < service_info["min"]:
-            fitness -= (service_info["min"] - num_assigned_users) * service_info["priority"]
+            fitness += (service_info["min"] - num_assigned_users) * service_info["priority"]
 
         # Punish solutions that assign users above the maximum value
+        # (positive value is punishment)
         else:  # num_assigned_users > service_info["max"]:
-            fitness -= (num_assigned_users - service_info["max"]) * service_info["priority"]
+            fitness += (num_assigned_users - service_info["max"]) * service_info["priority"]
 
         # Punish solutions that assign users to their cannot_assign services
+        # (positive value is punishment)
         for user in assigned_users:
             if service in users[user]["cannot_assign"]:
-                fitness -= 100 * service_info["priority"]
+                fitness += 100 * service_info["priority"]
 
         # Bonus solutions that assign users to their preferred services
+        # (negative value is bonus)
         for user, user_info in users.items():
             if service in user_info["preferences"] and user in assigned_users:
-                fitness += 10
+                fitness -= 20
 
-    return -fitness
+    return fitness
+
+
+def least_changed_fitness_function(prev_solution: dict, solution: dict, services: dict, users: dict) -> float:
+    """
+    A fitness function that favors solutions with the least changes from a previous solution.
+
+    Args:
+        solution (dict): The assignment solution to evaluate.
+        services (dict): The input services dictionary.
+        users (dict): The input users dictionary.
+        prev_solution (dict): The previous assignment solution to compare against.
+
+    Returns:
+        float: A fitness score for the assignment solution.
+    """
+    # Add the default fitness function score
+    fitness = default_fitness_function(solution, services, users)
+
+    # Bonus for users assigned to the same services as in the previous solution
+    same_user_assignments = 0
+    for service, assigned_users in solution.items():
+        if service in prev_solution:
+            prev_assigned_users = prev_solution[service]
+            same_users = set(assigned_users).intersection(set(prev_assigned_users))
+            same_user_assignments += len(same_users)
+
+    user_bonus = same_user_assignments * 100  # / sum(len(user_data["preferences"]) for user_data in users.values())
+    # (positive value is punishment)
+    fitness -= user_bonus
+
+    # Bonus for services having the same users assigned as in the previous solution
+    same_service_assignments = 0
+    for service, assigned_users in solution.items():
+        if service in prev_solution:
+            prev_assigned_users = prev_solution[service]
+            same_users = set(assigned_users).intersection(set(prev_assigned_users))
+            same_service_assignments += len(same_users)
+
+    service_bonus = same_service_assignments * 100
+    # (positive value is punishment)
+    fitness -= service_bonus
+
+    # Malus for user and service changes
+    user_changes = 0
+    service_changes = 0
+
+    for service, service_data in services.items():
+        if service in prev_solution:
+            prev_service_data = prev_solution[service]
+            if service_data != prev_service_data:
+                service_changes += 1
+            for user, user_data in users.items():
+                if user in prev_service_data and user not in service_data:
+                    user_changes += 1
+
+    user_malus = user_changes
+    service_malus = service_changes
+    # (positive value is punishment)
+    fitness += (user_malus + service_malus)
+
+    return fitness
 
 
 def selection(fitness_scores: list) -> Tuple[int, int]:
@@ -332,8 +399,51 @@ def polish_errors(errors: dict) -> dict:
     return polished_errors
 
 
+def calculate_diff(solution1: dict, solution2: dict) -> dict:
+    """
+    Calculate the differences between two solution JSON objects and return the differences categorized into
+    "added" and "removed" attributes for each service.
+
+    Args:
+        solution1 (dict): The first solution JSON object.
+        solution2 (dict): The second solution JSON object.
+
+    Returns:
+        dict: A dictionary with the differences between the two solutions, categorized into "added" and
+              "removed" attributes for each service.
+    """
+    diff = {}
+
+    all_services = set(solution1.keys()).union(set(solution2.keys()))
+
+    for service in all_services:
+        service_diff = {
+            "added": [],
+            "removed": []
+        }
+
+        if service not in solution1:
+            service_diff["added"] = solution2[service]
+        elif service not in solution2:
+            service_diff["removed"] = solution1[service]
+        else:
+            added_users = set(solution2[service]) - set(solution1[service])
+            removed_users = set(solution1[service]) - set(solution2[service])
+
+            if added_users:
+                service_diff["added"] = list(added_users)
+            if removed_users:
+                service_diff["removed"] = list(removed_users)
+
+        if service_diff["added"] or service_diff["removed"]:
+            diff[service] = service_diff
+
+    return diff
+
+
 def genetic_algorithm(services: dict, users: dict, population_size: int = 100, num_generations: int = 100,
-                      mutation_rate: float = 0.01, fitness_fn: Optional[Callable] = None) -> dict:
+                      mutation_rate: float = 0.01, fitness_fn: Optional[Callable] = None,
+                      initial_population: Optional[List[dict]] = None) -> dict:
     """
     Run the genetic algorithm to find an optimal assignment solution based on user preferences and constraints.
 
@@ -344,12 +454,13 @@ def genetic_algorithm(services: dict, users: dict, population_size: int = 100, n
         num_generations (int): The number of generations for the genetic algorithm to run (default: 100).
         mutation_rate (float): The probability of mutation for each individual in the population (default: 0.01).
         fitness_fn (Callable, optional): An optional custom fitness function.
+        initial_population: An optional previous solution to be used as starting point
 
     Returns:
         dict: The best assignment solution found by the genetic algorithm.
     """
     # Initialize the population
-    population = initialize_population(services, users, population_size)
+    population = initial_population or initialize_population(services, users, population_size)
 
     # If no custom fitness function is provided, use the default fitness function
     if fitness_fn is None:
@@ -389,3 +500,38 @@ def genetic_algorithm(services: dict, users: dict, population_size: int = 100, n
         report_generation(generation, fitness_scores, best_solution, services, users)
 
     return best_solution
+
+
+def update_genetic_algorithm(prev_solution: dict, updated_services: dict, updated_users: dict,
+                             population_size: int = 100, num_generations: int = 100, mutation_rate: float = 0.01,
+                             fitness_fn: Optional[Callable] = None) -> dict:
+    """
+    Update the previous assignment solution with updated services and users using a genetic algorithm.
+
+    Args:
+        prev_solution (dict): The previous assignment solution.
+        updated_services (dict): The updated services dictionary.
+        updated_users (dict): The updated users dictionary.
+        population_size (int): The size of the population for each generation in the genetic algorithm (default: 100).
+        num_generations (int): The number of generations for the genetic algorithm to run (default: 100).
+        mutation_rate (float): The probability of mutation for each individual in the population (default: 0.01).
+        fitness_fn (Optional[Callable]): An optional fitness function to use in the genetic algorithm.
+
+    Returns:
+        dict: The updated assignment solution.
+    """
+
+    # Create the initial population with the given previous solution
+    population = initialize_population(updated_services, updated_users, population_size - 1)
+    population.append(prev_solution)
+
+    if fitness_fn is None:
+        fitness_fn = partial(least_changed_fitness_function, prev_solution)
+
+    # Run the genetic algorithm using the initial population
+    updated_solution = genetic_algorithm(updated_services, updated_users, population_size, num_generations,
+                                         mutation_rate,
+                                         fitness_fn,
+                                         population)
+
+    return updated_solution