Prevented muliprocessing

.drone.yml

@@ -11,14 +11,6 @@ steps:
       sonar_token:
         from_secret: SONAR_CODE
 
-  - name: sentry
-    image: tormachris/drone-sentry
-    settings:
-      sentry_project: ${DRONE_REPO_NAME}
-      sentry_domain: sentry.kmlabz.com
-      sentry_token:
-        from_secret: SENTRY_TOKEN
-
   - name: ms-teams
     image: kuperiu/drone-teams
     settings:

requirements.txt

@@ -5,20 +5,25 @@ cython
 
 paho-mqtt
 
-cycler==0.10.0
-deprecation==2.0.7
+six
+deprecation
+
+cycler~=0.10.0
+
 eyeD3==0.9.5
 filetype==1.0.6
 hmmlearn==0.2.3
-joblib==0.14.1
-kiwisolver==1.2.0
-matplotlib==3.2.1
-numpy==1.18.2
-pyAudioAnalysis==0.3.0
+joblib~=1.0.1
+kiwisolver~=1.2.0
+matplotlib~=3.3.3
+numpy~=1.20.3
 pydub==0.23.1
 pyparsing==2.4.6
 python-dateutil==2.8.1
-scikit-learn==0.21.3
-scipy==1.4.1
-simplejson==3.17.0
-six==1.14.0
+scikit-learn~=0.24.0
+scipy~=1.6.2
+simplejson~=3.17.2
+
+
+pyAudioAnalysis~=0.3.0
+tqdm~=4.61.1

src/actuator/loudspeaker.py

@@ -1,9 +1,6 @@
 #!/usr/bin/env python3
-import random
-import os
-import os.path
 from .abcactuator import AbcActuator
-from birbnetes_iot_platform_raspberry import BirbnetesIoTPlatformPlaybackDriver, BirbnetesIoTPlatformStatusDriver
+from utils import BirbnetesIoTPlatformPlaybackDriver, BirbnetesIoTPlatformStatusDriver
 
 """
 Abstract base class for Sender

src/app.py

@@ -1,13 +1,16 @@
 #!/usr/bin/env python3
+from utils import config
+# config may modify imports
 import logging
 import sentry_sdk
 from sentry_sdk.integrations.logging import LoggingIntegration
-from utils import config, LoopingTimer
+from utils import LoopingTimer
 from signal_processor import SoundSignalProcessor
-from birbnetes_iot_platform_raspberry import BirbnetesIoTPlatformStatusDriver
+from utils import BirbnetesIoTPlatformStatusDriver, BirbnetesIoTPlatformRecordDriver
 from actuator import Loudspeaker
 import paho.mqtt.client
 import json
+import requests
 import sys
 
 """
@@ -28,7 +31,7 @@ if config.SENTRY_DSN:
     sentry_sdk.init(
         dsn=config.SENTRY_DSN,
         integrations=[sentry_logging],
-        traces_sample_rate=1.0,
+        traces_sample_rate=0.0,
         send_default_pii=True,
         release=config.RELEASE_ID,
         environment=config.RELEASEMODE,
@@ -71,6 +74,23 @@ def mqtt_on_command(client, userdata, message):
         userdata.act()
 
 
+def do_report():
+    report = {
+        "client": config.DEVICE_ID,
+        "measurements": {
+            "queue": BirbnetesIoTPlatformRecordDriver.get_queue_length()
+        }
+    }
+
+    print("Reporting queue length of", report)
+
+    r = requests.post(config.REPORT_URL, json=report)
+
+    r.raise_for_status()
+    if r.status_code != 201:
+        print(config.REPORT_URL, "Wrong response:", r.status_code)
+
+
 def main() -> None:
     """
     Main function
@@ -79,10 +99,16 @@ def main() -> None:
     logging.basicConfig(stream=sys.stdout, format="%(asctime)s - %(name)s [%(levelname)s]: %(message)s",
                         level=logging.DEBUG if '--debug' in sys.argv else logging.INFO)
     BirbnetesIoTPlatformStatusDriver.init()
-    listofabcsignaprocessors = [SoundSignalProcessor()]
+    listofabcsignaprocessors = [SoundSignalProcessor()]  # <- figuring out of the url of filter/input service
     loopingtimer = LoopingTimer(function=timer_tick, tick_args=listofabcsignaprocessors, interval=config.TICK_INTERVAL)
     loopingtimer.start()
 
+    report_timer = None
+    if config.REPORT_URL:
+        print("Setting up queue length reporting...")
+        report_timer = LoopingTimer(function=do_report, tick_args=None, interval=config.REPORT_INTERVAL)
+        report_timer.start()
+
     client = paho.mqtt.client.Client(userdata=Loudspeaker(config.ENEMY_SOUNDS), client_id=config.DEVICE_ID)
     client.on_connect = mqtt_on_connect
     client.on_disconnect = mqtt_on_disconnect
@@ -98,6 +124,9 @@ def main() -> None:
     except KeyboardInterrupt:
         logging.info("SIGINT recieved! Stopping...")
 
+    if report_timer:
+        report_timer.stop()
+
     client.disconnect()
     loopingtimer.stop()
     BirbnetesIoTPlatformStatusDriver.cleanup()

src/preprocessor/soundpreprocessor.py

@@ -1,18 +1,21 @@
 #!/usr/bin/env python3
-import requests
-from urllib.parse import urljoin
-from pyAudioAnalysis.audioTrainTest import load_model, load_model_knn, classifier_wrapper
 from utils import config
 from .abcpreprocessor import AbcPreProcessor
-import tempfile
-import os
+
 import logging
 
-from pyAudioAnalysis import audioBasicIO
-from pyAudioAnalysis import MidTermFeatures
-import numpy
+if not config.DISABLE_AI:
+    import tempfile
+    import requests
+    from urllib.parse import urljoin
+    import os
 
-from birbnetes_iot_platform_raspberry import BirbnetesIoTPlatformStatusDriver
+    from pyAudioAnalysis.audioTrainTest import load_model, load_model_knn, classifier_wrapper
+    from pyAudioAnalysis import audioBasicIO
+    from pyAudioAnalysis import MidTermFeatures
+    import numpy
+
+from utils import BirbnetesIoTPlatformStatusDriver
 
 """
 Abstract base class for Sender
@@ -24,7 +27,7 @@ __module_name__ = "soundpreprocessor"
 __version__text__ = "1"
 
 
-class SoundPreProcessor(AbcPreProcessor):
+class SoundPreProcessorLegit(AbcPreProcessor):
     """
     SoundPreProcessor class, responsible for detecting birb chirps in sound sample.
     """
@@ -142,3 +145,18 @@ class SoundPreProcessor(AbcPreProcessor):
             os.remove(self._temp_means_name)
         except FileNotFoundError:
             pass
+
+
+class SoundPreProcessorDummy(AbcPreProcessor):
+
+    def __init__(self):
+        logging.info("AI is disabled! Initializing dummy sound pre-processor...")
+
+    def preprocesssignal(self, file_path) -> bool:
+        return True
+
+
+if config.DISABLE_AI:
+    SoundPreProcessor = SoundPreProcessorDummy
+else:
+    SoundPreProcessor = SoundPreProcessorLegit

src/sender/soundsender.py

@@ -5,7 +5,10 @@ import logging
 from datetime import datetime
 import requests
 from .abcsender import AbcSender
-from utils import config
+from utils import config, LoopingTimer
+from urllib.parse import urljoin
+import time
+from threading import Lock
 
 """
 Send a sound sample
@@ -22,21 +25,118 @@ class SoundSender(AbcSender):
     SoundSender class, responsible for sending sound samples to the cloud.
     """
 
+    super_global_multi_sender_preventer = Lock()
+
+    def __init__(self):
+
+        self._feed_url = None
+        self._reassign_timer = None
+        if not config.NO_AUTODISCOVER:
+            retry = 0
+            timeout = 30
+            while True:
+                try:
+                    r = requests.post(
+                        urljoin(config.API_URL, "/assignment"),
+                        json={"device_id": config.DEVICE_ID},
+                        timeout=timeout
+                    )
+                except requests.exceptions.Timeout:
+                    logging.warning(f"/assignment timed out after {timeout} sec! Retrying...")
+                    continue
+
+                if r.status_code not in [200, 404]:
+                    logging.warning(
+                        f"Unexpected status code while acquiring an assignment: {r.status_code}. Retrying..."
+                    )
+                    time.sleep(5)
+                    continue
+
+                if r.status_code == 404:
+                    logging.info(
+                        "/assignment returned 404 error. Assuming missing capability... Defaulting to use API_URL"
+                    )
+                    self._feed_url = config.API_URL
+                    break
+
+                if r.json()['hard_default']:
+                    retry += 1
+                    if retry < 10:
+                        logging.info(f"Still waiting for non hard-default url... Attempt {retry}/10")
+                        time.sleep(6)
+                        continue
+                    else:
+                        logging.info(f"Given up waiting for non hard-default url... accepting hard default.")
+
+                logging.info(f"Assigned to {r.json()['site']}")
+                self._feed_url = r.json()['url']
+                break
+
+            self._reassign_timer = LoopingTimer(config.REASSIGN_INTERVAL, lambda: self.reassign())
+            self._reassign_timer.start()
+
+        # If the above block failed to assign a feed_url
+        if not self._feed_url:
+            logging.info("Using API_URL as feed url")
+            self._feed_url = config.API_URL
+
+    def reassign(self):
+        logging.debug("Asking for reassign...")
+        timeout = 30
+        try:
+            r = requests.post(
+                urljoin(config.API_URL, "/assignment"),
+                json={"device_id": config.DEVICE_ID},
+                timeout=timeout
+            )
+        except requests.exceptions.Timeout:
+            logging.warning(f"/assignment timed out after {timeout} sec! Ignoring...")
+            return
+
+        if r.status_code == 404:
+            logging.debug("/assignment returned 404 status code. Assuming missing capability, and doing nothing...")
+            return
+
+        if r.status_code != 200:
+            logging.debug(f"/assignment returned {r.status_code} status code. Ignoring...")
+            return
+
+        logging.debug(f"Received assignment to {r.json()['site']}")
+
+        if r.json()['url'] != self._feed_url:
+            logging.info(f"Reassigned to {r.json()['site']}!")
+            # Because of GIL and because this is a reference, we don't really have to be careful assigning this
+            self._feed_url = r.json()['url']
+
     def sendvalue(self, value: str, decision: bool) -> None:
         """
         Send a sound sample to the cloud.
+        value: is the file name
+        decision: if true the sample is sent, if false then not
         :return:
         """
         if decision:
             files = {
-                "file": (os.path.basename(value), open(value, 'rb').read(), 'audio/wave',
-                         {'Content-length': os.path.getsize(value)}),
+                "file": (
+                    os.path.basename(value),
+                    open(value, 'rb').read(),
+                    'audio/wave',
+                    {'Content-length': os.path.getsize(value)}
+                ),
                 "description": (
-                    None, json.dumps({'date': datetime.now().isoformat(), 'device_id': config.DEVICE_ID}),
-                    "application/json")
+                    None,
+                    json.dumps({'date': datetime.now().isoformat(), 'device_id': config.DEVICE_ID}),
+                    "application/json"
+                )
             }
 
-            r = requests.post(config.API_URL + "/sample", files=files)
+            with SoundSender.super_global_multi_sender_preventer:
+                r = requests.post(urljoin(self._feed_url, config.FEED_TYPE), files=files)
+
             logging.debug(f"Content: {r.content.decode()}")
             logging.debug(f"Headers: {r.headers}")
             r.raise_for_status()
+
+    def __del__(self):
+        if self._reassign_timer:
+            self._reassign_timer.stop()

src/sensor/soundsensor.py

@@ -1,8 +1,7 @@
 #!/usr/bin/env python3
 from typing import Optional
 from .abcsensor import AbcSensor
-from utils import config
-from birbnetes_iot_platform_raspberry import BirbnetesIoTPlatformRecordDriver
+from utils import config, BirbnetesIoTPlatformRecordDriver
 
 """
 Sound sensor high level API

src/signal_processor/soundsignalprocessor.py

@@ -7,7 +7,9 @@ from preprocessor import SoundPreProcessor
 from .abcsignalprocessor import AbcSignalProcessor
 import os
 
-from birbnetes_iot_platform_raspberry import BirbnetesIoTPlatformStatusDriver
+from utils import BirbnetesIoTPlatformStatusDriver
+
+from threading import Lock
 
 """
 Abstract base class for signalprocessor
@@ -24,6 +26,8 @@ class SoundSignalProcessor(AbcSignalProcessor):
     SoundSignalProcessor class, responsible for handling the sound signal processor pipeline.
     """
 
+    super_multi_signal_processing_preventor_to_make_queue_great_again = Lock()
+
     def __init__(self):
         """
         Create dependency objects.
@@ -37,28 +41,29 @@ class SoundSignalProcessor(AbcSignalProcessor):
         Process a sound sample.
         :return:
         """
-        soundsample_name = self.soundsensor.getvalue()
+        with SoundSignalProcessor.super_multi_signal_processing_preventor_to_make_queue_great_again:
+            soundsample_name = self.soundsensor.getvalue()
 
-        if not soundsample_name:  # No new sample... nothing to do
-            return
+            if not soundsample_name:  # No new sample... nothing to do
+                return
 
-        try:
-            sample_decision = self.soundpreprocessor.preprocesssignal(soundsample_name)
-        except Exception as e:
-            logging.exception(e)
-            BirbnetesIoTPlatformStatusDriver.enqueue_pattern('red', [1, 1, 1])
-            os.unlink(soundsample_name)
-            return
+            try:
+                sample_decision = self.soundpreprocessor.preprocesssignal(soundsample_name)
+            except Exception as e:
+                logging.exception(e)
+                BirbnetesIoTPlatformStatusDriver.enqueue_pattern('red', [1, 1, 1])
+                os.unlink(soundsample_name)
+                return
 
-        if sample_decision:
-            BirbnetesIoTPlatformStatusDriver.enqueue_pattern('green', [1, 0, 1])
+            if sample_decision:
+                BirbnetesIoTPlatformStatusDriver.enqueue_pattern('green', [1, 0, 1])
 
-        try:
-            self.soundsender.sendvalue(soundsample_name, sample_decision)
-        except (ConnectionError, requests.HTTPError) as e:
-            logging.exception(e)
-            BirbnetesIoTPlatformStatusDriver.enqueue_pattern('red', [1, 0, 1, 0, 1])
-            os.unlink(soundsample_name)
-            return
+            try:
+                self.soundsender.sendvalue(soundsample_name, sample_decision)
+            except (ConnectionError, requests.HTTPError) as e:
+                logging.exception(e)
+                BirbnetesIoTPlatformStatusDriver.enqueue_pattern('red', [1, 0, 1, 0, 1])
+                os.unlink(soundsample_name)
+                return
 
         os.unlink(soundsample_name)

src/utils/__init__.py

@@ -1,2 +1,4 @@
 #!/usr/bin/env python3
 from .loopingtimer import LoopingTimer
+from .platform import BirbnetesIoTPlatformStatusDriver, BirbnetesIoTPlatformRecordDriver, \
+        BirbnetesIoTPlatformPlaybackDriver

src/utils/config.py

@@ -29,4 +29,20 @@ MQTT_PASSWORD = os.getenv("GUARD_MQTT_PASSWORD", None)
 
 SVM_MODEL_ID = os.environ.get("SVM_MODEL_ID")
 
-API_URL = os.environ.get("API_URL", "http://localhost:8080")
+API_URL = os.environ.get("API_URL", "http://localhost:8080")
+REPORT_URL = os.environ.get("REPORT_URL", None)
+REPORT_INTERVAL = float(os.environ.get("REPORT_INTERVAL", 15))
+
+DISABLE_AI = os.environ.get("DISABLE_AI", 'no').lower() in ['yes', '1', 'true']
+
+PLATFORM = os.environ.get("PLATFORM", "raspberry")
+
+FEED_TYPE = os.environ.get("FEED_TYPE", "input")  # probably the worst naming the type of the accepting service.
+
+if FEED_TYPE not in ['input', 'filter']:
+    raise ValueError("FEED_TYPE must be either input or filter")
+
+# Skip fetching /assignment endpoint
+NO_AUTODISCOVER = os.environ.get("NO_AUTODISCOVER", 'no').lower() in ['yes', '1', 'true']
+
+REASSIGN_INTERVAL = int(os.environ.get("REASSIGN_INTERVAL", 120))

src/utils/platform.py

@@ -0,0 +1,20 @@
+from .config import PLATFORM
+
+print("Loading platform driver:", PLATFORM)
+
+if PLATFORM == 'raspberry':
+    from birbnetes_iot_platform_raspberry import BirbnetesIoTPlatformStatusDriver, BirbnetesIoTPlatformRecordDriver, \
+        BirbnetesIoTPlatformPlaybackDriver
+
+elif PLATFORM == 'emulator':
+    from birbnetes_iot_platform_emulator import BirbnetesIoTPlatformStatusDriver, BirbnetesIoTPlatformRecordDriver, \
+        BirbnetesIoTPlatformPlaybackDriver
+
+else:
+    raise ModuleNotFoundError(f"Could not load platform driver for {PLATFORM}")
+
+__all__ = [
+    'BirbnetesIoTPlatformStatusDriver',
+    'BirbnetesIoTPlatformRecordDriver',
+    'BirbnetesIoTPlatformPlaybackDriver'
+]