236 lines
8.5 KiB
Python
236 lines
8.5 KiB
Python
import socket
|
|
import threading
|
|
import time
|
|
|
|
from Participant import Participant
|
|
from Thing import Thing
|
|
from Messages.ParticipantMsg import ParticipantMsg
|
|
from Messages.NetworkIdMsg import NetworkIdMsg
|
|
from Messages.ThingMsg import ThingMsg
|
|
from Messages.NameMsg import NameMsg
|
|
from Messages.ModelUrlMsg import ModelUrlMsg
|
|
from Messages import *
|
|
|
|
import sys
|
|
micropython = 'micropython' in sys.modules
|
|
|
|
if micropython:
|
|
from MicroPython.uPythonParticipant import Bla
|
|
|
|
class ParticipantUDP(Participant):
|
|
"""! A local participant is the local device which can communicate with other participants.
|
|
|
|
It manages all local things and communcation with other participants. Each application has a local participant which is usually explicit in the code.
|
|
An participant can be isolated. In that case it is standalong and does not communicate with other participants.
|
|
|
|
Currently, only UDP communication is supported
|
|
"""
|
|
|
|
buffer = None
|
|
nextPublishMe = 0
|
|
others = None
|
|
thread = None
|
|
name = "Participant"
|
|
isolated_participant = None
|
|
|
|
def __init__(self, port=7681, ip_address=None, local_port=0):
|
|
super().__init__(ip_address = "127.0.0.1", port = port)
|
|
|
|
if local_port == 0:
|
|
local_port = port
|
|
self.local_port = local_port
|
|
|
|
## True if the participant is running isolated.
|
|
# Isolated participants do not communicate with other participants
|
|
self.is_isolated = True
|
|
self.remote_site = None
|
|
|
|
## The other participants communicating with this participant
|
|
self.others = []
|
|
|
|
if self.port != 0:
|
|
self.is_isolated = False
|
|
if ip_address is not None:
|
|
self.remote_site = Participant(ip_address, port)
|
|
self.others.append(self.remote_site)
|
|
|
|
self.buffer = bytearray(256)
|
|
self.publishInterval = 3000 # 3 seconds
|
|
|
|
self.udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
self.udp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
|
|
self.udp_socket.bind(("0.0.0.0", self.local_port))
|
|
|
|
self.thread = threading.Thread(target = self.Receiver)
|
|
self.thread.daemon = True
|
|
self.thread.start()
|
|
|
|
def Isolated():
|
|
if ParticipantUDP.isolated_participant == None:
|
|
ParticipantUDP.isolated_participant = ParticipantUDP(0)
|
|
return ParticipantUDP.isolated_participant
|
|
|
|
#region Update
|
|
|
|
def GetParticipant(self, ip_address, port):
|
|
# print(f'{self.name} Get participant {ip_address} {port}')
|
|
# for item in self.others:
|
|
# print(f'- {item.ip_address} {item.port}')
|
|
|
|
found_participants = (item for item in self.others
|
|
if item.ip_address == ip_address and item.port == port)
|
|
participant = next(found_participants, None)
|
|
return participant
|
|
|
|
def AddParticipant(self, ip_address, port):
|
|
print(f'{self.name} Add participant {ip_address} {port}')
|
|
remote_participant = Participant(ip_address = ip_address, port = port)
|
|
self.others.append(remote_participant)
|
|
return remote_participant
|
|
|
|
def Update(self, currentTimeMs = None):
|
|
if currentTimeMs is None:
|
|
currentTimeMs = time.time() * 1000
|
|
|
|
if self.is_isolated == False:
|
|
if self.publishInterval > 0 and currentTimeMs > self.nextPublishMe:
|
|
msg = ParticipantMsg(self.network_id)
|
|
if self.remote_site == None:
|
|
self.Publish(msg)
|
|
else:
|
|
self.Send(self.remote_site, msg)
|
|
|
|
self.nextPublishMe = currentTimeMs + self.publishInterval
|
|
|
|
self.UpdateMyThings(currentTimeMs)
|
|
# for thing in self.things:
|
|
# thing.Update(currentTimeMs)
|
|
|
|
# super().Update(currentTimeMs)
|
|
|
|
def UpdateMyThings(self, currentTimeMs):
|
|
for thing in self.things:
|
|
if thing is None:
|
|
continue
|
|
|
|
# if thing.hierarchyChanged and not (self.isIsolated or self.network_id == ):
|
|
# thingMsg = ThingMsg(self.network_id, thing)
|
|
# self.Send(self.remote_site, thingMsg)
|
|
|
|
thing.Update(currentTimeMs, False)
|
|
if not(self.is_isolated or self.network_id == 0):
|
|
# if thing.terminate:
|
|
# destroyMsg = DestroyMsg(self.network_id, thing)
|
|
# self.Send(self.remote_site, destroyMsg)
|
|
# else:
|
|
# Send to remote site
|
|
poseMsg = PoseMsg(thing.owner.network_id, thing)
|
|
self.Send(self.remote_site, poseMsg)
|
|
binaryMsg = BinaryMsg(thing.owner.network_id, thing)
|
|
self.Send(self.remote_site, binaryMsg)
|
|
# if thing.terminate:
|
|
# self.Remove(thing)
|
|
|
|
#endregion
|
|
|
|
#region Send
|
|
|
|
def SendThingInfo(self, owner, thing, recursively: bool = False):
|
|
self.Send(owner, ThingMsg(self.network_id, thing))
|
|
self.Send(owner, NameMsg(self.network_id, thing))
|
|
self.Send(owner, ModelUrlMsg(self.network_id, thing))
|
|
#self.Send(owner, PoseMsg(self.network_id, thing, True))
|
|
self.Send(owner, BinaryMsg(self.network_id, thing))
|
|
|
|
if recursively:
|
|
for child in thing.children:
|
|
self.SendThingInfo(owner, child, recursively)
|
|
|
|
|
|
def Send(self, owner, msg):
|
|
buffer_size = msg.Serialize([self.buffer])
|
|
if buffer_size <= 0:
|
|
return True
|
|
|
|
# print(f'{self.name} send {self.buffer[0]} to {owner.ip_address} {owner.port}')
|
|
self.udp_socket.sendto(self.buffer[:buffer_size], (owner.ip_address, owner.port))
|
|
return True
|
|
|
|
def Publish(self, msg):
|
|
buffer_size = msg.Serialize([self.buffer])
|
|
if buffer_size <= 0:
|
|
return True
|
|
|
|
# print(f'publish {self.buffer[0]} to {self.port}')
|
|
self.udp_socket.sendto(self.buffer[:buffer_size], ('<broadcast>', self.port))
|
|
return True
|
|
|
|
#endregion
|
|
|
|
#region Receive
|
|
|
|
def Receiver(self):
|
|
while True:
|
|
try:
|
|
data, addr = self.udp_socket.recvfrom(1024)
|
|
remote_ip_address = addr[0]
|
|
remote_port = addr[1]
|
|
# print(f'msg received from {remote_ip_address}:{remote_port}')
|
|
remote_participant = self.GetParticipant(remote_ip_address, remote_port)
|
|
if remote_participant is None:
|
|
# print(f'new participant')
|
|
remote_participant = self.AddParticipant(remote_ip_address, remote_port)
|
|
self.ReceiveData(data, remote_participant)
|
|
except ConnectionError:
|
|
self.network_id = 0
|
|
pass
|
|
|
|
def ReceiveData(self, data, sender):
|
|
msgId = data[0]
|
|
# print(f'msg {msgId} ')
|
|
match msgId:
|
|
case ParticipantMsg.id:
|
|
self.ProcessParticipantMsg(sender, ParticipantMsg(data))
|
|
case NetworkIdMsg.id:
|
|
self.ProcessSiteIdMsg(sender, NetworkIdMsg(data))
|
|
# case InvestigateMsg.id:
|
|
# self.ProcessInvestigateMsg(InvestigateMsg(data))
|
|
case ThingMsg.id:
|
|
self.ProcessThingMsg(ThingMsg(data))
|
|
case NameMsg.id:
|
|
self.ProcessNameMsg(NameMsg(data))
|
|
case ModelUrlMsg.id:
|
|
self.ProcessModelUrlMsg(ModelUrlMsg(data))
|
|
case BinaryMsg.id:
|
|
self.ProcessBinaryMsg(BinaryMsg(data))
|
|
|
|
def ProcessParticipantMsg(self, sender, msg: ParticipantMsg):
|
|
pass
|
|
|
|
def ProcessSiteIdMsg(self, sender, msg: NetworkIdMsg):
|
|
print(f'{self.name} Process SiteMsg {self.network_id} -> {msg.network_id}')
|
|
if self.network_id != msg.network_id:
|
|
self.network_id = msg.network_id
|
|
for thing in self.things:
|
|
#if thing.parent is None:
|
|
self.SendThingInfo(sender, thing, recursively=True)
|
|
|
|
def ProcessInvestigateMsg(self, data: bytearray):
|
|
pass
|
|
|
|
def ProcessThingMsg(self, msg: ThingMsg):
|
|
print(f'received thing {msg.thing_id}')
|
|
|
|
def ProcessNameMsg(self, msg: NameMsg):
|
|
print(f'received name {msg.name}')
|
|
|
|
def ProcessModelUrlMsg(self, msg: ModelUrlMsg):
|
|
print(f'received model url: {msg.url}')
|
|
|
|
def ProcessBinaryMsg(self, msg: BinaryMsg):
|
|
# print('received binary data')
|
|
thing: Thing = self.Get(msg.thing_id)
|
|
if thing != None:
|
|
thing.ProcessBinary(msg.data)
|
|
|
|
#endregion |