Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- MANUAL_MODE_NAME = "manual"
- AUTO_MODE_NAME = "auto"
- def check_prometheus_ready():
- try:
- r = requests.get(settings.PROMETHEUS_URL)
- r.raise_for_status()
- except requests.HTTPError:
- logging.error("Got HTTP error, probably Prometheus is not yet ready, throwing exception and exiting...")
- raise
- logging.info("Prometheus responded with HTTP OK, probably ready to serve requests")
- def get_prometheus_value(query): # TODO: copied form mqtt-importer-prometheus, probably should be extracted to separate module
- r = requests.get("{}/api/v1/query?query={}".format(settings.PROMETHEUS_URL, query))
- if r.status_code == requests.codes.ok:
- try:
- timestamp, query_result = r.json()['data']['result'][0]['value']
- logging.info("Got values from Prometheus: [{}] {}".format(timestamp, query_result))
- except (KeyError, ValueError):
- logging.error("Got unexpected response from Prometheus (HTTP {}): {}".format(r.status_code, r.text))
- return
- except IndexError:
- logging.error("There are no values for query {}".format(query))
- return
- else:
- logging.error("HTTP error {} when querying Prometheus".format(r.status_code))
- return
- return query_result
- @app.route('/')
- def hello():
- try:
- current_value_low = get_prometheus_value(settings.LOW_TEMP_METRIC)
- current_value_high = get_prometheus_value(settings.HIGH_TEMP_METRIC)
- current_value_mode = int(get_prometheus_value(settings.MODE_METRIC))
- current_engine_mode = int(get_prometheus_value(settings.ENGINE_MODE_METRIC))
- except Exception: # TODO: catch more meaningful exception, not all
- return "ERROR: Prometheus returned unexpected value!"
- # TODO: Move next lines to a template (i zjebana ifologia xD, ogolnie to jest chujowo napisane w huj xD)
- return f"""
- <form action="/set_params" method="GET">
- <div>
- <label for="value_low">Low</label>
- <input name="value_low" id="value_low" value="{current_value_low}">
- </div>
- <div>
- <label for="value_high">High</label>
- <input name="value_high" id="value_high" value="{current_value_high}">
- </div>
- <div>
- <label for="mode">Mode</label>
- <select name="mode">
- <option value="manual" {'selected' if current_value_mode == 0 else ''}>Manual</option>
- <option value="auto" {'selected' if current_value_mode == 1 else ''}>Auto</option>
- </select>
- </div>
- <div>
- <label for="engine_mode">Engine mode</label>
- <select name="engine_mode">
- <option value="0" {'selected' if current_engine_mode == 0 else ''}>Closed</pption>
- <option value="1" {'selected' if current_engine_mode == 1 else ''}>Open</option>
- </select>
- </div>
- <div>
- <button>Send</button>
- </div>
- </form>
- """
- def set_mode(mode):
- if mode not in (MANUAL_MODE_NAME, AUTO_MODE_NAME):
- return f"Mode must be '{MANUAL_MODE_NAME}' or '{AUTO_MODE_NAME}'"
- automat = 0 if mode == MANUAL_MODE_NAME else 1
- publish.single(settings.MODE_TOPIC, automat, hostname=settings.MQTT_BROKER_HOST, port=settings.MQTT_BROKER_PORT)
- return
- def set_temp(value, topic):
- if not value:
- return "Temp value invalid"
- try:
- float(value)
- except ValueError:
- return "Value must be a number!"
- publish.single(topic, value, hostname=settings.MQTT_BROKER_HOST, port=settings.MQTT_BROKER_PORT)
- return
- def set_engine_mode(mode, desired_mode):
- error_message = "Engine mode must be 0 or 1!"
- try:
- desired_mode = int(desired_mode)
- except ValueError:
- return error_message
- if not desired_mode in (0, 1):
- return error_message
- if mode == MANUAL_MODE_NAME:
- publish.single(settings.ENGINE_MODE_TOPIC, desired_mode, hostname=settings.MQTT_BROKER_HOST, port=settings.MQTT_BROKER_PORT)
- @app.route('/set_params')
- def set_params():
- # TODO: refactor this function!
- mode_set = set_mode(request.args.get("mode").lower())
- if mode_set:
- return mode_set
- low_temp_set = set_temp(request.args.get("value_low"), settings.LOW_TEMP_PROMETHEUS)
- if low_temp_set:
- return low_temp_set
- high_temp_set = set_temp(request.args.get("value_high"), settings.HIGH_TEMP_PROMETHEUS)
- if high_temp_set:
- return high_temp_set
- engine_mode_set = set_engine_mode(request.args.get("mode").lower(), request.args.get("engine_mode"))
- if engine_mode_set:
- return engine_mode_set
- return "OK</br><a href='/'>Powrot</a>"
- if __name__ == "__main__":
- check_prometheus_ready()
- logging.info(f"Starting flask server on {settings.FLASK_IP}:{settings.FLASK_PORT}")
- app.run(host=settings.FLASK_IP, port=settings.FLASK_PORT)
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement