Skip to content
Snippets Groups Projects
Select Git revision
  • 7b83da0f5a082e096c4afeb389838ea761460e3b
  • master default protected
  • develop
  • semantic
  • auto-inflate
  • feature/aps
  • legacy protected
  • 10.0.0
  • 9.3.8
  • 9.3.7
  • 9.3.6
  • 9.3.5
  • 9.3.4
  • 9.3.3
  • 9.3.2
  • 9.3.1
  • 9.3.0
  • 9.2.0
  • 9.1.2
  • 9.1.1
  • 9.1.0
  • 9.0.1
  • 9.0.0
  • 8.2.5
  • 8.2.4
  • 8.2.3
  • 8.2.2
27 results

setup.py

Blame
  • performance_test.py 828 B
    import src.mqtt as mqtt
    import json
    from datetime import datetime
    import random
    import time
    
    MQTT_USER = "test"
    MQTT_PASSWORD = "test"
    
    MQTT_BROKER = "localhost"#"134.130.225.37"#
    MQTT_PORT = 1883
    MQTT_VHOST = "/"
    
    client = mqtt.client.MQTTPublisher(MQTT_USER)
    client.connect(MQTT_BROKER, MQTT_PORT, MQTT_VHOST + ":"+  MQTT_USER, MQTT_PASSWORD)
    test=0
    i=0
    while True:
        i+=1
        test += random.random()-0.5
        client.publish(
            topic="quatsch"+"/OBJ-Environment/OBJ-0/VAR-TEST",
                message=json.dumps({
                    "value": [float(test)],
                    "timestamp": datetime.utcnow().isoformat()+"Z",
                    "covariance": None,
                    "nonce": {},
                    "hash": None,
                    "unit": "P1"
                }),
                qos=0,
                )
        print(i)
        time.sleep(1)