Merge branch 'feature/manual-at-commands' into develop

pyproject.toml

@@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta"
 [project]
 name = "pywaveshare"
 version = "0.5.0a0"
-dependencies = ["RPi.GPIO", "pyserial"]
+dependencies = ["pyserial", "RPi.GPIO", "typing-extensions"]
 requires-python = ">=3.7"
 authors = [{ name = "Tarek Tounsi", email = "software@tounsi.de" }]
 maintainers = [

src/pywaveshare/boards/sim868/client.py

@@ -1,4 +1,4 @@
-from . import config, worker
+from . import config, itc, worker
 
 
 class Client:
@@ -7,4 +7,10 @@ class Client:
     def __init__(self, config: config.Config) -> None:
         self.config = config
         self.worker = worker.Worker(self.config)
+
+    def start(self) -> None:
+        itc.IS_WORKER_RUNNING.set()
         self.worker.start()
+
+    def send_at_command(self, command: str):
+        pass

src/pywaveshare/boards/sim868/itc.py

@@ -5,7 +5,12 @@
 
 IS_WORKER_RUNNING = threading.Event()
 
-ACQUIRE_SERIAL_OBJECT = threading.Lock()
+SERIAL_LOCK = threading.Lock()
+C2W_Q_LOCK = threading.Lock()
+"""
+If acquired, then the acquirer can to safely work on `CLIENT_TO_WORKER_Q` queue.
+"""
 
-ON_SERIAL_RX: queue.Queue[bytes] = queue.Queue()
-ON_SERIAL_TX: queue.Queue[bytes] = queue.Queue()
+CLIENT_TO_WORKER_Q: queue.Queue[bytes] = queue.Queue()
+WORKER_TO_SERIAL_Q: queue.Queue[bytes] = queue.Queue()
+SERIAL_TO_WORKER_Q: queue.Queue[bytes] = queue.Queue()

src/pywaveshare/boards/sim868/protocols.py

@@ -2,7 +2,30 @@
 import typing
 
 
-class SupportedProtocol(abc.ABC):
+class SupportedStandard(abc.ABC):
     NAME: typing.Optional[str] = None
 
     RESPONSE_PATTERN = r""
+
+
+class SMS(SupportedStandard):
+    NAME = "SMS"
+
+    RESPONSE_PATTERN = (
+        r"loremipsumdolorsitamet"
+        r"loremipsumdolorsitamet"
+        r"loremipsumdolorsitamet"
+        r"loremipsumdolorsitamet"
+        r"loremipsumdolorsitamet"
+        r"loremipsumdolorsitamet"
+        r"loremipsumdolorsitamet"
+    )
+
+
+class SupportedProtocolFactory:
+    pass
+
+
+if __name__ == "__main__":
+    protocol = SMS()
+    print(protocol.RESPONSE_PATTERN)

src/pywaveshare/boards/sim868/typedefs.py

@@ -0,0 +1,11 @@
+import typing
+
+import typing_extensions
+
+SMSSend = typing_extensions.TypedDict(
+    "SMSSend",
+    {
+        "Recipients": typing.List[str],
+        "Message": str,
+    },
+)

src/pywaveshare/boards/sim868/worker.py

@@ -1,6 +1,6 @@
 import threading
 
-from . import config, itc
+from . import config, itc, protocols
 
 
 class Worker(threading.Thread):
@@ -10,8 +10,8 @@ def __init__(self, config: config.Config) -> None:
         super().__init__()
 
         self.config = config
+        self.protocols = protocols.SupportedProtocolFactory()
 
     def run(self) -> None:
-        itc.IS_WORKER_RUNNING.set()
         while itc.IS_WORKER_RUNNING.is_set():
             pass