From e8e1e7b7376fc1e4c3994893e1e3ca2da8f0cc28 Mon Sep 17 00:00:00 2001
From: shinedday <shinedday@gmail.com>
Date: Wed, 11 May 2022 10:03:18 +0200
Subject: [PATCH] Modify : how client are initialize in order to better match
future ihm interface
---
README.md | 3 +++
agent.py | 10 ++-------
amas.py | 17 ++++++++++----
environment.py | 9 ++------
ihm.py | 30 ++++++++++++-------------
philosophers/agent.py | 6 ++---
philosophers/amas.py | 6 ++---
philosophers/env.py | 6 ++---
scheduler.py | 14 ++++++------
tool/config.json | 10 ++++-----
tool/mqtt_client.py | 11 ++-------
tool/remote_client.py | 3 +++
tool/schedulable.py | 6 ++---
tool/ssh_client.py | 48 ++++++++++++++++++++++++++++++++++++---
tool/update.py | 52 -------------------------------------------
15 files changed, 109 insertions(+), 122 deletions(-)
delete mode 100644 tool/update.py
diff --git a/README.md b/README.md
index 5f8af0f..4ad75c4 100644
--- a/README.md
+++ b/README.md
@@ -40,6 +40,9 @@ WSL (pour pouvoir démarer l'amas)
# Random note
+First procedure :
+ 1. etre sur que le ssh est trust (faire un premier ssh hors du programe)
+
Start procédure :
1. start broker
diff --git a/agent.py b/agent.py
index dc77b1b..958e026 100644
--- a/agent.py
+++ b/agent.py
@@ -1,7 +1,6 @@
"""
Agent class file
"""
-import sys
from ast import literal_eval
from typing import Dict
@@ -13,10 +12,10 @@ class Agent(Schedulable):
base class for agent
"""
- def __init__(self, identifier: int) -> None:
+ def __init__(self, identifier: int, broker_ip: str) -> None:
self.id = identifier
- Schedulable.__init__(self, client_id="Agent" + str(self.id))
+ Schedulable.__init__(self, broker_ip, "Agent" + str(self.id))
self.subscribe("scheduler/agent/wakeup", self.wake_up)
@@ -122,8 +121,3 @@ class Agent(Schedulable):
self.publish("metric", str(self.send_metric()))
self.publish("cycle_done", "")
-
-
-if __name__ == '__main__':
- a = Agent(int(sys.argv[1]))
- a.run()
diff --git a/amas.py b/amas.py
index 2708131..e8e97d3 100644
--- a/amas.py
+++ b/amas.py
@@ -4,6 +4,7 @@ Amas class
from ast import literal_eval
from typing import List
+from tool.remote_client import RemoteClient
from tool.schedulable import Schedulable
from tool.ssh_client import SSHClient, Cmd
@@ -13,10 +14,16 @@ class Amas(Schedulable, SSHClient):
Amas class
"""
- def __init__(self) -> None:
- Schedulable.__init__(self, client_id="Amas")
- SSHClient.__init__(self)
+ def __init__(self, broker_ip: str, clients: str) -> None:
+ Schedulable.__init__(self, broker_ip, "Amas")
+ print(clients)
+ true_client = [RemoteClient(i.get("hostname"), i.get("user"), i.get("password")) for i in literal_eval(clients)]
+ print(true_client)
+
+ SSHClient.__init__(self, true_client)
+
+ self.broker_ip = broker_ip
self.subscribe("scheduler/schedulable/wakeup", self.wake_up)
self.next_id = 0
@@ -46,7 +53,8 @@ class Amas(Schedulable, SSHClient):
args = []
command = "nohup python "
command += "\"Desktop/mqtt_goyon/iotamak-core/" + experience_name + "/agent.py\" "
- command += str(self.next_id) + " "
+ command += str(self.next_id) + " \""
+ command += str(self.broker_ip) + "\" "
for arg in args:
command += str(arg) + " "
command += "&"
@@ -70,6 +78,7 @@ class Amas(Schedulable, SSHClient):
self.agents_cmd) // total_pi
])
)
+ print("Amas, push agent done")
def agent_log(self, client, userdata, message) -> None:
"""
diff --git a/environment.py b/environment.py
index 9eacea2..24ea96e 100644
--- a/environment.py
+++ b/environment.py
@@ -10,8 +10,8 @@ class Environment(Schedulable):
Environment class
"""
- def __init__(self) -> None:
- Schedulable.__init__(self, client_id="Env")
+ def __init__(self, broker_ip: str) -> None:
+ Schedulable.__init__(self, broker_ip, "Env")
self.subscribe("scheduler/schedulable/wakeup", self.wake_up)
@@ -56,8 +56,3 @@ class Environment(Schedulable):
self.client.publish("env/action_done", "")
self.nbr_cycle += 1
-
-
-if __name__ == '__main__':
- s = Environment()
- s.run()
diff --git a/ihm.py b/ihm.py
index ba0a1c7..6e520c4 100644
--- a/ihm.py
+++ b/ihm.py
@@ -6,17 +6,17 @@ from os import path
from subprocess import Popen
from time import sleep
-from tool.confi_reader import read_ssh
+from tool.confi_reader import read_ssh, read_broker
from tool.mqtt_client import MqttClient
from tool.ssh_client import SSHClient, Cmd
-from tool.update import VersionManager
class Ihm(MqttClient, SSHClient):
def __init__(self):
- MqttClient.__init__(self, "Ihm")
- SSHClient.__init__(self)
+ self.broker_ip = read_broker(str(pathlib.Path(__file__).parent.resolve()) + "/tool/config.json")
+ MqttClient.__init__(self, self.broker_ip, "Ihm")
+ SSHClient.__init__(self, read_ssh(str(pathlib.Path(__file__).parent.resolve()) + "/tool/config.json"))
def run(self):
@@ -30,7 +30,7 @@ class Ihm(MqttClient, SSHClient):
# Envoie un signal a tout le monde pour exit
if cmd.lower() == "exit":
- self.client.publish("ihm/exit")
+ self.client.publish("ping/exit")
exit_bool = True
# cherche tout les potentiel agents sur les raspberry et les tuent
@@ -77,8 +77,7 @@ class Ihm(MqttClient, SSHClient):
print("Hostname :", self.clients[i_client].hostname, " User :", self.clients[i_client].user)
self.run_cmd(i_client, commands)
- updater = VersionManager()
- updater.update()
+ self.update()
# charge une experience et verifie le format
if cmd.lower() == "load":
@@ -107,21 +106,22 @@ class Ihm(MqttClient, SSHClient):
cmd = input(">")
# start subprocess scheduler
- p1 = Popen([sys.executable, 'scheduler.py', cmd])
+ p1 = Popen([sys.executable, 'scheduler.py', cmd, self.broker_ip])
sleep(1)
# start subprocess amas
- p2 = Popen([sys.executable, experiment_folder + '/amas.py'])
+ send_client = [ c.to_send() for c in self.clients]
+ p2 = Popen([sys.executable, experiment_folder + '/amas.py', self.broker_ip, str(send_client)])
# start subprocess env
- p3 = Popen([sys.executable, experiment_folder + '/env.py'])
+ p3 = Popen([sys.executable, experiment_folder + '/env.py', self.broker_ip])
if cmd.lower() == "pause":
- self.client.publish("ihm/pause")
+ self.client.publish("ping/pause")
if cmd.lower() == "unpause":
- self.client.publish("ihm/unpause")
+ self.client.publish("ping/unpause")
if cmd.lower() in ["s", "step"]:
- self.client.publish("ihm/step")
+ self.client.publish("ping/step")
if cmd.lower() == "ping":
@@ -134,9 +134,9 @@ class Ihm(MqttClient, SSHClient):
print("Hostname :", client.hostname, " Responded : ", result)
if cmd.lower() == "mode":
- self.client.publish("ihm/mode")
+ self.client.publish("ping/mode")
- self.client.publish("ihm/step")
+ self.client.publish("ping/step")
sleep(2)
diff --git a/philosophers/agent.py b/philosophers/agent.py
index 6de0ab2..6826aed 100644
--- a/philosophers/agent.py
+++ b/philosophers/agent.py
@@ -13,9 +13,9 @@ from random import randrange
class PhiAgent(Agent):
- def __init__(self, identifier: int, total_agent):
+ def __init__(self, identifier: int, broker_ip: str, total_agent):
self.total_agent = total_agent
- super().__init__(identifier)
+ Agent.__init__(self, identifier, broker_ip)
def on_initialization(self):
# 0: thinking, 1: hungry, 2: eat
@@ -96,5 +96,5 @@ class PhiAgent(Agent):
if __name__ == '__main__':
- a = PhiAgent(int(sys.argv[1]), int(sys.argv[2]))
+ a = PhiAgent(int(sys.argv[1]), str(sys.argv[2]), int(sys.argv[3]))
a.run()
diff --git a/philosophers/amas.py b/philosophers/amas.py
index 886fd89..af9eeb2 100644
--- a/philosophers/amas.py
+++ b/philosophers/amas.py
@@ -8,9 +8,9 @@ from amas import Amas
class PhiAmas(Amas):
- def __init__(self, nbr_agent):
+ def __init__(self, broker_ip: str, clients, nbr_agent):
self.agent_to_create = nbr_agent
- super().__init__()
+ super().__init__(broker_ip, clients)
def on_initial_agents_creation(self):
for _ in range(self.agent_to_create):
@@ -18,5 +18,5 @@ class PhiAmas(Amas):
if __name__ == '__main__':
- s = PhiAmas(3)
+ s = PhiAmas(str(sys.argv[1]), sys.argv[2], 5)
s.run()
diff --git a/philosophers/env.py b/philosophers/env.py
index 34e7c8f..8ababf9 100644
--- a/philosophers/env.py
+++ b/philosophers/env.py
@@ -10,9 +10,9 @@ from philosophers.fork import Fork
class PhiEnv(Environment):
- def __init__(self, nbr_phil):
+ def __init__(self, broker_ip, nbr_phil):
self.nbr_phil = nbr_phil
- super().__init__()
+ super().__init__(broker_ip)
def on_initialization(self):
self.forks = []
@@ -65,5 +65,5 @@ class PhiEnv(Environment):
if __name__ == '__main__':
- s = PhiEnv(3)
+ s = PhiEnv(str(sys.argv[1]), 5)
s.run()
\ No newline at end of file
diff --git a/scheduler.py b/scheduler.py
index 2e00a58..e42d8cd 100644
--- a/scheduler.py
+++ b/scheduler.py
@@ -12,9 +12,9 @@ class Scheduler(Schedulable):
Scheduler class, it's role is to make sure the amas, the env and all the agents stay in sync
"""
- def __init__(self, execution_policy: int) -> None:
+ def __init__(self, execution_policy: int, broker_ip: str) -> None:
- Schedulable.__init__(self, client_id="Scheduler")
+ Schedulable.__init__(self, broker_ip, "Scheduler")
self.sleep_between_cycle = 0
# 0: pas a pas, 1: auto
@@ -26,10 +26,10 @@ class Scheduler(Schedulable):
self.subscribe("amas/agent/new", self.update_nbr_agent)
self.subscribe("amas/action_done", self.update_schedulable)
self.subscribe("env/action_done", self.update_schedulable)
- self.subscribe("ihm/step", self.step)
- self.subscribe("ihm/pause", self.pause)
- self.subscribe("ihm/unpause", self.unpause)
- self.subscribe("ihm/mode", self.mode)
+ self.subscribe("ping/step", self.step)
+ self.subscribe("ping/pause", self.pause)
+ self.subscribe("ping/unpause", self.unpause)
+ self.subscribe("ping/mode", self.mode)
self.agent_waiting = 0
self.schedulable_waiting = 0
@@ -172,5 +172,5 @@ class Scheduler(Schedulable):
if __name__ == '__main__':
- s = Scheduler(int(sys.argv[1]))
+ s = Scheduler(int(sys.argv[1]), str(sys.argv[2]))
s.run()
diff --git a/tool/config.json b/tool/config.json
index a8df663..84fb1a9 100644
--- a/tool/config.json
+++ b/tool/config.json
@@ -1,23 +1,23 @@
{
- "broker" : "192.168.24.209",
+ "broker" : "192.168.153.209",
"clients_ssh" : [
{
- "hostname" : "192.168.24.18",
+ "hostname" : "192.168.153.18",
"user" : "pi",
"password" : "raspberry"
},
{
- "hostname" : "192.168.24.61",
+ "hostname" : "192.168.153.227",
"user" : "pi",
"password" : "raspberry"
},
{
- "hostname" : "192.168.24.227",
+ "hostname" : "192.168.153.61",
"user" : "pi",
"password" : "raspberry"
},
{
- "hostname" : "192.168.24.75",
+ "hostname" : "192.168.153.75",
"user" : "pi",
"password" : "raspberry"
}
diff --git a/tool/mqtt_client.py b/tool/mqtt_client.py
index b5901b9..1311df1 100644
--- a/tool/mqtt_client.py
+++ b/tool/mqtt_client.py
@@ -1,19 +1,12 @@
-import pathlib
-import sys
-
-sys.path.insert(0, str(pathlib.Path(__file__).parent.parent))
-
from paho.mqtt.client import Client
-from tool.confi_reader import read_broker
-
class MqttClient:
- def __init__(self, client_id: str = None):
+ def __init__(self, broker_ip, client_id):
self.client = Client(client_id=client_id)
self.client.username_pw_set(username="goyon", password="mosquitto")
- self.client.connect(read_broker(str(pathlib.Path(__file__).parent.resolve()) + "/config.json"))
+ self.client.connect(host=broker_ip)
self.client.loop_start()
def subscribe(self, topic, fun):
diff --git a/tool/remote_client.py b/tool/remote_client.py
index 2627909..f257a53 100644
--- a/tool/remote_client.py
+++ b/tool/remote_client.py
@@ -4,3 +4,6 @@ class RemoteClient:
self.hostname = hostname
self.user = user
self.password = password
+
+ def to_send(self) -> dict:
+ return {"hostname": self.hostname, "user": self.user, "password": self.password}
diff --git a/tool/schedulable.py b/tool/schedulable.py
index dac1734..2b6f5ae 100644
--- a/tool/schedulable.py
+++ b/tool/schedulable.py
@@ -15,11 +15,11 @@ class Schedulable(MqttClient):
Base class for Agent/Amas/Env/scheduler
"""
- def __init__(self, client_id: str):
- MqttClient.__init__(self, client_id)
+ def __init__(self, broker_ip: str, client_id: str):
+ MqttClient.__init__(self, broker_ip, client_id)
self.exit_bool: bool = False
- self.subscribe("ihm/exit", self.exit_procedure)
+ self.subscribe("ping/exit", self.exit_procedure)
self.nbr_cycle: int = 0
self.wait_delay: float = 0.01
diff --git a/tool/ssh_client.py b/tool/ssh_client.py
index 1fed649..ea20d4e 100644
--- a/tool/ssh_client.py
+++ b/tool/ssh_client.py
@@ -1,10 +1,14 @@
+import os
import pathlib
import sys
+from typing import List
+import paramiko
from pexpect import pxssh
+
sys.path.insert(0, str(pathlib.Path(__file__).parent.parent))
-from tool.confi_reader import read_ssh
+from tool.remote_client import RemoteClient
class Cmd:
@@ -17,13 +21,14 @@ class Cmd:
class SSHClient:
- def __init__(self):
- self.clients = read_ssh(str(pathlib.Path(__file__).parent.resolve()) + "/config.json")
+ def __init__(self, clients: List[RemoteClient]):
+ self.clients = clients
def run_cmd(self, client: int, cmd: list):
try:
s = pxssh.pxssh()
dest = self.clients[client]
+ print(dest.hostname, dest.user, dest.password)
s.login(dest.hostname, dest.user, dest.password)
for command in cmd:
@@ -36,3 +41,40 @@ class SSHClient:
except pxssh.ExceptionPxssh as e:
print("pxssh failed on login.")
print(e)
+
+ def update(self):
+ for client in self.clients:
+ transport = paramiko.Transport((client.hostname, 22))
+ transport.connect(username=client.user, password=client.password)
+ sftp = MySFTPClient.from_transport(transport)
+ sftp.mkdir("./Desktop/mqtt_goyon/iotamak-core", ignore_existing=True)
+ sftp.put_dir("/mnt/d/work/stage m1/iotamak-core", "./Desktop/mqtt_goyon/iotamak-core")
+ sftp.close()
+
+class MySFTPClient(paramiko.SFTPClient):
+ def put_dir(self, source, target):
+ ''' Uploads the contents of the source directory to the target path. The
+ target directory needs to exists. All subdirectories in source are
+ created under target.
+ '''
+ for item in os.listdir(source):
+ if os.path.isfile(os.path.join(source, item)):
+ self.put(os.path.join(source, item), '%s/%s' % (target, item))
+ print(os.path.join(source, item))
+ else:
+ if any([i in item for i in [".git", "__pycache__", ".idea", ".vscode"]]):
+ pass
+ else:
+ self.mkdir('%s/%s' % (target, item), ignore_existing=True)
+ self.put_dir(os.path.join(source, item), '%s/%s' % (target, item))
+ print(os.path.join(source, item))
+
+ def mkdir(self, path, mode=511, ignore_existing=False):
+ ''' Augments mkdir by adding an option to not fail if the folder exists '''
+ try:
+ super(MySFTPClient, self).mkdir(path, mode)
+ except IOError:
+ if ignore_existing:
+ pass
+ else:
+ raise
diff --git a/tool/update.py b/tool/update.py
deleted file mode 100644
index f261e89..0000000
--- a/tool/update.py
+++ /dev/null
@@ -1,52 +0,0 @@
-import os
-import pathlib
-import sys
-
-import paramiko
-
-sys.path.insert(0, str(pathlib.Path(__file__).parent.parent))
-from tool.confi_reader import read_ssh
-
-
-class VersionManager:
-
- def __init__(self):
- self.clients = read_ssh(str(pathlib.Path(__file__).parent.resolve()) + "/config.json")
-
- def update(self):
- for client in self.clients:
- transport = paramiko.Transport((client.hostname, 22))
- transport.connect(username=client.user, password=client.password)
- sftp = MySFTPClient.from_transport(transport)
- sftp.mkdir("./Desktop/mqtt_goyon/iotamak-core", ignore_existing=True)
- sftp.put_dir("/mnt/d/work/stage m1/iotamak-core", "./Desktop/mqtt_goyon/iotamak-core")
- sftp.close()
-
-
-class MySFTPClient(paramiko.SFTPClient):
- def put_dir(self, source, target):
- ''' Uploads the contents of the source directory to the target path. The
- target directory needs to exists. All subdirectories in source are
- created under target.
- '''
- for item in os.listdir(source):
- if os.path.isfile(os.path.join(source, item)):
- self.put(os.path.join(source, item), '%s/%s' % (target, item))
- print(os.path.join(source, item))
- else:
- if any([i in item for i in [".git", "__pycache__", ".idea", ".vscode"]]):
- pass
- else:
- self.mkdir('%s/%s' % (target, item), ignore_existing=True)
- self.put_dir(os.path.join(source, item), '%s/%s' % (target, item))
- print(os.path.join(source, item))
-
- def mkdir(self, path, mode=511, ignore_existing=False):
- ''' Augments mkdir by adding an option to not fail if the folder exists '''
- try:
- super(MySFTPClient, self).mkdir(path, mode)
- except IOError:
- if ignore_existing:
- pass
- else:
- raise
--
GitLab