upy-thingserver/main.py at master · labthings/upy-thingserver · GitHub

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
import sys

sys.path.append("/webthing")
sys.path.append("/example")

import logging
import connect

logging.basicConfig(level=logging.DEBUG)
log = logging.getLogger(__name__)

connect.connect_to_ap()
connect.start_ntp()

from action import Action
from property import Property
from thing import Thing
from value import Value
from server import WebThingServer
import logging
import time
import uuid

log = logging.getLogger(__name__)


def make_thing():
    thing = Thing(
        "urn:dev:ops:my-lamp-1234",
        "My Lamp",
        ["OnOffSwitch", "Light"],
        "A web connected lamp",
    )

    on_property = Property(
        thing,
        "on",
        writeproperty=lambda v: print(v),
        initial_value=True,
        metadata={
            "@type": "OnOffProperty",
            "title": "On/Off",
            "type": "boolean",
            "description": "Whether the lamp is turned on",
        },
    )

    brightness_property = Property(
        thing,
        "brightness",
        writeproperty=lambda v: print(v),
        initial_value=50,
        metadata={
            "@type": "BrightnessProperty",
            "title": "Brightness",
            "type": "integer",
            "description": "The level of light from 0-100",
            "minimum": 0,
            "maximum": 100,
            "unit": "percent",
        },
    )

    thing.add_property(on_property)
    thing.add_property(brightness_property)

    def fade_function(args):
        time.sleep(args["duration"] / 1000)
        brightness_property.set_value(args["brightness"])

    fade_action = Action(
        thing,
        "fade",
        invokeaction=fade_function,
        metadata={
            "title": "Fade",
            "description": "Fade the lamp to a given level",
            "input": {
                "type": "object",
                "required": ["brightness", "duration",],
                "properties": {
                    "brightness": {
                        "type": "integer",
                        "minimum": 0,
                        "maximum": 100,
                        "unit": "percent",
                    },
                    "duration": {
                        "type": "integer",
                        "minimum": 1,
                        "unit": "milliseconds",
                    },
                },
            },
        },
    )

    thing.add_action(fade_action)

    return thing


def run_server():
    log.info("run_server")

    thing = make_thing()

    # If adding more than one thing, use MultipleThings() with a name.
    # In the single thing case, the thing's name will be broadcast.
    server = WebThingServer(thing)
    try:
        log.info("starting the server")
        server.start()
    except KeyboardInterrupt:
        log.info("stopping the server")
        server.stop()
        log.info("done")


if __name__ == "__main__":
    run_server()