mirror of
https://github.com/markqvist/Reticulum.git
synced 2024-12-27 08:29:36 -05:00
171 lines
6.0 KiB
Python
171 lines
6.0 KiB
Python
##########################################################
|
|
# This RNS example demonstrates setting up announce #
|
|
# callbacks, which will let an application receive a #
|
|
# notification when an announce relevant for it arrives #
|
|
##########################################################
|
|
|
|
import argparse
|
|
import random
|
|
import RNS
|
|
|
|
# Let's define an app name. We'll use this for all
|
|
# destinations we create. Since this basic example
|
|
# is part of a range of example utilities, we'll put
|
|
# them all within the app namespace "example_utilities"
|
|
APP_NAME = "example_utilities"
|
|
|
|
# We initialise two lists of strings to use as app_data
|
|
fruits = ["Peach", "Quince", "Date", "Tangerine", "Pomelo", "Carambola", "Grape"]
|
|
noble_gases = ["Helium", "Neon", "Argon", "Krypton", "Xenon", "Radon", "Oganesson"]
|
|
|
|
# This initialisation is executed when the program is started
|
|
def program_setup(configpath):
|
|
# We must first initialise Reticulum
|
|
reticulum = RNS.Reticulum(configpath)
|
|
|
|
# Randomly create a new identity for our example
|
|
identity = RNS.Identity()
|
|
|
|
# Using the identity we just created, we create two destinations
|
|
# in the "example_utilities.announcesample" application space.
|
|
#
|
|
# Destinations are endpoints in Reticulum, that can be addressed
|
|
# and communicated with. Destinations can also announce their
|
|
# existence, which will let the network know they are reachable
|
|
# and autoomatically create paths to them, from anywhere else
|
|
# in the network.
|
|
destination_1 = RNS.Destination(
|
|
identity,
|
|
RNS.Destination.IN,
|
|
RNS.Destination.SINGLE,
|
|
APP_NAME,
|
|
"announcesample",
|
|
"fruits"
|
|
)
|
|
|
|
destination_2 = RNS.Destination(
|
|
identity,
|
|
RNS.Destination.IN,
|
|
RNS.Destination.SINGLE,
|
|
APP_NAME,
|
|
"announcesample",
|
|
"noble_gases"
|
|
)
|
|
|
|
# We configure the destinations to automatically prove all
|
|
# packets adressed to it. By doing this, RNS will automatically
|
|
# generate a proof for each incoming packet and transmit it
|
|
# back to the sender of that packet. This will let anyone that
|
|
# tries to communicate with the destination know whether their
|
|
# communication was received correctly.
|
|
destination_1.set_proof_strategy(RNS.Destination.PROVE_ALL)
|
|
destination_2.set_proof_strategy(RNS.Destination.PROVE_ALL)
|
|
|
|
# We create an announce handler and configure it to only ask for
|
|
# announces from "example_utilities.announcesample.fruits".
|
|
# Try changing the filter and see what happens.
|
|
announce_handler = ExampleAnnounceHandler(
|
|
aspect_filter="example_utilities.announcesample.fruits"
|
|
)
|
|
|
|
# We register the announce handler with Reticulum
|
|
RNS.Transport.register_announce_handler(announce_handler)
|
|
|
|
# Everything's ready!
|
|
# Let's hand over control to the announce loop
|
|
announceLoop(destination_1, destination_2)
|
|
|
|
|
|
def announceLoop(destination_1, destination_2):
|
|
# Let the user know that everything is ready
|
|
RNS.log("Announce example running, hit enter to manually send an announce (Ctrl-C to quit)")
|
|
|
|
# We enter a loop that runs until the users exits.
|
|
# If the user hits enter, we will announce our server
|
|
# destination on the network, which will let clients
|
|
# know how to create messages directed towards it.
|
|
while True:
|
|
entered = input()
|
|
|
|
# Randomly select a fruit
|
|
fruit = fruits[random.randint(0,len(fruits)-1)]
|
|
|
|
# Send the announce including the app data
|
|
destination_1.announce(app_data=fruit.encode("utf-8"))
|
|
RNS.log(
|
|
"Sent announce from "+
|
|
RNS.prettyhexrep(destination_1.hash)+
|
|
" ("+destination_1.name+")"
|
|
)
|
|
|
|
# Randomly select a noble gas
|
|
noble_gas = noble_gases[random.randint(0,len(noble_gases)-1)]
|
|
|
|
# Send the announce including the app data
|
|
destination_2.announce(app_data=noble_gas.encode("utf-8"))
|
|
RNS.log(
|
|
"Sent announce from "+
|
|
RNS.prettyhexrep(destination_2.hash)+
|
|
" ("+destination_2.name+")"
|
|
)
|
|
|
|
# We will need to define an announce handler class that
|
|
# Reticulum can message when an announce arrives.
|
|
class ExampleAnnounceHandler:
|
|
# The initialisation method takes the optional
|
|
# aspect_filter argument. If aspect_filter is set to
|
|
# None, all announces will be passed to the instance.
|
|
# If only some announces are wanted, it can be set to
|
|
# an aspect string.
|
|
def __init__(self, aspect_filter=None):
|
|
self.aspect_filter = aspect_filter
|
|
|
|
# This method will be called by Reticulums Transport
|
|
# system when an announce arrives that matches the
|
|
# configured aspect filter. Filters must be specific,
|
|
# and cannot use wildcards.
|
|
def received_announce(self, destination_hash, announced_identity, app_data):
|
|
RNS.log(
|
|
"Received an announce from "+
|
|
RNS.prettyhexrep(destination_hash)
|
|
)
|
|
|
|
if app_data:
|
|
RNS.log(
|
|
"The announce contained the following app data: "+
|
|
app_data.decode("utf-8")
|
|
)
|
|
|
|
##########################################################
|
|
#### Program Startup #####################################
|
|
##########################################################
|
|
|
|
# This part of the program gets run at startup,
|
|
# and parses input from the user, and then starts
|
|
# the desired program mode.
|
|
if __name__ == "__main__":
|
|
try:
|
|
parser = argparse.ArgumentParser(
|
|
description="Reticulum example that demonstrates announces and announce handlers"
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--config",
|
|
action="store",
|
|
default=None,
|
|
help="path to alternative Reticulum config directory",
|
|
type=str
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.config:
|
|
configarg = args.config
|
|
else:
|
|
configarg = None
|
|
|
|
program_setup(configarg)
|
|
|
|
except KeyboardInterrupt:
|
|
print("")
|
|
exit() |