Oui merci ^^
En effet car il s'agit d'une supervision sous Vijeo Citect
. Code : Tout sélectionner
import asyncio
from pymodbus.server.async_io import ModbusTcpServer
from pymodbus.datastore import ModbusSequentialDataBlock, ModbusSlaveContext, ModbusServerContext
from pymodbus.device import ModbusDeviceIdentification
import logging
from datetime import datetime
import re
# Récupérer l'heure actuelle
current_time = datetime.now()
# Extraire l'année, le mois, le jour, l'heure et les minutes dans des variables
year = current_time.year
month = current_time.month
day = current_time.day
hour = current_time.hour
minute = current_time.minute
week_number = current_time.isocalendar()[1] # Numéro de la semaine
# Afficher les variables
print(f"Year: {year}")
print(f"Month: {month}")
print(f"Day: {day}")
print(f"Hour: {hour}")
print(f"Minute: {minute}")
print(f"Week number: {week_number}")
# Configuration du journal
logging.basicConfig(level=logging.INFO)
log = logging.getLogger()
# Vérification de l'adresse IP
def is_valid_ip(ip):
pattern = r"^(?:[0-9]{1,3}\.){3}[0-9]{1,3}$"
if re.match(pattern, ip):
parts = ip.split(".")
return all(0 <= int(part) <= 255 for part in parts)
return False
# Vérification du port
def is_valid_port(port):
return 1 <= port <= 65535
# Vérification de l'ID
def is_valid_id(slave_id):
return 1 <= slave_id <= 247 # Limites selon la norme Modbus
# Demande de l'adresse IP
def get_ip_address():
ip_address = input("Entrez l'adresse IP du serveur (ex: 127.0.0.1) : ")
if not ip_address: # Si l'utilisateur n'entre rien
ip_address = "127.0.0.1" # Valeur par défaut
print(f"Adresse IP non fournie, utilisation de la valeur par défaut : {ip_address}")
elif not is_valid_ip(ip_address) or ip_address == "localhost":
print("Adresse IP invalide. Utilisation de la valeur par défaut : 127.0.0.1")
ip_address = "127.0.0.1"
return ip_address
# Demande du port
def get_port():
port_input = input("Entrez le numéro de port du serveur (entre 1 et 65535, par défaut : 502) : ")
if not port_input: # Si l'utilisateur n'entre rien
port = 502 # Valeur par défaut
print(f"Port non fourni, utilisation de la valeur par défaut : {port}")
else:
try:
port = int(port_input)
if not is_valid_port(port):
print("Numéro de port invalide. Utilisation du port par défaut : 502")
port = 502
except ValueError:
print("Entrée invalide. Utilisation du port par défaut : 502")
port = 502
return port
# Demande de l'ID d'esclave
def get_slave_id():
slave_id_input = input("Entrez le numéro d'ID de l'esclave (entre 1 et 247, par défaut : 1) : ")
if not slave_id_input: # Si l'utilisateur n'entre rien
slave_id = 1 # Valeur par défaut
print(f"ID non fourni, utilisation de la valeur par défaut : {slave_id}")
else:
try:
slave_id = int(slave_id_input)
if not is_valid_id(slave_id):
print("ID invalide. Utilisation de l'ID par défaut : 1")
slave_id = 1
except ValueError:
print("Entrée invalide. Utilisation de l'ID par défaut : 1")
slave_id = 1
return slave_id
# Configuration des données pour l'esclave avec les variables d'heure
def configure_server(ip_address, port, slave_id):
# Créer un tableau avec des valeurs pour chaque registre
register_values = [
year, # Année dans le premier registre
month, # Mois dans le deuxième registre
day, # Jour dans le troisième registre
hour, # Heure dans le quatrième registre
minute, # Minute dans le cinquième registre
week_number, # Minute dans le sixième registre
]
# Compléter les autres registres avec une valeur par défaut (par exemple 0)
register_values += [0] * (20 - len(register_values)) # Compléter jusqu'à 100 registres
store = ModbusSlaveContext(
hr=ModbusSequentialDataBlock(1, register_values) # Registres holding pour l'esclave
)
# Associer l'esclave à l'ID fourni par l'utilisateur
context = ModbusServerContext(
slaves={slave_id: store}, # Associe le contexte à l'ID
single=False
)
# Configuration des informations sur l'appareil
identity = ModbusDeviceIdentification()
identity.VendorName = "Mon Serveur Modbus"
identity.ProductCode = "MODBUS"
identity.VendorUrl = "https://example.com"
identity.ProductName = "Serveur Modbus TCP"
identity.ModelName = "Modbus Server"
identity.MajorMinorRevision = "1.0"
# Adresse et port du serveur
address = (ip_address, port)
return context, identity, address
async def start_server(context, identity, address):
"""Démarre le serveur Modbus."""
server = ModbusTcpServer(context, identity=identity, address=address)
log.info(f"Serveur Modbus démarré sur {address[0]}:{address[1]}")
await server.serve_forever()
def main():
# Demander les informations de configuration pour le serveur
ip_address = get_ip_address()
port = get_port()
slave_id = get_slave_id()
# Configurer le serveur avec les informations entrées par l'utilisateur
context, identity, address = configure_server(ip_address, port, slave_id)
# Démarrer le serveur Modbus
asyncio.run(start_server(context, identity, address))
if __name__ == "__main__":
main()
