Is there any way to get notified in python when a Quantumleap o Orion subscription fires for a changed value?
I've dockerized the nickjj web server, switched the bind_host
to 0.0.0.0
as specified in the documentation, and exposed the port 8008:8008
in docker-compose
.
from http.server import HTTPServer, BaseHTTPRequestHandler
from sys import argv
BIND_HOST = '0.0.0.0'
PORT = 8008
class SimpleHTTPRequestHandler(BaseHTTPRequestHandler):
def do_GET(self):
self.write_response(b'')
def do_POST(self):
content_length = int(self.headers.get('content-length', 0))
body = self.rfile.read(content_length)
self.write_response(body)
def write_response(self, content):
self.send_response(200)
self.end_headers()
self.wfile.write(content)
print(self.headers)
print(content.decode('utf-8'))
if len(argv) > 1:
arg = argv[1].split(':')
BIND_HOST = arg[0]
PORT = int(arg[1])
print(f'Listening on http://{BIND_HOST}:{PORT}\n')
httpd = HTTPServer((BIND_HOST, PORT), SimpleHTTPRequestHandler)
httpd.serve_forever()
Dockerfile to build the web server service:
# Build the service from the base image
FROM python:3.9.16-alpine3.17
# During building, run the following commands to install dependecies
RUN apk add --no-cache openssl-dev curl-dev
RUN pip install --upgrade pip
# RUN pip install pycurl crate requests
# Set the work directory
WORKDIR /webserver
# Copy in the continer the following files
COPY ./webserver ./webserver
# Run this command at startup
CMD ["python","-u","webserver", "webserver:8008"]
docker-compose configuration for the web server, with curl
for debug:
webserver:
image: webserver
stdin_open: true # docker run -i
tty: true # docker run -t
container_name: 'webserver'
ports:
- "8008:8008"
networks:
- default
I've provisioned the relative subscription in Orion:
curl -iX POST \
'http://localhost:1026/v2/subscriptions/' \
-H 'Content-Type: application/json' \
-H 'fiware-service: opcua_car' \
-H 'fiware-servicepath: /demo' \
-d '{
"description": "Subscriptions for Python webserver",
"subject": {
"entities": [
{
"idPattern": ".*",
"type": "PLC"
}
],
"condition": {
"attrs": [
"processStatus"
]
}
},
"notification": {
"http": {
"url": "http://webserver:8008/"
},
"attrs": [
"processStatus"
],
"metadata": [
"dateCreated",
"dateModified"
]
},
"throttling": 1
}'
And I finally get my subscription notifications in Python.