status update consumption done

src/app.py

@@ -8,6 +8,8 @@ from sentry_sdk.integrations.flask import FlaskIntegration
 from config import *
 from db import db
 from marshm import ma
+from mqtt_flask_instance import mqtt
+
 """
 Main Flask RESTful API
 """
@@ -28,11 +30,17 @@ if SENTRY_DSN:
 
 
 app = Flask(__name__)
+app.config['MQTT_BROKER_URL'] = MQTT_HOSTNAME
+app.config['MQTT_BROKER_PORT'] = MQTT_PORT
+app.config['MQTT_USERNAME'] = MQTT_USERNAME
+app.config['MQTT_PASSWORD'] = MQTT_PASSWORD
+app.config['MQTT_REFRESH_TIME'] = 1.0  # refresh time in seconds
 app.config['SQLALCHEMY_DATABASE_URI'] = f"postgresql://{POSTGRES_USERNAME}:{POSTGRES_PASSWORD}@{POSTGRES_HOSTNAME}:5432/{POSTGRES_DB}"
 
 api = Api(app)
 db.init_app(app)
 ma.init_app(app)
+mqtt.init_app(app)
 
 with app.app_context():
     db.create_all()
@@ -48,6 +56,17 @@ logger = logging.getLogger(__name__)
 logger.setLevel(logging.DEBUG)
 logger.addHandler(handler)
 
+
+@mqtt.on_log()
+def handle_logging(client, userdata, level, buf):
+    logger.log(level, buf)
+
+
+@mqtt.on_connect()
+def handle_connect(client, userdata, flags, rc):
+    mqtt.subscribe(MQTT_STATUS_TOPIC)
+
+
 # api.add_resource(SampleResource, "/sample")
 # api.add_resource(SampleParameterResource, '/sample/<tag>')
 
@@ -55,5 +74,6 @@ if __name__ == "__main__":
     app.run(
         debug=bool(DEBUG),
         host="0.0.0.0",
+        use_reloader=False,
         port=int(PORT),
-    )
+    )

src/config.py

@@ -16,7 +16,6 @@ __version__text__ = "1"
 PORT = os.environ.get("CNC_SERVICE_PORT", 8080)
 DEBUG = os.environ.get("CNC_SERVICE_DEBUG", True)
 
-
 SENTRY_DSN = os.environ.get("SENTRY_DSN")
 RELEASE_ID = os.environ.get("RELEASE_ID", "test")
 RELEASEMODE = os.environ.get("CNC_SERVICE_RELEASEMODE", "dev")
@@ -25,3 +24,11 @@ POSTGRES_HOSTNAME = os.getenv("CNC_POSTGRES_HOSTNAME", "localhost")
 POSTGRES_USERNAME = os.getenv("CNC_POSTGRES_USERNAME", "cnc-service")
 POSTGRES_PASSWORD = os.getenv("CNC_POSTGRES_PASSWORD", "cnc-service")
 POSTGRES_DB = os.getenv("CNC_POSTGRES_DB", "cnc-service")
+
+MQTT_HOSTNAME = os.getenv("CNC_MQTT_HOSTNAME", "localhost")
+MQTT_PORT = os.getenv("CNC_MQTT_PORT", "1883")
+MQTT_USERNAME = os.getenv("CNC_MQTT_USERNAME", "guard-service")
+MQTT_PASSWORD = os.getenv("CNC_MQTT_PASSWORD", "guard-service")
+MQTT_STATUS_TOPIC = os.getenv("CNC_MQTT_STATUS_TOPIC", "guard-service")
+MQTT_COMMAND_TOPIC = os.getenv("CNC_MQTT_COMMAND_TOPIC", "guard-service")
+

src/models.py

@@ -0,0 +1,42 @@
+#!/usr/bin/env python3
+import enum
+from sqlalchemy.sql import func
+from db import db
+
+"""
+SQLAlchemy models
+"""
+
+__author__ = '@tormakris'
+__copyright__ = "Copyright 2020, Birbnetes Team"
+__module_name__ = "models"
+__version__text__ = "1"
+
+
+class StatusEnum(enum.Enum):
+    online = "online"
+    offline = "offline"
+
+
+class DeviceStatusEnum(StatusEnum):
+    error = "error"
+
+
+class SensorStatusEnum(StatusEnum):
+    unknown = "unknown"
+
+
+class Device(db.Model):
+    __tablename__ = 'device'
+    id = db.Column(db.UUID, primary_key=True)
+    status = db.Column(db.Enum(DeviceStatusEnum), nullable=False)
+    url = db.Column(db.String, nullable=False)
+    lastupdate = db.Column(db.TIMESTAMP, nullable=False, server_default=func.now(), onupdate=func.current_timestamp())
+    sensors = db.relationship("Sensor", nullable=True)
+
+
+class Sensor(db.Model):
+    __tablename__ = 'sensor'
+    id = db.Column(db.UUID, primary_key=True)
+    status = db.Column(db.Enum(SensorStatusEnum), nullable=False)
+    device_id = db.Column(db.UUID, db.ForeignKey('device.id'))

src/mqtt_flask_instance.py

@@ -0,0 +1,13 @@
+#!/usr/bin/env python3
+from flask_mqtt import Mqtt
+
+"""
+Flask-MQTT Api
+"""
+
+__author__ = '@tormakris'
+__copyright__ = "Copyright 2020, Birbnetes Team"
+__module_name__ = "db"
+__version__text__ = "1"
+
+mqtt = Mqtt()

src/mqtt_methods.py

@@ -0,0 +1,33 @@
+#!/usr/bin/env python3
+import logging
+from mqtt_flask_instance import mqtt
+from db import db
+from schemas import DeviceSchema
+
+"""
+MQTT Stuff
+"""
+
+__author__ = "@tormakris"
+__copyright__ = "Copyright 2020, Birbnetes Team"
+__module_name__ = "mqtt_methods"
+__version__text__ = "1"
+
+deviceschema = DeviceSchema(many=False)
+
+
+@mqtt.on_message()
+def handle_status_message(client, userdata, message):
+    data = dict(
+        topic=message.topic,
+        payload=message.payload.decode()
+    )
+    try:
+        status_message = deviceschema.load(data['payload'], session=db.session, transient=True).data
+        logging.info(f"Recieved status message from {data['payload']['id']}, persisting to db.")
+        db.session.merge(status_message)
+    except Exception as e:
+        db.session.rollback()
+        logging.exception(e)
+    else:
+        db.session.commit()

src/resources.py

@@ -0,0 +1,12 @@
+#!/usr/bin/env python3
+import logging
+
+
+"""
+Flask Restful endpoints
+"""
+
+__author__ = '@tormakris'
+__copyright__ = "Copyright 2020, Birbnetes Team"
+__module_name__ = "resources"
+__version__text__ = "1"

src/schemas.py

@@ -0,0 +1,35 @@
+#!/usr/bin/env python3
+import models
+from marshm import ma
+from marshmallow import fields
+
+"""
+Schemas of SQLAlchemy objects
+"""
+
+
+__author__ = "@tormakris"
+__copyright__ = "Copyright 2020, Birbnetes Team"
+__module_name__ = "schemas"
+__version__text__ = "1"
+
+
+class SensorSchema(ma.SQLAlchemyAutoSchema):
+    """
+    Sensor schema autogenerated
+    """
+    class Meta:
+        model = models.Sensor
+        include_fk = True
+
+
+class DeviceSchema(ma.SQLAlchemyAutoSchema):
+    """
+    Device schema autogenerated
+    """
+    sensors = fields.Nested(SensorSchema, many=True)
+
+    class Meta:
+        model = models.Device
+        include_relationships = True
+        exclude = ('lastupdate',)