402 lines
13 KiB
Python
402 lines
13 KiB
Python
import meshtastic
|
|
import meshtastic.serial_interface
|
|
|
|
from pubsub import pub
|
|
from Packets.Message import Message
|
|
from Packets.Messages.Protocols.map.Announce import AnnounceMessage
|
|
|
|
from Packets.SubMessage import SubMessage
|
|
|
|
import time
|
|
import sys
|
|
import random
|
|
|
|
import msgpack
|
|
|
|
import asyncio
|
|
import uuid
|
|
|
|
import traceback
|
|
import logging
|
|
|
|
logger = logging.getLogger("__main__." + __name__)
|
|
|
|
class Transceiver:
|
|
"""
|
|
Handling LoRa transceiving
|
|
|
|
`🔗 Source <https://git.utopic.work/PierMesh/piermesh/src/branch/main/src/Transceiver/Transceiver.py>`__
|
|
|
|
Attributes
|
|
----------
|
|
cLog
|
|
Reference to `run.Node.cLog` for logging
|
|
|
|
cryptographyInfo: Cryptography.WhaleSong.DHEFern
|
|
Cryptography instance for encrypting transmissions
|
|
|
|
filter: Sponge.base.Filter
|
|
`Sponge.base.Filter` instance for filtering packets
|
|
|
|
tcache: Daisy.Cache.Cache
|
|
Data backend Daisy Cache
|
|
|
|
tcatch: Daisy.Catch.Catch
|
|
Daisy Catch Cache for Catch operations
|
|
|
|
notConnected: bool
|
|
Whether the transceiver has been connected to yet
|
|
|
|
acks: dict
|
|
Acknowledgements received per packet
|
|
|
|
onodeID
|
|
PierMesh node ID
|
|
|
|
messages: dict
|
|
Message completion acknowldgements
|
|
|
|
Notes
|
|
-----
|
|
TODO: Check if we can remove cpid
|
|
|
|
"""
|
|
|
|
def __init__(self, device, filter, onodeID, cache, catch, cryptographyInfo, network):
|
|
try:
|
|
self.busy = False
|
|
self.network = network
|
|
self.cryptographyInfo = cryptographyInfo
|
|
self.filter = filter
|
|
self.tcache = cache
|
|
self.tcatch = catch
|
|
self.notConnected = True
|
|
self.messages = {}
|
|
self.acks = {}
|
|
self.onodeID = onodeID
|
|
# Be careful with this
|
|
self.cpid = 0
|
|
self.tasks = {}
|
|
pub.subscribe(self.onReceive, "meshtastic.receive")
|
|
pub.subscribe(self.onConnection, "meshtastic.connection.established")
|
|
self.interface = meshtastic.serial_interface.SerialInterface(device)
|
|
i = 0
|
|
while self.notConnected:
|
|
if i % 5000000 == 0:
|
|
logger.log(20, "Waiting for node initialization...")
|
|
i += 1
|
|
logger.log(20, "Initialized")
|
|
except Exception as e:
|
|
logger.log(30, traceback.format_exc())
|
|
raise e
|
|
|
|
# TODO: Sending packets across multiple nodes/load balancing/distributed packet transmission/reception
|
|
def onReceive(self, packet, interface):
|
|
"""
|
|
Run each received packet through Sponge.base.Filters sieve using a new event loop
|
|
"""
|
|
self.busy = True
|
|
asyncio.new_event_loop().run_until_complete(self.filter.sieve(packet))
|
|
self.busy = False
|
|
self.tcache.refresh()
|
|
|
|
async def sendAnnounce(self, dontRespond=False):
|
|
"""
|
|
Send an announce packet (contains basic network mapping information) every so often so new nodes autoconnect
|
|
"""
|
|
try:
|
|
r = AnnounceMessage(
|
|
self.onodeID,
|
|
0,
|
|
self.onodeID,
|
|
self.cryptographyInfo,
|
|
{"onodeID": self.onodeID, "mnodeID": self.interface.localNode.nodeNum, "dontRespond": dontRespond},
|
|
)
|
|
await self.sendMessage(r)
|
|
except:
|
|
logger.log(30, traceback.format_exc())
|
|
|
|
def onConnection(self, interface, topic=pub.AUTO_TOPIC):
|
|
"""
|
|
When the node connects start announce loop and end the waiting loop
|
|
"""
|
|
#asyncio.run(self.sendAnnounce())
|
|
self.notConnected = False
|
|
|
|
def responseCheck(self, packet):
|
|
"""
|
|
On acknowldgement response set acks based on response
|
|
TODO: Stop this being sent to sieve
|
|
"""
|
|
rid = packet["decoded"]["requestId"]
|
|
if packet["decoded"]["routing"]["errorReason"] == "MAX_RETRANSMIT":
|
|
logger.debug(f"Got ack error for packet {rid}")
|
|
self.acks[str(rid)] = False
|
|
else:
|
|
self.acks[str(rid)] = True
|
|
|
|
# TODO: Threaded send method
|
|
|
|
def send(self, packet, recipientNode=False):
|
|
"""
|
|
Send individual packet
|
|
|
|
Parameters
|
|
----------
|
|
recipientNode
|
|
If set send to specified node
|
|
"""
|
|
interface = self.interface
|
|
if recipientNode == False:
|
|
pid = interface.sendData(
|
|
packet, wantAck=True, onResponse=self.responseCheck
|
|
)
|
|
else:
|
|
pid = interface.sendData(
|
|
packet,
|
|
destinationId=recipientNode,
|
|
wantAck=True,
|
|
onResponse=self.responseCheck,
|
|
)
|
|
|
|
# Can I use waitForAckNak on cpid?
|
|
self.cpid = pid.id
|
|
return True
|
|
|
|
async def awaitResponse(self, pid):
|
|
"""
|
|
Wait for acknowldgement response
|
|
"""
|
|
for i in range(120):
|
|
await asyncio.sleep(1)
|
|
if str(pid) in self.acks:
|
|
break
|
|
return True
|
|
|
|
async def initNodeDH(self, dhefOb, recipientNode, onodeID):
|
|
"""
|
|
Send Diffie Hellman initialization message
|
|
"""
|
|
await self.addPackets(
|
|
msgpack.dumps(
|
|
{"params": dhefOb.getParamsBytes(), "publicKey": dhefOb.publicKey}
|
|
),
|
|
self.onodeID,
|
|
000000,
|
|
000000,
|
|
onodeID,
|
|
directID=recipientNode,
|
|
packetsClass=3,
|
|
)
|
|
|
|
def awaitFullResponse(self, pid):
|
|
"""
|
|
TODO
|
|
|
|
Wait for message completed response
|
|
"""
|
|
for i in range(1_000_000_000):
|
|
time.sleep(5)
|
|
if pid in self.messages.keys():
|
|
if self.messages[pid]["finished"]:
|
|
break
|
|
return True
|
|
|
|
# TODO: Deprecate
|
|
async def addPackets(
|
|
self,
|
|
data,
|
|
sender,
|
|
senderName,
|
|
recipient,
|
|
recipientNode,
|
|
directID=False,
|
|
packetsClass=None,
|
|
encrypt=False,
|
|
):
|
|
"""
|
|
Convert binary data to Message and send each packet
|
|
Parameters
|
|
----------
|
|
data: bytes
|
|
Data to send
|
|
|
|
sender
|
|
Peer/Node ID of sender
|
|
|
|
senderName
|
|
ID matching specific user title
|
|
|
|
recipient
|
|
Peer/Node ID of recipient
|
|
|
|
recipientNode
|
|
Node ID of node to route to
|
|
|
|
directID
|
|
If set send to this Node only
|
|
|
|
packetsClass
|
|
Protocol for message
|
|
"""
|
|
tp = Message(
|
|
data,
|
|
sender,
|
|
senderName,
|
|
recipient,
|
|
recipientNode,
|
|
self.cryptographyInfo,
|
|
packetsClass=packetsClass,
|
|
)
|
|
for p in tp.packets:
|
|
if recipientNode == None:
|
|
self.send(p)
|
|
else:
|
|
logger.log(10, "Sending target: " + str(directID))
|
|
if directID != False:
|
|
recipientNode = directID
|
|
self.send(p, recipientNode=recipientNode)
|
|
awaitTask = asyncio.create_task(self.awaitResponse(self.cpid))
|
|
await asyncio.sleep(1)
|
|
# TODO: pid fix
|
|
currentTask = {
|
|
"ob": awaitTask,
|
|
"pid": str(self.cpid),
|
|
"packet": p,
|
|
"retry": False,
|
|
}
|
|
self.tasks[str(self.cpid)] = currentTask
|
|
|
|
async def sendMessage(self, message: Message):
|
|
|
|
try:
|
|
# self.busy = True
|
|
# TODO: Send nonce subMessage after main message
|
|
# TODO: Instantiate SubMessages with HeaderPacket of primary message
|
|
hpacket = message.fullPackets[0]
|
|
logger.log(30, message.nonce)
|
|
nsm = SubMessage(
|
|
hpacket.sender,
|
|
hpacket.senderDisplayName,
|
|
hpacket.sourceNode,
|
|
hpacket.recipient,
|
|
hpacket.recipientNode,
|
|
self.cryptographyInfo,
|
|
hpacket.packetsClass,
|
|
hpacket.pAction,
|
|
{"nonce": message.nonce, "tag": message.tag},
|
|
target=message.target,
|
|
primaryMessage=hpacket.packetsID
|
|
)
|
|
message.fullPackets[0].submessages.append(nsm.packetsID)
|
|
message.packets[0] = message.fullPackets[0].dump()
|
|
logger.log(30, message.packets[0])
|
|
logger.log(30, nsm.packets)
|
|
for m in [{"primary": message}, {"nonce": nsm}]:
|
|
while self.busy:
|
|
await asyncio.sleep(2)
|
|
m = [v for v in m.values()][0]
|
|
# logger.log(30, m)
|
|
await asyncio.sleep(5)
|
|
isHead = True
|
|
curPacketCount = len(m.packets)
|
|
for it, p in enumerate(m.packets):
|
|
it += 1
|
|
pid = str(uuid.uuid4())
|
|
#print(p)
|
|
logger.info(f"Packet {it:03d}/{curPacketCount:03d}, Size is: " + str(sys.getsizeof(p)))
|
|
doNumber = 3
|
|
if isHead:
|
|
doNumber = 5
|
|
isHead = False
|
|
if m.target == False:
|
|
logger.log(10, "Sending any")
|
|
for i in range(doNumber):
|
|
self.send(p)
|
|
await asyncio.sleep(1)
|
|
else:
|
|
logger.log(10, "Sending target: " + str(m.recipientNode))
|
|
for i in range(doNumber):
|
|
self.send(p, recipientNode=self.network.doLookup(m.recipientNode))
|
|
await asyncio.sleep(1)
|
|
awaitTask = asyncio.create_task(self.awaitResponse(pid))
|
|
|
|
currentTask = {
|
|
"ob": awaitTask,
|
|
"pid": pid,
|
|
"packet": p,
|
|
"retry": False,
|
|
}
|
|
self.tasks[pid] = currentTask
|
|
await asyncio.sleep(3)
|
|
except Exception:
|
|
logger.log(30, traceback.format_exc)
|
|
self.busy = False
|
|
|
|
async def progressCheck(self):
|
|
"""
|
|
Checks if acknowldgement was received per packet and if not resends
|
|
"""
|
|
while True:
|
|
await asyncio.sleep(480)
|
|
logger.log(
|
|
20, "Checking progress of {0} tasks".format(len(self.tasks.keys()))
|
|
)
|
|
doneFlag = True
|
|
dcTasks = [k for k in self.tasks.keys()]
|
|
for task in dcTasks:
|
|
task = self.tasks[task]
|
|
if task["ob"]:
|
|
if task["pid"] in self.acks:
|
|
if not self.acks[task["pid"]]:
|
|
retry = task["retry"]
|
|
remove = False
|
|
if retry == False:
|
|
retry = 1
|
|
elif retry < 3:
|
|
retry += 1
|
|
else:
|
|
logger.log(30, "Too many retries")
|
|
remove = True
|
|
if remove:
|
|
del self.tasks[task["pid"]]
|
|
else:
|
|
logger.log(20, "Doing retry")
|
|
doneFlag = False
|
|
# TODO: Resend to specific node
|
|
self.send(task["packet"])
|
|
await_thread = asyncio.create_task(
|
|
self.awaitResponse(task["pid"])
|
|
)
|
|
|
|
await asyncio.sleep(1)
|
|
currentTask = {
|
|
"ob": await_thread,
|
|
"pid": str(self.cpid),
|
|
"packet": task["packet"],
|
|
}
|
|
currentTask["retry"] = retry
|
|
self.tasks[task["pid"]] = currentTask
|
|
else:
|
|
del self.tasks[task["pid"]]
|
|
|
|
async def announce(self):
|
|
"""
|
|
Announce loop runner
|
|
"""
|
|
while (self.notConnected or self.busy):
|
|
if self.busy:
|
|
logger.log(30, "Transceiver is busy, waiting")
|
|
await asyncio.sleep(2)
|
|
while True:
|
|
wait = random.randrange(600, 900)
|
|
logger.info(f"Waiting {wait} seconds for next announce")
|
|
await asyncio.sleep(wait)
|
|
while self.busy:
|
|
logger.info("Transceiver is busy, waiting")
|
|
await asyncio.sleep(2)
|
|
await asyncio.sleep(5)
|
|
logger.info("Announce")
|
|
await asyncio.create_task(self.sendAnnounce())
|
|
await asyncio.sleep(1)
|