more asyncloop tests

app/main.py

@@ -6,6 +6,11 @@ DEFAULT_SCREEN='profile'
 HORIZONTAL = False
 WINDOW_SIZE = (1136,640) if HORIZONTAL else (640,1136)
 
+import asyncio
+import os
+os.environ['KIVY_EVENTLOOP'] = 'async'
+# loop = asyncio.get_event_loop()
+# loop.set_debug(True)
 
 # imports
 from kivy.uix.screenmanager import Screen,ScreenManager
@@ -48,19 +53,19 @@ Window.size = WINDOW_SIZE
 # with open('log.txt','w') as of:
 #     of.write('### LOG ###\n')
 
-# import logging
-# handler = logging.StreamHandler()
-# formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
-# handler.setFormatter(formatter)
-# logger = logging.getLogger('app')
-# logger.addHandler(handler)
-# logger.setLevel(logging.DEBUG)
+import logging
+handler = logging.StreamHandler()
+formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+handler.setFormatter(formatter)
+logger = logging.getLogger('app')
+logger.addHandler(handler)
+logger.setLevel(logging.DEBUG)
 
 def log(*args):
-    with open('log.txt','a+') as of:
-        of.write(' '.join([str(x) for x in args])+'\n')
-    # line = ' '.join(str(x) for x in args)
-    # logger.debug(line)
+    # with open('log.txt','a+') as of:
+        # of.write(' '.join([str(x) for x in args])+'\n')
+    line = ' '.join(str(x) for x in args)
+    logger.debug(line)
 
 class MyLayout(MDBoxLayout):
     scr_mngr = ObjectProperty(None)
@@ -146,6 +151,8 @@ class MainApp(MDApp):
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
         self.event_loop_worker = None
+        self.loop=asyncio.get_event_loop()
+
 
     
 
@@ -168,7 +175,7 @@ class MainApp(MDApp):
         # self.texture.uvsize = (-2, -2)
         # self.start_event_loop_thread()
 
-        with open('log.txt','w') as of: of.write('## LOG ##\n')
+        # with open('log.txt','w') as of: of.write('## LOG ##\n')
         self.load_store()
 
         # self.boot_kad()
@@ -364,11 +371,14 @@ class MainApp(MDApp):
 
 
 
-if __name__ == '__main__':
-
-    #### LOGIN
-
-
-
+def main():
     App = MainApp()
     App.run()
+
+
+if __name__ == '__main__':
+    # loop = asyncio.get_event_loop()
+    # asyncio.set_event_loop(loop)
+    # loop.run_until_complete(main())
+    # loop.close()
+    main()

app/screens/feed/feed.py

@@ -144,8 +144,9 @@ class PostCard(MDCard):
                 self.image.reload()
 
             #self.open_dialog('posting')
-            Thread(target=do_download).start()
-        
+            #Thread(target=do_download).start()
+            do_download()
+
 
     @property
     def app(self):

app/screens/post/post.py

@@ -133,8 +133,9 @@ class PostScreen(ProtectedScreen):
         # self.open_dialog('uploading')
         # self.upload()
         # self.close_dialog()
-        mythread = threading.Thread(target=self.upload)
-        mythread.start()
+        #mythread = threading.Thread(target=self.upload)
+        #mythread.start()
+        self.upload()
 
     def upload(self):
         # get file id
@@ -199,8 +200,8 @@ class PostScreen(ProtectedScreen):
             self.close_dialog()
         
         self.open_dialog('posting')
-        Thread(target=do_post).start()
-        
+        #Thread(target=do_post).start()
+        do_post()
         
 
 # class ViewPostScreen(ProtectedScreen): 

app/watcher.py

@@ -13,7 +13,7 @@ def run_command():
     p = subprocess.Popen(cmd.split(), stdout=subprocess.PIPE)
 
 class Watcher:
-    DIRECTORY_TO_WATCH = "/home/ryan/github/Komrade/client"
+    DIRECTORY_TO_WATCH = os.path.join(os.path.expanduser('~'),"/github/Komrade/p2p")
 
     def __init__(self):
         self.observer = Observer()

p2p/api.py

@@ -43,18 +43,27 @@ class Api(object):
 
 
     def __init__(self,app_storage):
-        #self.connect()
+        self._node=self.connect()
         self.app_storage = app_storage
-        # self.node = self.connect()
+        
         # log('starting selfless daemon...')
         # self.selfless = Thread(target=start_selfless_thread)
         # self.selfless.daemon = True
         # self.selfless.start()
         pass
 
+    @property
+    def node(self):
+        if not hasattr(self,'_node'):
+            self._node=self.connect()
+        return self._node
+
     def connect(self):
         log('connecting...')
-        async def _connect():
+        #loop=asyncio.get_event_loop()
+
+
+        async def _getdb():
             from .kad import KadServer
             log('starting server..')
             node = KadServer() #storage=HalfForgetfulStorage())
@@ -63,16 +72,32 @@ class Api(object):
             await node.bootstrap(NODES_PRIME)
             return node
 
+        async def _connect():
+            self._node0 = node = await _getdb() #await loop.create_task(_getdb())
+            # log('!!!',type(self._node))
+            #await node
+            #self.node = node
+            return node
+
+        # return asyncio.run(_connect())
+        # loop.set_debug(True)
+        # log('loop???',loop)
         return asyncio.run(_connect())
 
 
     def get(self,key_or_keys):
         from .kad import KadServer
+        # loop=asyncio.get_event_loop()
+        # asyncio.set_event_loop(loop)
 
         async def _get():
-            node = KadServer() #storage=HalfForgetfulStorage())
-            await node.listen(PORT_LISTEN)
-            await node.bootstrap(NODES_PRIME)
+            try:
+                await self.node
+            except TypeError:
+                pass
+
+            log('wtf??',self.node)
+            node = self.node
             # node=self.node
             
             if type(key_or_keys) in {list,tuple,dict}:
@@ -84,14 +109,17 @@ class Api(object):
                 key = key_or_keys
                 res = await node.get(key)
 
-            node.stop()
+            # node.stop()
             return res
             
         return asyncio.run(_get())
 
+        # return loop.create_task(_get())
+
 
     def get_json(self,key_or_keys):
         res = self.get(key_or_keys)
+        log('GET_JSON',res)
         if type(res)==list:
             return [None if x is None else json.loads(x) for x in res]
         else:
@@ -100,16 +128,14 @@ class Api(object):
 
     def set(self,key_or_keys,value_or_values):
         # log('hello?')
-        from .kad import KadServer
+        # loop=asyncio.get_event_loop()
+
         async def _set():
-            # log('starting server...')
-            node = KadServer() #storage=HalfForgetfulStorage())
-            
-            # log('listening...')
-            await node.listen(PORT_LISTEN)
-            
-            # log('bootstrapping...')
-            await node.bootstrap(NODES_PRIME)
+            try:
+                await self.node
+            except TypeError:
+                pass
+            node=self.node
             
 
             if type(key_or_keys) in {list,tuple,dict}:
@@ -124,8 +150,12 @@ class Api(object):
                 value = value_or_values
                 res = await node.set(key,value) #'this is a test')
 
-            node.stop()
+            # node.stop()
             return res
+
+        # loop=asyncio.get_event_loop()
+        # loop.create_task(_set())
+
         return asyncio.run(_set(), debug=True)
 
     def set_json(self,key,value):

p2p/kad.py

@@ -136,6 +136,12 @@ class KadProtocol(KademliaProtocol):
     # remote_address = PROXY_ADDR
     # REMOTES_D={}
 
+    def __init__(self, source_node, storage, ksize):
+        RPCProtocol.__init__(self,wait_timeout=5)
+        self.router = RoutingTable(self, ksize, source_node)
+        self.storage = storage
+        self.source_node = source_node
+
     # def datagram_received(self, data, addr):
     #     #if not hasattr(self,'remotes_d'): self.remotes_d={}
     #     # print('\n\n!?!?!?',self.REMOTES_D, type(self.REMOTES_D))
@@ -167,6 +173,31 @@ class KadServer(Server):
     protocol_class = KadProtocol # KadProtocol #KademliaProtocol
 
 
-    pass
+    async def set_digest(self, dkey, value):
+        """
+        Set the given SHA1 digest key (bytes) to the given value in the
+        network.
+        """
+        node = Node(dkey)
+
+        nearest = self.protocol.router.find_neighbors(node)
+        if not nearest:
+            log.warning("There are no known neighbors to set key %s",
+                        dkey.hex())
+            return False
+
+        spider = NodeSpiderCrawl(self.protocol, node, nearest,
+                                 self.ksize, self.alpha)
+        nodes = await spider.find()
+        log.info("setting '%s' on %s", dkey.hex(), list(map(str, nodes)))
+
+        # if this node is close too, then store here as well
+        neighbs=[n.distance_to(node) for n in nodes]
+        biggest = max(neighbs) if neighbs else 0
+        if self.node.distance_to(node) < biggest:
+            self.storage[dkey] = value
+        results = [self.protocol.call_store(n, dkey, value) for n in nodes]
+        # return true only if at least one store call succeeded
+        return any(await asyncio.gather(*results))
 
 

p2p/p2p.py

@@ -8,7 +8,7 @@ import pickle,os
 
 # NODES_PRIME = [("128.232.229.63",8468), ("68.66.241.111",8468)]    
 
-NODES_PRIME = [("68.66.241.111",8467)]
+NODES_PRIME = [("68.66.241.111",8467), ("10.42.0.13",8467)]
 
 async def echo(msg):
     print('echo',msg)