-
Notifications
You must be signed in to change notification settings - Fork 2
/
Copy pathservice.py
131 lines (91 loc) · 3.99 KB
/
service.py
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
#!/usr/bin/env/python3
""" A Flask web service to accept remote control command requests (e.g. volume decrease, volume increase)
and instruct a transmitter (e.g. infrared) to transmit command
"""
from flask import Flask, jsonify, request
from remote_command import RemoteCommand
from ir_remote import transmit_command_ir
import scheduler
import service_constants
import logging_util
# instantiate at module level, not class level
# https://stackoverflow.com/questions/22807972/python-best-practice-in-terms-of-logging
logger = logging_util.get_logger(__name__)
# app is a flask object
app = Flask(__name__)
scheduler = scheduler.Scheduler()
def route(command):
"""
:param command: a RemoteCommand
:return: route string
"""
return "/api/v1/{}/{}/".format(service_constants.API_NAME, command.value)
def transmit_command(command):
"""
instruct infrared transmitter to transmit command
:parameter command: a RemoteCommand
:return: data dictionary with status 'success'
Note success indicates command was sent, not if any television received command
"""
# mac doesn't have an ir transmitter.
# Can run unit tests on macOS by temporarily commenting out call to transmit_command_ir
transmit_command_ir(command)
# f string requires Python >= 3.6, so don't use it yet.
# message = f'transmitted {command}'
message = 'transmitted {}'.format(command.value)
data = {service_constants.API_NAME_KEY: service_constants.API_NAME,
service_constants.VERSION_KEY: service_constants.VERSION,
service_constants.MESSAGE_KEY: message}
return jsonify(data)
# / is the website root, the entry point
# http://10.0.0.4:5000
# home http://127.0.0.1
# port :5000
@app.route('/')
@app.route("/api/v1/{}/ping/".format(service_constants.API_NAME), methods=['GET'])
def api_status():
if request.method == 'GET':
data = {service_constants.API_NAME_KEY: service_constants.API_NAME,
service_constants.VERSION_KEY: service_constants.VERSION,
service_constants.MESSAGE_KEY: 'pong'}
return jsonify(data)
# POST but not GET because GET should not change any state on the server
@app.route(route(RemoteCommand.MUTE), methods=['POST'])
def mute():
return transmit_command(RemoteCommand.MUTE)
@app.route(route(RemoteCommand.POWER), methods=['POST'])
def power():
return transmit_command(RemoteCommand.POWER)
@app.route(route(RemoteCommand.VOICE_DECREASE), methods=['POST'])
def voice_decrease():
return transmit_command(RemoteCommand.VOICE_DECREASE)
@app.route(route(RemoteCommand.VOICE_INCREASE), methods=['POST'])
def voice_increase():
return transmit_command(RemoteCommand.VOICE_INCREASE)
@app.route(route(RemoteCommand.VOLUME_DECREASE), methods=['POST'])
def volume_decrease():
return transmit_command(RemoteCommand.VOLUME_DECREASE)
@app.route(route(RemoteCommand.VOLUME_INCREASE), methods=['POST'])
def volume_increase():
return transmit_command(RemoteCommand.VOLUME_INCREASE)
@app.route("/api/v1/{}/volume-decrease-increase/".format(service_constants.API_NAME), methods=['POST'])
def volume_decrease_increase():
# apparently route decorator adds reference to "request"
post_data_dict = request.get_json()
# spell dictionary key duration-seconds with '-' similar to headers convention
duration_seconds = post_data_dict.get('duration-seconds')
logger.debug('duration_seconds: {}'.format(duration_seconds))
return scheduler.volume_decrease_increase(duration_seconds=duration_seconds, decrease_count=4, increase_count=3)
if __name__ == '__main__':
# optionally schedule jobs at pre-defined times
# scheduler.schedule_jobs()
try:
# start Flask web service
# '0.0.0.0' accessible to any device on the network
app.run(host='0.0.0.0', debug=True)
except RuntimeError:
pass
finally:
# fix RunTimeWarning This channel is already in use
# may need to put in a try:catch:finally finally section to handle exceptions
pass