warpapi.py

from flask import Flask
from flask import request
from flask import redirect
from flask import render_template
from flask import send_from_directory
from flask import json
from flask_mqtt import Mqtt
from logging.config import dictConfig

dictConfig({
    'version': 1,
    'formatters': {'default': {
        'format': '[%(asctime)s] %(levelname)s in %(module)s: %(message)s',
    }},
    'handlers': {'wsgi': {
        'class': 'logging.StreamHandler',
        'stream': 'ext://flask.logging.wsgi_errors_stream',
        'formatter': 'default'
    }},
    'root': {
        'level': 'INFO',
        'handlers': ['wsgi']
    }
})

app = Flask(__name__, static_url_path='')
app.config['MQTT_BROKER_URL'] = '192.168.0.201'  
app.config['MQTT_BROKER_PORT'] = 1883 
app.config['MQTT_USERNAME'] = ''  
app.config['MQTT_PASSWORD'] = ''  
app.config['MQTT_KEEPALIVE'] = 5  
app.config['MQTT_TLS_ENABLED'] = False  
zone_door_status = "UNKNOWN"
mqtt = Mqtt(app)

if __name__ == "__main__":
    app.run()

@mqtt.on_connect()
def handle_connect(client, userdata, flags, rc):
    app.logger.info("mqtt connected")
    mqtt.subscribe("warpzone/door/status")
    app.logger.info("mqtt subscribed: warpzone/door/status")

@mqtt.on_message()
def handle_mqtt_message(client, userdata, message):
    global zone_door_status
    data = dict(
        topic=message.topic,
        payload=message.payload.decode()
    )
    app.logger.info("mqtt message")
    app.logger.info(data)
    zone_door_status = message.payload.decode()
    app.logger.info("zone_door_status = " + zone_door_status)

# CORS Header setzen     
@app.after_request 
def after_request(response):
    header = response.headers
    header['Access-Control-Allow-Origin'] = '*'
    return response

# Startseite
@app.route('/')
def view_index():
    global zone_door_status   
    return 'Warpzone API, current status = ' + zone_door_status + ', see: https://gitlab.warpzone.ms/infrastruktur/warpapi/blob/master/README.md'

# Statische Dateien 
@app.route('/files/<path:path>')
def send_file(path):
    return send_from_directory('files', path)

# Statusanzeige auf der Webseite 
@app.route('/static')
def view_static():
    global zone_door_status   
    spaceopen = False
    if zone_door_status == "OPEN":
      spaceopen = True
    return render_template('static.html', open = spaceopen)

# Statusanzeige auf der Webseite 
@app.route('/status')
def data_status():
    global zone_door_status   
    spaceopen = 0
    if zone_door_status == "OPEN":
      spaceopen = 1
    data = {
        "tuerOffen": spaceopen,
        "tempLounge": 250,
        "tempWerkstatt": 250,
        "timestamp": 0,
        "age": 500
    } 
    response = app.response_class(
        response=json.dumps(data),
        status=200,
        mimetype='application/json'
    )
    return response

# Endpunkt für Einbindung in cpu.ccc.de 
# Eigentlich die SpaceAPI, aber so können einfacher Abweichungen implementiert werden
@app.route('/cccapi')
def data_cccapi():
    return redirect('/spaceapi', code=302)

# Kompatibilität für alte urls 
@app.route('/spaceapi.php')
def data_spaceapiphp():
    return redirect('/spaceapi', code=302)

# Implementierung der SpaceAPI
@app.route('/spaceapi')
def data_spaceapi():
    global zone_door_status   
    spaceopen = False
    if zone_door_status == "OPEN":
      spaceopen = True
    data = {
        "api": "0.14",
        "space": "warpzone ", 
        "logo": "https://api.warpzone.ms/files/warpzone_logo.png",
        "url": "https://www.warpzone.ms",
        "location": {
            "address": "warpzone e.V., Am Hawerkamp 31 Maus G, 48155 Münster, Germany",
            "lat": 51.9436176,
            "lon": 7.6381682,
        },
        "contact": {
            "phone": "+49 251 39500271",
            "twitter": "@warpzone_ms",
            "mastodon": "@warpzone@social.bau-ha.us",
            "email": "verwaltung@warpzone.ms",
            "irc": "irc://hackint.org/#warpzone",
            "ml": "warpzone@warpzone.ms",
            "matrix": "#warpzone:matrix.warpzone.ms"
        },
        "issue_report_channels": [ 
            "email" 
        ],
        "feeds": {
            "blog": { 
                "type": "application/rss+xml",
                "url": "https://www.warpzone.ms/feed/",
            },
            "calendar": {
                "type": "application/calendar",
                "url": "https://www.google.com/calendar/ical/hc82sdjshuf1ru9f9eqcfq9bvs%40group.calendar.google.com/public/basic.ics",
            },
        },
        "state": {
            "open": spaceopen,
            "icon": {
                "open": "https://api.warpzone.ms/files/warpzone_open.png",
                "closed": "https://api.warpzone.ms/files/warpzone_closed.png",
            },
        },
    }
    
    response = app.response_class(
        response=json.dumps(data),
        status=200,
        mimetype='application/json'
    )
    return response