Merge branch 'develop' into feature/publishing

Author: acmacunlay

.github/workflows/pipeline.yaml

@@ -54,6 +54,10 @@ jobs:
         run: |
           mypy .
 
-      - name: Run unit tests
+      - name: Run common security issue checker
         run: |
-          pytest .
+          bandit -c pyproject.toml -r .
+      
+      - name: Run unit tests + code coverage
+        run: |
+          coverage run -m pytest .

.gitignore

@@ -1,8 +1,10 @@
 __pycache__
 *.egg-info
+*.log
+.coverage
 .mypy_cache
+.pypirc
 .pytest_cache
 .ruff_cache
-.pypirc
 .venv
 dist

README.md

@@ -3,7 +3,7 @@
 With pywaveshare, you can easily use the functionality of the [Waveshare GSM/GPRS/GNSS HAT for Raspberry Pi](https://www.waveshare.com/gsm-gprs-gnss-hat.htm). On this module a SIM868 Controller is doing the job to connect your Raspberry Pi with the world just by using a SIM card.
 
 ## Changelog
-[See this document.](https://github.com/acmacunlay/pywaveshare/blob/master/CHANGELOG.md)
+[See this document.](https://github.com/acmacunlay/pywaveshare/blob/main/CHANGELOG.md)
 
 ## Overview
 pywaveshare was written for Python 3. It provides the following features
@@ -180,4 +180,4 @@ if gsm.UrlResponse_available() > 0:
 
 
 ## License
-[See this document.](https://github.com/acmacunlay/pywaveshare/blob/master/LICENSE)
+[See this document.](https://github.com/acmacunlay/pywaveshare/blob/main/LICENSE)

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 = "[email protected]" }]
 maintainers = [
@@ -38,13 +38,25 @@ classifiers = [
 ]
 
 [project.optional-dependencies]
-dev = ["build", "mypy", "pytest", "ruff", "setuptools", "twine"]
+dev = [
+    "bandit",
+    "build",
+    "coverage",
+    "mypy",
+    "pytest",
+    "ruff",
+    "setuptools",
+    "twine",
+]
 
 [project.urls]
 Homepage = "https://github.com/acmacunlay/pywaveshare"
 Repository = "https://github.com/acmacunlay/pywaveshare.git"
-# TODO: Changelog = ""
+Changelog = "https://github.com/acmacunlay/pywaveshare/blob/main/CHANGELOG.md"
 
 [[tool.mypy.overrides]]
 module = ["RPi", "RPi.GPIO", "serial"]
 ignore_missing_imports = true
+
+[tool.bandit]
+exclude_dirs = ["tests", ".venv"]

src/pywaveshare/boards/sim868/__init__.py

@@ -320,7 +320,7 @@ def __processData(self):
                     rawData = match[0][1].split(",")
                     self.__GPRSIPaddress = rawData[2].replace('"', "")
 
-                    if self.__GPRSIPaddress != "0.0.0.0":
+                    if self.__GPRSIPaddress != "0.0.0.0":  # nosec
                         self.__GPRSready = True
                     else:
                         self.__GPRSready = False

src/pywaveshare/boards/sim868/client.py

@@ -1,8 +1,16 @@
-from . import config, worker
+from . import config, itc, worker
 
 
 class Client:
+    logger = config.get_logger("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/config.py

@@ -1,3 +1,72 @@
+import json
+import logging
+import logging.config
+import time
+
+LOGGING_CONFIG = {
+    "version": 1,
+    "formatters": {
+        "txt": {
+            "format": "[%(asctime)s.%(msecs)03d][%(levelname)s]: %(message)s",
+            "datefmt": "%Y-%m-%dT%H:%M:%S",
+        },
+        "jsonl": {
+            "format": json.dumps(
+                {
+                    "timestamp": "%(asctime)s.%(msecs)03d",
+                    "process": "%(process)d",
+                    "thread": "%(thread)d",
+                    "logger": "%(name)s",
+                    "level": "%(levelname)s",
+                    "source": "%(pathname)s:%(lineno)d",
+                    "message": "%(message)s",
+                }
+            ),
+            "datefmt": "%Y-%m-%dT%H:%M:%S",
+        },
+    },
+    "filters": {},
+    "handlers": {
+        "file": {
+            "class": "logging.FileHandler",
+            "filename": "event.log",
+            "formatter": "jsonl",
+            "level": "DEBUG",
+            "filters": [],
+        },
+        "stream": {
+            "class": "logging.StreamHandler",
+            "formatter": "txt",
+            "level": "DEBUG",
+            "filters": [],
+        },
+    },
+    "loggers": {
+        "client": {
+            "handlers": ["file", "stream"],
+            "level": "DEBUG",
+            "propagate": True,
+        },
+        "worker": {
+            "handlers": ["file", "stream"],
+            "level": "DEBUG",
+            "propagate": True,
+        },
+        "protocol": {
+            "handlers": ["file", "stream"],
+            "level": "DEBUG",
+            "propagate": True,
+        },
+    },
+}
+logging.Formatter.converter = time.gmtime
+logging.config.dictConfig(LOGGING_CONFIG)
+
+
+def get_logger(id: str) -> logging.Logger:
+    return logging.getLogger(id)
+
+
 class Config:
     def __init__(self, serial_port: str, baud_rate: int, encoding: str) -> None:
         self.serial_port = serial_port

src/pywaveshare/boards/sim868/itc.py

@@ -1,9 +1,16 @@
+from __future__ import annotations
+
 import queue
 import threading
 
 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

@@ -1,5 +1,31 @@
 import abc
+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,15 +1,17 @@
 import threading
 
-from . import config, itc
+from . import config, itc, protocols
 
 
 class Worker(threading.Thread):
+    logger = config.get_logger("worker")
+
     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