MO_Mobile_Systeme/APED_Device
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

implemented digital API including error handling when writing

Browse Source
This commit is contained in:
p.loedige 2020-12-28 12:28:17 +01:00
parent 011147bc4f
commit 37cb64a2ab
2 changed files with 34 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
app.py
View File

@ -4,26 +4,35 @@ from flask import request
from flask_api import FlaskAPI
import RPi.GPIO as GPIO
LEDS = {"green": 16, "red": 18}
GPIO.setmode(GPIO.BOARD)
GPIO.setup(LEDS["green"], GPIO.OUT)
GPIO.setup(LEDS["red"], GPIO.OUT)
from xml_reader import Xml_reader
from interface_handler import Interface_handler
xml_reader = Xml_reader('XML/config.xml','XML/config.xsd')
interface_handler = Interface_handler(xml_reader)
device_names = xml_reader.get_device_names()
app = FlaskAPI(__name__)
@app.route('/', methods=["GET"])
def api_root():
return {
"led_url": request.url + "led/(green | red)/",
"led_url_POST": {"state": "(0 | 1)"}
"device_url": request.url + "device/<device_name>",
"device_url_POST": {"output": "<value>"}
}
@app.route('/led/<color>/', methods=["GET", "POST"])
def api_leds_control(color):
@app.route('/device/<device_name>/', methods=["GET", "POST"])
def api_leds_control(device_name):
if request.method == "POST":
if color in LEDS:
GPIO.output(LEDS[color], int(request.data.get("state")))
return {color: GPIO.input(LEDS[color])}
# if color in LEDS:
# GPIO.output(LEDS[color], int(request.data.get("state")))
try:
response = interface_handler.write(device_name,request.data.get("output"))
return {'output': response}
except Exception as e:
return {'error': str(e)}
# return {color: GPIO.input(LEDS[color])}
return {'state': interface_handler.read(device_name)}
if __name__ == "__main__":
app.run(port=8080,host="0.0.0.0")

25
interface_handler.py
View File

@ -95,9 +95,9 @@ class Interface_handler:
pins = port_info["pins"]
for pin in self.xml_reader.get_port(device_name)["pins"]:
if protocol == "DI":
if protocol == "DI" or "DO":
returnValues.update({
pin:GPIO.input(self.Pin[pin].value)
pin : GPIO.input(self.Pin[pin].value)
})
else: #throw error for protocols without write functionality
raise ValueError("you can not read from device %s" % device_name)
@ -105,19 +105,21 @@ class Interface_handler:
def write(self, device_name:str, value):
"""writes a value to all the pins of a device
"""writes an output to a given device
Args:
device_name (string): name of the device
value (float): the value to be written
device_name (str): the name of the relevant device
value (<type depends on protocol>): the value to be written
Raises:
NameError: device not found in config.xml
TypeError: value has wrong type
NameError: the device name could not be found in the config.xml
TypeError: the given Value has the wrong type for the protocol
ValueError: the value cannot be written to the device
Returns:
value if everything went well, None otherwise
<type depends on protocol>: value if everything worked correctly
"""
#check device_name is valid
if self.xml_reader.get_port(device_name) == None:
raise NameError('the device with the name %s is unknown' % device_name)
@ -129,10 +131,11 @@ class Interface_handler:
for pin in self.xml_reader.get_port(device_name)["pins"]:
if protocol == "DO":
#type check input value
if not value in [GPIO.HIGH,GPIO.LOW]:
raise ValueError("value must be GPIO.HIGH or GPIO.LOW")
if not value in ['0','1']:
raise TypeError("value must be '0' or '1'")
value = GPIO.HIGH if value == '1' else GPIO.LOW
GPIO.output(self.Pin[pin].value,value)
return value
return GPIO.input(self.Pin[pin].value)
else: #throw error for protocols without write functionality
raise ValueError("you cannot write to device %s" % device_name)