Changeset - 047cc1b40374
[Not reviewed]
default
1 2 0
Dennis Fink - 3 years ago 2022-07-19 19:18:33
dennis.fink@c3l.lu
Remove state endpoint
3 files changed with 2 insertions and 35 deletions:
0 comments (0 inline, 0 general)
spaceapi/__init__.py
Show inline comments
 
@@ -46,61 +46,57 @@ def create_app():
 

	
 
    app.config.from_json(config_file, silent=True)
 
    app.config.setdefault("BOOTSTRAP_SERVE_LOCAL", True)
 

	
 
    bootstrap.init_app(app)
 

	
 
    app.logger.setLevel(logging.DEBUG)
 
    stream_handler = logging.StreamHandler()
 
    stream_handler.setLevel(logging.DEBUG)
 
    stream_handler.setFormatter(logging_debug_formatter)
 

	
 
    if not os.path.exists("spaceapi.log"):
 
        open("spaceapi.log", mode="a").close()
 

	
 
    rotating_file_handler = logging.handlers.RotatingFileHandler(
 
        "spaceapi.log", maxBytes=1300000, backupCount=10, encoding="utf-8"
 
    )
 
    rotating_file_handler.setLevel(logging.INFO)
 
    rotating_file_handler.setFormatter(logging_formatter)
 

	
 
    if app.debug or (
 
        "ENABLE_DEBUG_LOG" in app.config and app.config["ENABLE_DEBUG_LOG"]
 
    ):
 

	
 
        if not os.path.exists("spaceapi_debug.log"):
 
            open("spaceapi_debug.log", mode="a").close()
 

	
 
        second_rotating_file_handler = logging.handlers.RotatingFileHandler(
 
            "spaceapi_debug.log", maxBytes=1300000, backupCount=20, encoding="utf-8"
 
        )
 
        second_rotating_file_handler.setLevel(logging.DEBUG)
 
        second_rotating_file_handler.setFormatter(logging_debug_formatter)
 
        app.logger.addHandler(second_rotating_file_handler)
 

	
 
    app.logger.addHandler(stream_handler)
 
    app.logger.addHandler(rotating_file_handler)
 

	
 
    @app.after_request
 
    def add_headers(response):
 
        response.headers.setdefault("Access-Control-Allow-Origin", "*")
 
        response.headers.setdefault("Cache-Control", "no-cache")
 

	
 
        return response
 

	
 
    from .views import root_views
 

	
 
    app.register_blueprint(root_views)
 

	
 
    from .state import state_views
 

	
 
    app.register_blueprint(state_views, url_prefix="/state")
 

	
 
    from .sensors import sensors_views
 

	
 
    app.register_blueprint(sensors_views, url_prefix="/sensors")
 

	
 
    from .utils import ActiveStatusv14
 

	
 
    ActiveStatusv14().reload()
 

	
 
    return app
spaceapi/state.py
Show inline comments
 
deleted file
spaceapi/utils.py
Show inline comments
 
import email
 
import email.policy
 
import json
 
import os.path
 
import random
 
import smtplib
 
import ssl
 
from functools import partial, wraps
 
from functools import partial
 
from time import time
 

	
 
import mastodon
 
import tweepy
 
from flask import current_app, request
 

	
 
default_json_file_v14 = os.path.abspath("default_v14.json")
 
last_state_file_v14 = os.path.abspath("laststate_v14.json")
 

	
 
if not os.path.exists(default_json_file_v14):
 
    raise RuntimeError("default_v14.json does not exists!")
 
elif not os.path.isfile(default_json_file_v14):
 
    raise RuntimeError("default_v14.json is not a file!")
 

	
 
standard_open_message = "The space is now open!"
 
standard_close_message = "The space is now closed!"
 

	
 
possible_open_messages = (
 
    standard_open_message,
 
    "The space is open! Come in and hack something!",
 
    "Yes, we're open! Come in and create something!",
 
    "Come by and hack something! We've just opened!",
 
    "The ChaosStuff is now open for everyone!",
 
    "Let the Chaos begin! We're open!",
 
    "What do we hack now? Come and see, we've just opened!",
 
    "TUWAT! Come and hack. We are open!",
 
)
 

	
 
possible_closed_messages = (
 
    standard_close_message,
 
    "We're closed now! See you soon.",
 
    "Sorry, we are closed now!",
 
    "The ChaosStuff is now closed! Come back another time!",
 
    "Poweroff process finished! We're closed!",
 
    "Singularity reached! The space is closed!",
 
    "Dream of electric sheeps! We are closed!",
 
)
 

	
 
get_identification_key = partial(first, keys=frozenset(("name", "location")))
 

	
 
RADIATON_SUBKEYS = frozenset(("alpha", "beta", "gamma", "beta_gamma"))
 

	
 

	
 
class Singleton:
 
    def __new__(cls, *args, **kwargs):
 
        key = str(hash(cls))
 

	
 
        if not hasattr(cls, "_instance_dict"):
 
            cls._instance_dict = {}
 

	
 
        if key not in cls._instance_dict:
 
            cls._instance_dict[key] = super().__new__(cls, *args, **kwargs)
 

	
 
        return cls._instance_dict[key]
 

	
 

	
 
class ActiveStatusv14(Singleton, dict):
 
    def __init__(self):
 
        self.default_json_file = default_json_file_v14
 
        self.last_state_file = last_state_file_v14
 

	
 
    def reload(self):
 

	
 
        with open(self.default_json_file, encoding="utf-8") as f:
 
            self.update(json.load(f))
 

	
 
        if os.path.exists(self.last_state_file) and os.path.isfile(
 
            self.last_state_file
 
        ):
 
            with open(self.last_state_file, encoding="utf-8") as f:
 
                last_state_json = json.load(f)
 

	
 
            self["state"] = last_state_json["state"]
 
            self["sensors"].update(last_state_json["sensors"])
 

	
 
    def save_last_state(self):
 

	
 
        with open(self.last_state_file, mode="w", encoding="utf-8") as f:
 
            last_state = {}
 
            last_state["state"] = self["state"]
 
            last_state["sensors"] = self["sensors"]
 
            json.dump(last_state, f, sort_keys=True)
 

	
 
    def add_user_present(self, username):
 
        if self["state"]["open"]:
 
            if "people_now_present" not in self["sensors"]:
 
                self["sensors"]["people_now_present"] = [{"value": 0}]
 
@@ -246,57 +245,49 @@ class ActiveStatusv14(Singleton, dict):
 

	
 
    def set_radiation_sensor_value(self, data):
 
        radiation_keys = [k for k in RADIATON_SUBKEYS if k in data]
 
        if not radiation_keys:
 
            raise ValueErrr
 

	
 
        for first_subkey in radiation_keys:
 
            try:
 
                second_subkey = get_identification_key(data[first_subkey])
 
            except ValueError:
 
                raise
 

	
 
            try:
 
                index = fuzzy_list_find(
 
                    self["sensors"][first_subkey][second_subkey],
 
                    second_subkey,
 
                    data[first_subkey][second_subkey],
 
                )
 
                self["sensors"]["radiation"][first_subkey][index].update(data)
 
            except ValueError:
 
                self["sensors"]["radiation"][first_subkey].append(data)
 

	
 

	
 
def request_wants_json():
 
    best = request.accept_mimetypes.best_match(["application/json", "text/html"])
 
    return (
 
        best == "application/json"
 
        and request.accept_mimetypes[best] > request.accept_mimetypes["text/html"]
 
    )
 

	
 

	
 
def fuzzy_list_find(lst, key, value):
 

	
 
    for i, dic in enumerate(lst):
 
        if dic[key] == value:
 
            return i
 

	
 
    raise ValueError
 

	
 

	
 
def first(iterable, keys):
 
    for key in keys:
 
        if key in iterable:
 
            return key
 

	
 
    raise ValueError
 

	
 

	
 
def pass_active_status(f):
 
    @wraps(f)
 
    def wrapper(*args, **kwargs):
 
        status = ActiveStatusv14()
 
        rv = f(status, *args, **kwargs)
 
        status.save_last_state()
 
        return rv
 

	
 
    return wrapper
 
get_identification_key = partial(first, keys=frozenset(("name", "location")))
0 comments (0 inline, 0 general)