lets do this

Dockerfile

@@ -0,0 +1,16 @@
+# read the doc: https://huggingface.co/docs/hub/spaces-sdks-docker
+# you will also find guides on how best to write your Dockerfile
+
+FROM python:3.9
+
+WORKDIR /code
+
+COPY ./requirements.txt /code/requirements.txt
+
+RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt
+
+COPY . .
+
+ENV BOKEH_ALLOW_WS_ORIGIN=kkr5155-distributedswarmintelligence.hf.space
+
+CMD ["panel", "serve", "/code/app.py", "--address", "0.0.0.0","--port", "7860", "--allow-websocket-origin=kkr5155-distributedswarmintelligence.hf.space"]

app.py

@@ -0,0 +1,372 @@
+import random
+import numpy as np
+import threading
+import panel as pn
+pn.extension(template='bootstrap')
+import holoviews as hv
+import time
+import pandas as pd
+from holoviews.streams import Stream
+hv.extension('bokeh', logo=False)
+
+# Particle class: Each particle will be an object of this class with all the properties defined in __init__() method
+class Particle():
+
+    # Method to initialize particle properties
+    def __init__(self, initial):
+        self.position = []
+        self.velocity = []
+        self.initial = initial
+        self.best_position = []
+        self.best_error = float('inf')  # Initialize best_error with infinity
+        self.error = float('inf')  # Initialize error with infinity
+        self.num_dimensions = 2
+        for i in range(0, self.num_dimensions):
+            self.velocity.append(random.uniform(-1, 1))
+            self.position.append(initial[i])
+
+    # Method to update velocity of a particle object
+    def update_velocity(self, global_best_position, max_iter, iter_count):
+        c1_start = 2.5
+        c1_end = 0.5
+        c2_start = 0.5
+        c2_end = 2.5
+        w = 0.7298
+
+        c1 = c1_start - (c1_start - c1_end) * (iter_count / max_iter)
+        c2 = c2_start + (c2_end - c2_start) * (iter_count / max_iter)
+
+        for i in range(0, self.num_dimensions):
+            r1 = random.random()
+            r2 = random.random()
+
+            cog_vel = c1 * r1 * (self.best_position[i] - self.position[i])
+            social_vel = c2 * r2 * (global_best_position[i] - self.position[i])
+            self.velocity[i] = w * self.velocity[i] + cog_vel + social_vel
+
+    # Method to update position of a particle object
+    def update_position(self, bounds):
+        for i in range(0, self.num_dimensions):
+            self.position[i] = self.position[i] + self.velocity[i]
+
+            if self.position[i] > bounds[i][1]:
+                self.position[i] = bounds[i][1]
+
+            if self.position[i] < bounds[i][0]:
+                self.position[i] = bounds[i][0]
+
+    # Method to evaluate fitness of a particle
+    def evaluate_fitness(self, number, target, function):
+        if number == 1:
+            self.error = fitness_function(self.position, target)
+        else:
+            self.error = cost_function(self.position, function)
+
+        if self.error < self.best_error:
+            self.best_position = self.position[:]  # Create a copy of the position list
+            self.best_error = self.error
+
+    # Getter method to return the present error of a particle
+    def get_error(self):
+        return self.error
+
+    # Getter method to return the best position of a particle
+    def get_best_pos(self):
+        return self.best_position[:]  # Return a copy of the best position list
+
+    # Getter method to return the best error of a particle
+    def get_best_error(self):
+        return self.best_error
+
+    # Getter method to return the best position of a particle
+    def get_pos(self):
+        return self.position[:]  # Return a copy of the position list
+
+    # Getter method to return the velocity of a particle
+    def get_velocity(self):
+        return self.velocity[:]  # Return a copy of the velocity list
+
+# Function to calculate the euclidean distance from a particle to target
+def fitness_function(particle_position, target):
+    x_pos, y_pos = float(target[0]), float(target[1])
+    return (x_pos - particle_position[0])**2 + (y_pos - particle_position[1])**2
+
+# Function to calculate the value of the mathematical function at the position of a particle
+import sympy as sp
+
+def cost_function(particle_position, function_str):
+    x, y = sp.symbols('x y')
+    function = sp.sympify(function_str)
+    return function.subs({x: particle_position[0], y: particle_position[1]})
+
+# Interactive Class: to create a swarm of particles and an interactive PSO
+class Interactive_PSO():
+
+    # Method to initialize properties of an Interactive PSO
+    def __init__(self):
+        self._running = False
+        self.max_iter = 500  # Set the desired maximum number of iterations
+        self.num_particles = 25
+        self.initial = [5, 5]
+        self.bounds = [(-500, 500), (-500, 500)]
+        self.x_axis = []
+        self.y_axis = []
+        self.target = [5] * 2
+        self.global_best_error = float('inf')  # Initialize global_best_error with infinity
+        self.update_particles_position_lists_with_random_values()
+        self.global_best_position = [0, 0]
+
+    # Method to initialize swarm to find the target in a given search space
+    # Method to initialize swarm to find the target in a given search space
+    def swarm_initialization(self, number, max_iter):
+        swarm = []
+        self.global_best_position = [0, 0]
+        self.global_best_error = float('inf')  # Initialize global_best_error with infinity
+        self.gamma = 0.0001
+        function = function_select.value
+    
+        for i in range(0, self.num_particles):  # For loop to initialize the swarm of particles
+            swarm.append(Particle([self.x_axis[i], self.y_axis[i]]))
+    
+        iter_count = 0
+        while self._running:  # Loop to identify the best solution depending upon the problem
+            if self.global_best_error <= 0.00001:
+                break
+    
+            for j in range(0, self.num_particles):
+                swarm[j].evaluate_fitness(number, self.target, function)
+    
+                if swarm[j].get_error() < self.global_best_error:
+                    self.global_best_position = swarm[j].get_best_pos()
+                    self.global_best_error = swarm[j].get_best_error()
+    
+            for j in range(0, self.num_particles):
+                swarm[j].update_velocity(self.global_best_position, max_iter, iter_count)
+                swarm[j].update_position(self.bounds)
+                self.x_axis[j] = swarm[j].get_pos()[0]
+                self.y_axis[j] = swarm[j].get_pos()[1]
+    
+            # Add a delay to see the particle movement
+            time.sleep(0.05)  # Adjust the delay as needed
+    
+            iter_count += 1
+    
+            # Update the table with the current global best position
+            update_table = True  # <-- Set update_table to True
+            hv.streams.Stream.trigger(table_dmap.streams)
+    
+        self.initial = self.global_best_position
+        self._running = False
+        print('Best Position:', self.global_best_position)
+        print('Best Error:', self.global_best_error)
+        print('Function:', function)
+
+
+    # Method to terminate finding the solution of a problem
+    def terminate(self):
+        self._running = False
+
+    # Method to set _running parameter before initializing the swarm
+    def starting(self):
+        self._running = True
+
+    # Method to check if the swarm of particles are in action
+    def isrunning(self):
+        return self._running
+
+    # Getter method to return the number of particles
+    def get_num_particles(self):
+        return self.num_particles
+
+    # Setter method to update the number of particles
+    def update_num_particles(self, new_value):
+        self.num_particles = new_value
+
+    # Getter method to return the x_axis position list for particles in a swarm
+    def get_xaxis(self):
+        return self.x_axis[:]  # Return a copy
+
+    # Getter method to return the y_axis position list for particles in a swarm
+    def get_yaxis(self):
+        return self.y_axis[:]  # Return a copy
+
+    # Setter method to update the target position
+    def set_target(self, x, y):
+        self.target = [x, y]
+    
+    # Getter method to return the target position
+    def get_target(self):
+        return self.target[:]  # Return a copy
+    
+    # Method to update the length of particles position lists if there is a change in num of particles
+    def update_particles_position_lists(self, updated_num_particles):
+        old_x_value = self.x_axis[0]
+        old_y_value = self.y_axis[0]
+        if updated_num_particles > self.num_particles:
+            for i in range(self.num_particles, updated_num_particles):
+                self.x_axis.append(old_x_value)
+                self.y_axis.append(old_y_value)
+        else:
+            for i in range((self.num_particles) - 1, updated_num_particles - 1, -1):
+                self.x_axis.pop(i)
+                self.y_axis.pop(i)
+    
+    # Method to initialize the particles positions randomly
+    def update_particles_position_lists_with_random_values(self):
+        self.x_axis = random.sample(range(-500, 500), self.num_particles)
+        self.y_axis = random.sample(range(-500, 500), self.num_particles)
+
+pso_swarm = Interactive_PSO()  # Creating an interactive PSO to find the target
+pso_computation_swarm = Interactive_PSO()  # Creating an interactive PSO to find the optimal solution of a mathematical function
+
+update_table = False
+
+# Method to initialize swarm to find the target in a given search space
+def start_finding_the_target():
+    pso_swarm.swarm_initialization(1, pso_swarm.max_iter)
+
+# Method to initialize swarm to compute an optimal solution for a given problem
+def start_computation():
+    pso_computation_swarm.swarm_initialization(2, pso_computation_swarm.max_iter)
+
+# On event function for single tap to create and return the target with updated position
+def create_target_element(x, y):
+    pso_swarm.terminate()
+    if x is not None:
+        pso_swarm.set_target(x, y)
+    return hv.Points((x, y, 1), label='Target').opts(color='red', marker='^', size=10)
+
+# Function to stream the particles of pso_swarm to dynamic map in regular intervals
+def update():
+    x_axis = pso_swarm.get_xaxis()
+    y_axis = pso_swarm.get_yaxis()
+    data = (x_axis, y_axis, np.random.random(size=len(x_axis)))
+    pop_scatter = hv.Scatter(data, vdims=['y_axis', 'z'])
+    pop_scatter.opts(size=8, color='z', cmap='Coolwarm_r')
+    return pop_scatter
+
+# On event function for update button click to update the number of particles in both the swarms
+def computational_update():
+    x_axis = pso_computation_swarm.get_xaxis()
+    y_axis = pso_computation_swarm.get_yaxis()
+    data = (x_axis, y_axis, np.random.random(size=len(x_axis)))
+    pop_scatter1 = hv.Scatter(data, vdims=['y_axis', 'z'])
+    pop_scatter1.opts(size=8, color='z', cmap='Coolwarm_r')
+    return pop_scatter1
+
+# On event function for update button click to update the number of particles in both the swarms
+def update_num_particles_event(event):
+    if population_slider.value == pso_swarm.get_num_particles():
+        return
+    pso_swarm.terminate()
+    pso_computation_swarm.terminate()
+    time.sleep(1)
+    updated_num_particles = population_slider.value
+    pso_swarm.update_particles_position_lists(updated_num_particles)
+    pso_swarm.update_num_particles(updated_num_particles)
+    pso_computation_swarm.update_num_particles(updated_num_particles)
+    pso_computation_swarm.update_particles_position_lists_with_random_values()
+    pso_swarm.update_particles_position_lists_with_random_values()  # Update positions for pso_swarm as well
+    hv.streams.Stream.trigger(pso_scatter1.streams)
+    hv.streams.Stream.trigger(pso_scatter.streams)
+    
+# Periodic Callback function for every 3 seconds to stream the data to dynamic maps
+def trigger_streams():
+    global update_table
+    hv.streams.Stream.trigger(pso_scatter.streams)
+    hv.streams.Stream.trigger(pso_scatter1.streams)
+    if update_table:
+        update_table = False
+        hv.streams.Stream.trigger(table_dmap.streams)
+
+    # Update the target position
+    tap.event(x=pso_swarm.get_target()[0], y=pso_swarm.get_target()[1])
+
+    # Slow down the swarm's speed
+    time.sleep(0.05)  # Adjust the delay as needed
+
+# On event function for begin the hunting button click to start hunting for the target
+def hunting_button_event(event):
+    if not pso_swarm.isrunning():
+        pso_swarm.starting()
+        threading.Thread(target=start_finding_the_target).start()
+
+# On event function for start the computation button click to start computation for a mathematical function
+def computation_button_event(event):
+    if not pso_computation_swarm.isrunning():
+        pso_computation_swarm.starting()
+        threading.Thread(target=start_computation).start()
+
+def table():
+    position = pso_computation_swarm.global_best_position
+    df = pd.DataFrame({
+        'x_position': [round(position[0])],
+        'y_position': [round(position[1])]
+    })
+
+    # Create an hv.Table with the data
+    hv_table = hv.Table(df).opts(width=300, height=100)
+
+    return hv_table
+
+
+# Function to update the mathematical function for which swarm finds the optimal solution
+def update_function(event):
+    pso_computation_swarm.terminate()
+    time.sleep(1)
+    pso_computation_swarm.update_particles_position_lists_with_random_values()
+
+    
+# Two dynamic maps for two interactive PSOs, one for finding a target and one for computation of a mathematical function
+pso_scatter = hv.DynamicMap(update, streams=[Stream.define('Next')()]).opts(xlim=(-500, 500), ylim=(-500, 500),
+                                                                             title="Plot 2 : PSO for target finding ")
+pso_scatter1 = hv.DynamicMap(computational_update, streams=[Stream.define('Next')()]).opts(xlim=(-500, 500),
+                                                                                            ylim=(-500, 500),
+                                                                                            title="Plot 1 : PSO for a mathematical computation")
+
+# Dynamic map to update and display target
+tap = hv.streams.SingleTap(x=pso_swarm.get_target()[0], y=pso_swarm.get_target()[1])
+target_dmap = hv.DynamicMap(create_target_element, streams=[tap])
+
+# Define custom CSS styles for the table container
+custom_style = {
+    'background': '##4287f5',  # Background color
+    'border': '1px solid black',  # Border around the table
+    'padding': '8px',  # Padding inside the container
+    'box-shadow': '5px 5px 5px #bcbcbc'  # Box shadow for a 3D effect
+}
+
+# Dynamic map to update the table with continuous global best position of the swarm
+table_dmap = hv.DynamicMap(table,streams=[hv.streams.Stream.define('Next')()])
+table_label = pn.pane.Markdown("Once an optimal solution is found in plot 1 it is updated in the below table")
+
+# Button to order the swarm of particles to start finding the target
+start_hunting_button = pn.widgets.Button(name=' Click to find target for plot 2 ', width=50)
+start_hunting_button.on_click(hunting_button_event)
+
+# Button to order the swarm of particles to start computation for selected mathematical function
+start_finding_button = pn.widgets.Button(name=' Click to start computation for plot 1', width=50)
+start_finding_button.on_click(computation_button_event)
+
+# Button to update number of particles 
+update_num_particles_button = pn.widgets.Button(name='Update number of particles', width=50)
+update_num_particles_button.on_click(update_num_particles_event)
+
+# periodic callback for every three seconds to trigger streams method
+pn.state.add_periodic_callback(trigger_streams, 3)
+
+# Slider to change the number of particles
+population_slider = pn.widgets.IntSlider(name='Number of praticles', start=10, end=100, value=25)
+
+# Dropdown list to select a mathematical function
+function_select = pn.widgets.Select(name='Select', options=['x^2+(y-100)^2','(x-234)^2+(y+100)^2', 'x^3 + y^3 - 3*x*y', 'x^2 * y^2'])
+function_select.param.watch(update_function,'value')
+
+#combining the dynamic maps with particles and target into one dynamicmap
+plot_for_finding_the_target = pso_scatter*target_dmap    
+    
+# Building the layout and returning the dashboard     
+dashboard =  pn.Column(pn.Row(pn.Row(pso_scatter1.opts(width=500, height=500)), pn.Column(plot_for_finding_the_target.opts(width=500, height=500)),
+                             pn.Column(pn.Column(table_label, table_dmap, styles=custom_style), start_finding_button, start_hunting_button, update_num_particles_button, population_slider,function_select)))
+
+pn.panel(dashboard).servable(title='Swarm Particles Visualization')

requirements.txt

@@ -0,0 +1,10 @@
+panel
+bokeh
+numpy
+holoviews
+pandas
+colormap
+matplotlib
+webcolors
+thread6
+sympy