Announce.py
1 ########################################################## 2 # This RNS example demonstrates setting up announce # 3 # callbacks, which will let an application receive a # 4 # notification when an announce relevant for it arrives # 5 ########################################################## 6 7 import argparse 8 import random 9 import sys 10 import RNS 11 12 # Let's define an app name. We'll use this for all 13 # destinations we create. Since this basic example 14 # is part of a range of example utilities, we'll put 15 # them all within the app namespace "example_utilities" 16 APP_NAME = "example_utilities" 17 18 # We initialise two lists of strings to use as app_data 19 fruits = ["Peach", "Quince", "Date", "Tangerine", "Pomelo", "Carambola", "Grape"] 20 noble_gases = ["Helium", "Neon", "Argon", "Krypton", "Xenon", "Radon", "Oganesson"] 21 22 # This initialisation is executed when the program is started 23 def program_setup(configpath): 24 # We must first initialise Reticulum 25 reticulum = RNS.Reticulum(configpath) 26 27 # Randomly create a new identity for our example 28 identity = RNS.Identity() 29 30 # Using the identity we just created, we create two destinations 31 # in the "example_utilities.announcesample" application space. 32 # 33 # Destinations are endpoints in Reticulum, that can be addressed 34 # and communicated with. Destinations can also announce their 35 # existence, which will let the network know they are reachable 36 # and automatically create paths to them, from anywhere else 37 # in the network. 38 destination_1 = RNS.Destination( 39 identity, 40 RNS.Destination.IN, 41 RNS.Destination.SINGLE, 42 APP_NAME, 43 "announcesample", 44 "fruits" 45 ) 46 47 destination_2 = RNS.Destination( 48 identity, 49 RNS.Destination.IN, 50 RNS.Destination.SINGLE, 51 APP_NAME, 52 "announcesample", 53 "noble_gases" 54 ) 55 56 # We configure the destinations to automatically prove all 57 # packets addressed to it. By doing this, RNS will automatically 58 # generate a proof for each incoming packet and transmit it 59 # back to the sender of that packet. This will let anyone that 60 # tries to communicate with the destination know whether their 61 # communication was received correctly. 62 destination_1.set_proof_strategy(RNS.Destination.PROVE_ALL) 63 destination_2.set_proof_strategy(RNS.Destination.PROVE_ALL) 64 65 # We create an announce handler and configure it to only ask for 66 # announces from "example_utilities.announcesample.fruits". 67 # Try changing the filter and see what happens. 68 announce_handler = ExampleAnnounceHandler( 69 aspect_filter="example_utilities.announcesample.fruits" 70 ) 71 72 # We register the announce handler with Reticulum 73 RNS.Transport.register_announce_handler(announce_handler) 74 75 # Everything's ready! 76 # Let's hand over control to the announce loop 77 announceLoop(destination_1, destination_2) 78 79 80 def announceLoop(destination_1, destination_2): 81 # Let the user know that everything is ready 82 RNS.log("Announce example running, hit enter to manually send an announce (Ctrl-C to quit)") 83 84 # We enter a loop that runs until the users exits. 85 # If the user hits enter, we will announce our server 86 # destination on the network, which will let clients 87 # know how to create messages directed towards it. 88 while True: 89 entered = input() 90 91 # Randomly select a fruit 92 fruit = fruits[random.randint(0,len(fruits)-1)] 93 94 # Send the announce including the app data 95 destination_1.announce(app_data=fruit.encode("utf-8")) 96 RNS.log( 97 "Sent announce from "+ 98 RNS.prettyhexrep(destination_1.hash)+ 99 " ("+destination_1.name+")" 100 ) 101 102 # Randomly select a noble gas 103 noble_gas = noble_gases[random.randint(0,len(noble_gases)-1)] 104 105 # Send the announce including the app data 106 destination_2.announce(app_data=noble_gas.encode("utf-8")) 107 RNS.log( 108 "Sent announce from "+ 109 RNS.prettyhexrep(destination_2.hash)+ 110 " ("+destination_2.name+")" 111 ) 112 113 # We will need to define an announce handler class that 114 # Reticulum can message when an announce arrives. 115 class ExampleAnnounceHandler: 116 # The initialisation method takes the optional 117 # aspect_filter argument. If aspect_filter is set to 118 # None, all announces will be passed to the instance. 119 # If only some announces are wanted, it can be set to 120 # an aspect string. 121 def __init__(self, aspect_filter=None): 122 self.aspect_filter = aspect_filter 123 124 # This method will be called by Reticulums Transport 125 # system when an announce arrives that matches the 126 # configured aspect filter. Filters must be specific, 127 # and cannot use wildcards. 128 def received_announce(self, destination_hash, announced_identity, app_data): 129 RNS.log( 130 "Received an announce from "+ 131 RNS.prettyhexrep(destination_hash) 132 ) 133 134 if app_data: 135 RNS.log( 136 "The announce contained the following app data: "+ 137 app_data.decode("utf-8") 138 ) 139 140 ########################################################## 141 #### Program Startup ##################################### 142 ########################################################## 143 144 # This part of the program gets run at startup, 145 # and parses input from the user, and then starts 146 # the desired program mode. 147 if __name__ == "__main__": 148 try: 149 parser = argparse.ArgumentParser( 150 description="Reticulum example that demonstrates announces and announce handlers" 151 ) 152 153 parser.add_argument( 154 "--config", 155 action="store", 156 default=None, 157 help="path to alternative Reticulum config directory", 158 type=str 159 ) 160 161 args = parser.parse_args() 162 163 if args.config: 164 configarg = args.config 165 else: 166 configarg = None 167 168 program_setup(configarg) 169 170 except KeyboardInterrupt: 171 print("") 172 sys.exit(0)