ftespnow: Initial commit

Author: Schneck2004

.gitignore

@@ -1,5 +1,6 @@
 MANIFEST
 __pycache__
+.vscode/
 *.egg-info
 */dist/
 *.org

micropython/ftespnow/ftespnow-client/ftespnow/client.py

@@ -0,0 +1,77 @@
+import espnow
+import json
+
+class CLIENT:
+    def __init__(self) -> None:
+        self.esp = espnow.ESPNow()
+    
+    def configure(self, timeout=5) -> None:
+        self.timeout: int = timeout
+
+    def connect(self, peer: str) -> None:
+        self.peer: str = peer
+        self.esp.active(True)
+        self.esp.add_peer(self.peer)
+
+    def send_message(self, data: str) -> bool:
+        ack: bool = self.esp.send(self.peer, data)
+        return ack
+            
+    def receive_message(self, recv_timeout :int= 5) -> list | None:
+        received = self.esp.recv(recv_timeout)
+        if received[0] == None: return
+        return received
+    
+    def send_txt(self, filename: str) -> bool:
+        with open(filename, 'r') as f:
+            data: str = str(f.readlines())
+        sent: bool = self.send_message(data)
+        return sent
+    
+    def send_json(self, filename: str, *, indent: int=4) -> bool:
+        with open(filename, 'r') as f:
+            unparsed = json.load(f)
+        parsed: str = json.dumps(unparsed, indent=indent)
+        sent: bool = self.send_message(parsed)
+        return sent
+    
+    def receive_to_txt(self, target_file: str, mode: str='a') -> bool:
+        if ".txt" not in target_file: raise SyntaxError("File format must be .txt")
+        try:
+            data: list | None = self.receive_message()
+            if data == None: return False
+            data_list: list[str] = str(data[-1]).split("\n")
+            if data_list[-1] == "": data_list = data_list[:-1]
+            with open(target_file, mode) as f:
+                f.writelines(data_list)
+            return True
+        except SyntaxError:
+            raise
+    
+    def receive_to_json(self, target_file: str, mode: str='a') -> bool:
+        if ".json" not in target_file: raise SyntaxError("File format must be .json")
+        try:
+            data: list | None = self.receive_message()
+            if data == None: return False
+            mac: str = str(data[0])
+            message = json.loads(str(data[-1]))
+            unparsed: dict = {
+                "mac": mac,
+                "message": message
+            }
+            with open(target_file, mode) as f:
+                json.dump(unparsed, f)
+            return True
+        except SyntaxError:
+            raise
+    
+    def receive_to_dict(self) -> dict:
+        data: list | None = self.receive_message()
+        if data == None: return {}
+        mac: str = str(data[0])
+        message = json.loads(str(data[-1]))
+        unparsed: dict = {
+            "mac": mac,
+            "message": message
+        }
+        return unparsed

micropython/ftespnow/ftespnow-client/manifest.py

@@ -0,0 +1,9 @@
+metadata(
+    description="Client-side commands",
+    version="0.1.0",
+)
+
+require("ftespnow")
+require("json")
+require("espnow")
+package("ftespnow")

micropython/ftespnow/ftespnow-server/ftespnow/server.py

@@ -0,0 +1,45 @@
+import espnow
+import json
+
+class SERVER:
+    def __init__(self, *, timeout: int=5) -> None:
+        self.esp = espnow.ESPNow()
+        self.timeout = timeout
+    
+    def configure(self, *, timeout: int=5) -> None:
+        self.timeout = timeout
+    
+    def send_message(self, peer: str, data: str) -> bool:
+        ack: bool = self.esp.send(peer, data)
+        return ack
+    
+    def receive_message(self, recv_timeout :int= 5) -> list | None:
+        received = self.esp.recv(recv_timeout)
+        if received[0] == None: return
+        return received
+    
+    def send_txt(self, peer: str, filename: str) -> bool:
+        with open(filename, 'r') as f:
+            data: str = str(f.readlines())
+        sent: bool = self.send_message(peer, data)
+        return sent
+    
+    def send_json(self, peer: str, filename: str, *, indent: int=4) -> bool:
+        with open(filename, 'r') as f:
+            unparsed = json.load(f)
+        parsed: str = json.dumps(unparsed, indent=indent)
+        sent: bool = self.send_message(peer, parsed)
+        return sent
+    
+    def receive_to_txt(self, target_file: str, mode: str='a') -> bool:
+        if ".txt" not in target_file: raise SyntaxError("File format must be .txt")
+        try:
+            data: list | None = self.receive_message()
+            if data == None: return False
+            data_list: list[str] = str(data[-1]).split("\n")
+            if data_list[-1] == "": data_list = data_list[:-1]
+            with open(target_file, mode) as f:
+                f.writelines(data_list)
+            return True
+        except SyntaxError:
+            raise

micropython/ftespnow/ftespnow-server/manifest.py

@@ -0,0 +1,9 @@
+metadata(
+    description="Server-side commands",
+    version="0.1.0",
+)
+
+require("ftespnow")
+require("json")
+require("espnow")
+package("ftespnow")

micropython/ftespnow/ftespnow/__init__.py

@@ -0,0 +1,9 @@
+try:
+    from .clinet import *
+except ImportError:
+    pass
+
+try:
+    from.server import *
+except ImportError:
+    pass

micropython/ftespnow/manifest.py

@@ -0,0 +1,6 @@
+metadata(
+    description="Extends the micropython espnow module with methods to support file transfers.",
+    version="0.1.0",
+)
+
+module("ftespnow")