File: spaceapi.py

package info (click to toggle)
bumblebee-status 2.2.0-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 2,844 kB
  • sloc: python: 13,430; sh: 68; makefile: 29
file content (148 lines) | stat: -rw-r--r-- 4,660 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
#!/usr/bin/env python
# -*- coding: utf-8 -*-

# pylint: disable=C0111,R0903

"""Displays the state of a Space API endpoint
Space API is an API for hackspaces based on JSON. See spaceapi.io for
an example.

Requires the following libraries:
    * requests

Parameters:
    * spaceapi.url: String representation of the api endpoint
    * spaceapi.format: Format string for the output

Format Strings:
    * Format strings are indicated by double %%
    * They represent a leaf in the JSON tree, layers separated by '.'
    * Boolean values can be overwritten by appending '%true%false'
      in the format string
    * Example: to reference 'open' in '{'state':{'open': true}}'
      you would write '%%state.open%%', if you also want
      to say 'Open/Closed' depending on the boolean you
      would write '%%state.open%Open%Closed%%'

contributed by `rad4day <https://github.com/rad4day>`_ - many thanks!
"""

import requests
import threading
import re
import json

import core.module
import core.widget
import core.input
import core.decorators


def formatStringBuilder(s, json):
    """
    Parses Format Strings
    Parameter:
        s -> format string
        json -> the spaceapi response object
    """
    identifiers = re.findall(r"%%.*?%%", s)
    for i in identifiers:
        ic = i[2:-2]  # Discard %%
        j = ic.split("%")

        # Only neither of, or both true AND false may be overwritten
        if len(j) != 3 and len(j) != 1:
            return "INVALID FORMAT STRING"

        if len(j) == 1:  # no overwrite
            s = s.replace(i, json[j[0]])
        elif json[j[0]]:  # overwrite for True
            s = s.replace(i, j[1])
        else:  # overwrite for False
            s = s.replace(i, j[2])
    return s


class Module(core.module.Module):
    @core.decorators.every(minutes=15)
    def __init__(self, config, theme):
        super().__init__(config, theme, core.widget.Widget(self.getState))

        core.input.register(self, button=core.input.LEFT_MOUSE, cmd=self.__forceReload)

        self.__data = {}
        self.__error = None
        self.__thread = None

        # The URL representing the api endpoint
        self.__url = self.parameter("url", default="http://club.entropia.de/spaceapi")
        self._format = self.parameter(
            "format", default=" %%space%%: %%state.open%Open%Closed%%"
        )

    def state(self, widget):
        try:
            if self.__error is not None:
                return ["critical"]
            elif self.__data["state.open"]:
                return ["warning"]
            else:
                return []
        except KeyError:
            return ["critical"]

    def update(self):
        if not self.__thread or self.__thread.is_alive() == False:
            self.__thread = threading.Thread(target=self.get_api_async, args=())
            self.__thread.start()

    def getState(self, widget):
        text = self._format
        if self.__error is not None:
            text = self.__error
        else:
            try:
                text = formatStringBuilder(self._format, self.__data)
            except KeyError:
                text = "KeyError"
        return text

    def get_api_async(self):
        try:
            with requests.get(self.__url, timeout=10) as request:
                # Can't implement error handling for python2.7 if I use
                # request.json() as it uses simplejson in newer versions
                self.__data = self.__flatten(json.loads(request.text))
                self.__error = None
        except requests.exceptions.Timeout:
            self.__error = "Timeout"
        except requests.exceptions.HTTPError:
            self.__error = "HTTP Error"
        except ValueError:
            self.__error = "Not a JSON response"
        core.event.trigger("update", [self.id], redraw_only=True)

    # left_mouse_button handler
    def __forceReload(self, event):
        if self.__thread:
            self.__thread.raise_exception()
        self.__error = "RELOADING"
        core.event.trigger("update", [self.id], redraw_only=True)

    # Flattens the JSON structure recursively, e.g. ['space']['open']
    # becomes ['space.open']
    def __flatten(self, json):
        out = {}
        for key in json:
            value = json[key]
            if type(value) is dict:
                flattened_key = self.__flatten(value)
                for fk in flattened_key:
                    out[key + "." + fk] = flattened_key[fk]
            else:
                out[key] = value
        return out


# Author: Tobias Manske <tobias@chaoswg.xyz>
# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4