benchmarks/benchmark.py
Torma 336a2c0a12
All checks were successful
continuous-integration/drone/push Build is passing
benchmark script done
2020-10-19 19:11:18 +02:00

111 lines
2.7 KiB
Python

#!/usr/bin/env python3
import logging
import os
import sys
import time
import json
import datetime
import sentry_sdk
from sentry_sdk.integrations.logging import LoggingIntegration
import paho.mqtt.client
import requests
"""
Main Entrypoint
"""
__author__ = "@tormakris"
__copyright__ = "Copyright 2020, Birbnetes Team"
__module_name__ = "app"
__version__text__ = "1"
SENTRY_DSN = os.environ.get("SENTRY_DSN")
RELEASE_ID = os.environ.get("RELEASE_ID", "test")
RELEASEMODE = os.environ.get("RELEASEMODE", "dev")
DEVICE_ID = os.environ.get("DEVICE_ID", "devraspi")
MQTT_USERNAME = ""
MQTT_PASSWORD = ""
MQTT_HOSTNAME = ""
MQTT_PORT = ""
STARTTIME = time.time()
ENDTIME = time.time()
URL = "https://birb.k8s.kmlabz.com/sample"
FILE = 'CommonStarling_100962.wav'
sentry_logging = LoggingIntegration(
level=logging.INFO,
event_level=logging.ERROR
)
if SENTRY_DSN:
sentry_sdk.init(
dsn=SENTRY_DSN,
integrations=[sentry_logging],
traces_sample_rate=1.0,
send_default_pii=True,
release=RELEASE_ID,
environment=RELEASEMODE,
_experiments={"auto_enabling_integrations": True}
)
def mqtt_on_connect(client, userdata, flags, rc):
client.subscribe(f"command/benchmark-script")
logging.info("Sending Message")
files = {
"file": (
os.path.basename(FILE), open(FILE, 'rb').read(), 'audio/wave', {'Content-length': os.path.getsize(FILE)}),
"description": (None, json.dumps({'date': datetime.now().isoformat(), 'device_id': '1'}), "application/json")
}
requests.post(URL, files=files)
nonlocal STARTTIME
STARTTIME = time.time()
def mqtt_on_command(client, userdata, message):
try:
msg = json.loads(message.payload.decode())
except (UnicodeError, json.JSONDecodeError) as e:
logging.error(f"MQTT Invalid message recieved: {e}")
return
if msg.get("command") == 'doAlert':
nonlocal ENDTIME
ENDTIME = time.time()
elapsed = ENDTIME - STARTTIME
logging.info(f"Elapsed time: {elapsed}")
def main() -> None:
"""
Main function
:return:
"""
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)
client = paho.mqtt.client.Client(client_id="benchmark-script")
client.on_connect = mqtt_on_connect
client.on_message = mqtt_on_command
if MQTT_USERNAME:
client.username_pw_set(MQTT_USERNAME, MQTT_PASSWORD)
client.connect(MQTT_HOSTNAME, MQTT_PORT, 60)
try:
client.loop_forever()
except KeyboardInterrupt:
logging.info("SIGINT recieved! Stopping...")
client.disconnect()
if __name__ == "__main__":
main()