C3L-NOC/spaceapi Changeset - 3e571d06ecc9 · Chaos Computer Club Lëtzebuerg Projects

Changeset - 3e571d06ecc9

Parent rev.

Child rev.

[Not reviewed]

default

0 6 0

Dennis Fink - 10 years ago 2015-07-11 22:23:21
dennis.fink@c3l.lu

Use singleton dictonary and reimplement reload view

6 files changed with 111 insertions and 32 deletions:

spaceapi/__init__.py

spaceapi/active.py

spaceapi/sensors.py

spaceapi/state.py

spaceapi/utils.py

spaceapi/views.py

0 comments (0 inline, 0 general)

spaceapi/__init__.py

➞

Show inline comments

@@ @@ -47,13 +47,16 @@ def create_app(): @@
         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 ActiveStatus
     ActiveStatus().reload()
     return app

spaceapi/active.py

➞

Show inline comments

 import copy
 import json
 import os.path
 from .utils import ActiveStatus
 default_json_file = os.path.abspath('default.json')
 last_state_file = os.path.abspath('laststate.json')
 if not os.path.exists(default_json_file):
     raise RuntimeError('default.json does not exists!')
 elif not os.path.isfile(default_json_file):
     raise RuntimeError('defaul.json is not a file!')
 default_json = {}
-active_json = {}
+active_json = ActiveStatus()
-def reload_json():
+def _reload_json():
     global default_json
     global active_json
     with open(default_json_file, encoding='utf-8') as f:
         default_json = json.load(f)
     if os.path.exists(last_state_file) and os.path.isfile(last_state_file):
         with open(last_state_file, encoding='utf-8') as f:
-            active_json = json.load(f)
+            active_json.update(json.load(f))
         if os.path.getmtime(last_state_file) \
            < os.path.getmtime(default_json_file):
             backup = copy.deepcopy(active_json)
             backup_state_open = active_json['state']['open']
             backup_state_lastchange = active_json['state']['lastchange']
             active_json.update(default_json)
             active_json['state']['open'] = backup['state']['open']
             active_json['state']['lastchange'] = backup['state']['lastchange']
             active_json['state']['open'] = backup_state_open
             active_json['state']['lastchange'] = backup_state_lastchange
     else:
-        active_json = copy.deepcopy(default_json)
+        active_json.update(default_json)
-reload_json()
+_reload_json()
 def save_last_state():
     with open(last_state_file, mode='w', encoding='utf-8') as f:
         json.dump(active_json, f, sort_keys=True)

spaceapi/sensors.py

➞

Show inline comments

 import json
 from functools import partial
 from pkg_resources import resource_filename
 import jsonschema
 from flask import Blueprint, jsonify, request, abort, current_app
 from .auth import httpauth
 from .active import active_json, save_last_state
 from .utils import first, fuzzy_list_find
 from .active import save_last_state
 from .utils import first, fuzzy_list_find, ActiveStatus
 sensors_views = Blueprint('sensors', __name__)
 ALLOWED_SENSORS_KEYS = json.load(
     open(resource_filename('spaceapi', 'schema/sensors.json'),
          encoding='utf-8')
+)
 RADIATON_SUBKEYS = frozenset(('alpha', 'beta', 'gamma', 'beta_gamma'))
 get_identification_key = partial(first, keys=frozenset(('name', 'location')))
 def set_value(data, key):
     active = ActiveStatus()
     try:
         subkey = get_identification_key(data)
     except ValueError:
         return abort(400)
     try:
-        index = fuzzy_list_find(active_json['sensors'][key],
+        index = fuzzy_list_find(active['sensors'][key],
                                 subkey, data[subkey])
-        active_json['sensors'][key][index].update(data)
+        active['sensors'][key][index].update(data)
     except ValueError:
-        active_json['sensors'][key].append(data)
+        active['sensors'][key].append(data)
     save_last_state()
     return jsonify(active_json)
     active.save_last_state()
     return jsonify(active)
 def set_radiation_value(data):
     active = ActiveStatus()
     radiation_keys = [k for k in RADIATON_SUBKEYS if k in data]
     if not radiation_keys:
         return abort(400)
     for first_subkey in radiation_keys:
         try:
             second_subkey = get_identification_key(data[first_subkey])
         except ValueError:
             return abort(400)
         try:
             index = fuzzy_list_find(
-                active_json['sensors']['radiation'][first_subkey],
+                active['sensors']['radiation'][first_subkey],
                 second_subkey,
                 data[first_subkey][second_subkey])
-            active_json['sensors']['radiation'][first_subkey][index].update(
+            active['sensors']['radiation'][first_subkey][index].update(
                 data)
         except ValueError:
-            active_json['sensors']['radiaton'][first_subkey].append(data)
+            active['sensors']['radiaton'][first_subkey].append(data)
     save_last_state()
     return jsonify(active_json)
     active.save_last_state()
     return jsonify(active)
 @sensors_views.route('/set/<key>', methods=['POST'])
 @httpauth.login_required
 def set_sensors(key):
     if key in ALLOWED_SENSORS_KEYS and key in active_json['sensors']:
     active = ActiveStatus()
     if key in ALLOWED_SENSORS_KEYS and key in active['sensors']:
         data = request.data.decode('utf-8')
         try:
             data = json.loads(data)
             try:
                 jsonschema.validate(data, ALLOWED_SENSORS_KEYS[key])
             except jsonschema.ValidationError:
                 current_app.logger.info('Validation Error')
                 return abort(400)
             if key != 'radiation':
                 return set_value(data, key)

spaceapi/state.py

➞

Show inline comments

 import json
 from time import time
 from flask import Blueprint, jsonify, request, current_app, abort
-from .active import active_json, save_last_state
+from .utils import ActiveStatus
 from .auth import httpauth
 state_views = Blueprint('state', __name__)
 ALLOWED_STATE_KEYS = frozenset(('open', 'lastchange',
                                 'trigger_person', 'message'))
 @state_views.route('/set/<key>', methods=['POST'])
 @httpauth.login_required
 def set_state(key):
     active = ActiveStatus()
     value = json.loads(request.data.decode('utf-8'))['value']
     current_app.logger.info(value)
     current_app.logger.info(type(value))
     if key in ALLOWED_STATE_KEYS:
-        active_json['state'][key] = value
+        active['state'][key] = value
         if key == 'open':
-            active_json['state']['lastchange'] = int(time())
+            active['state']['lastchange'] = int(time())
         save_last_state()
         return jsonify(active_json)
         active.save_last_state()
         return jsonify(active)
     else:
         return abort(400)

spaceapi/utils.py

➞

Show inline comments

 import json
 import os.path
 from flask import request
 default_json_file = os.path.abspath('default.json')
 last_state_file = os.path.abspath('laststate.json')
 if not os.path.exists(default_json_file):
     raise RuntimeError('default.json does not exists!')
 elif not os.path.isfile(default_json_file):
     raise RuntimeError('default.json is not a file!')
 _default_json = {}
 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 ActiveStatus(Singleton, dict):
     def reload(self):
         global _default_json
         with open(default_json_file, encoding='utf-8') as f:
             _default_json = json.load(f)
         if os.path.exists(last_state_file) and os.path.isfile(last_state_file):
             with open(last_state_file, encoding='utf-8') as f:
                 self.update(json.load(f))
             if os.path.getmtime(last_state_file) \
                < os.path.getmtime(default_json_file):
                 backup_open = self['state']['open']
                 backup_lastchange = self['state']['lastchange']
                 backup_sensors = self['sensors']
                 self.update(_default_json)
                 self['state']['open'] = backup_open
                 self['state']['lastchange'] = backup_lastchange
                 self['sensors'].update(backup_sensors)
         else:
             self.update(_default_json)
     def save_last_state(self):
         with open(last_state_file, mode='w', encoding='utf-8') as f:
             json.dump(self, f, sort_keys=True)
 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):

spaceapi/views.py

➞

Show inline comments

 from flask import Blueprint, jsonify, render_template, abort
 from .active import active_json
 from .utils import request_wants_json
 from .utils import request_wants_json, ActiveStatus
 from .auth import httpauth
 root_views = Blueprint('root', __name__)
 @root_views.route('/')
 def index():
     if request_wants_json():
         return jsonify(active_json)
     return render_template('index.html', status=active_json)
         return jsonify(ActiveStatus())
     return render_template('index.html', status=ActiveStatus())
 @root_views.route('/status.json')
 def status_json():
     return jsonify(active_json)
     return jsonify(ActiveStatus())
 @root_views.route('/reload')
 @httpauth.login_required
 def reload():
     active = ActiveStatus()
     active.reload()
     return jsonify(active)

0 comments (0 inline, 0 general)