Skip to main content

Python 3 API wrapper for FireServiceRota/BrandweerRooster

Project description

Python: FireServiceRota / BrandweerRooster

Basic Python 3 API wrapper for FireServiceRota and BrandweerRooster for use with Home Assistant

About

This package allows you to get notified about emergency incidents from https://www.FireServiceRota.co.uk and https://www.BrandweerRooster.nl. Those are services used by firefighters.

It currently provides the following limited functionality:

  • Connect to the websocket to get incident details in near-realtime
  • Get user availability (duty)
  • Set user incident response status

See https://fireservicerota.co.uk and https://brandweerrooster.nl for more details.

NOTE: You need a subscription and login account to be able to use it.

Installation

pip3 install pyfireservicerota

Usage

Initialise module using user credentials to get token_info

#!/usr/bin/env python3

from pyfireservicerota import FireServiceRota, FireServiceRotaIncidents, ExpiredTokenError, InvalidTokenError, InvalidAuthError
import logging
import sys
import json
import time
import threading

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

token_info = {}

api = FireServiceRota(
      base_url="www.brandweerrooster.nl",
      username="your@email.address",
      password="yourpassword",
)

try:
    token_info = api.request_tokens()
except InvalidAuthError:
    token_info = None

if not token_info:
    _LOGGER.error("Failed to get access tokens")

NOTE: You don't need to store user credentials, at first authentication just the token_info dictionary is enough use api.refresh_tokens to refresh it.

Initialise module with stored token_info

#!/usr/bin/env python3

from pyfireservicerota import FireServiceRota, FireServiceRotaIncidents, ExpiredTokenError, InvalidTokenError, InvalidAuthError
import logging
import sys
import json
import time


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

token_info = {}

api = FireServiceRota(
      base_url = "www.brandweerrooster.nl",
      token_info = token_info
    )

# Get user availability (duty)
try:
   print(api.get_availability('Europe/Amsterdam'))
except ExpiredTokenError:
   _LOGGER.debug("Tokens are expired, refreshing")
   try:
       token_info = api.refresh_tokens()
   except InvalidAuthError:
       _LOGGER.debug("Invalid refresh token, you need to re-login")
except InvalidTokenError:
   _LOGGER.debug("Tokens are invalid")
   try:
       token_info = api.refresh_tokens()
   except InvalidAuthError:
       _LOGGER.debug("Invalid refresh token, you need to re-login")

# Get incident response status for incident with id 123456

incident_id = 123456

try:
   print(api.get_incident_response(incident_id))
except ExpiredTokenError:
   _LOGGER.debug("Tokens are expired, refreshing")
   try:
       token_info = api.refresh_tokens()
   except InvalidAuthError:
       _LOGGER.debug("Invalid refresh token, you need to re-login")
except InvalidTokenError:
   _LOGGER.debug("Tokens are invalid")
   try:
       token_info = api.refresh_tokens()
   except InvalidAuthError:
       _LOGGER.debug("Invalid refresh token, you need to re-login")


# Set incident response to acknowlegded (False = 'rejected')
try:
   api.set_incident_response(id, True)
except ExpiredTokenError:
   _LOGGER.debug("Tokens are expired, refreshing")
   try:
       token_info = api.refresh_tokens()
   except InvalidAuthError:
       _LOGGER.debug("Invalid refresh token, you need to re-login")
except InvalidTokenError:
   _LOGGER.debug("Tokens are invalid")
   try:
       token_info = api.refresh_tokens()
   except InvalidAuthError:
       _LOGGER.debug("Invalid refresh token, you need to re-login")


# Connect to websocket channel for incidents
wsurl = f"wss://www.brandweerrooster.nl/cable?access_token={token_info['access_token']}"

class FireService():

    def __init__(self, url):

        self._data = None
        self.listener = None
        self.url = url
        self.incidents_listener()

    def on_incident(self, data):
        _LOGGER.debug("INCIDENT: %s", data)
        self._data = data

    def incidents_listener(self):
        """Spawn a new Listener and links it to self.on_incident."""

        self.listener = FireServiceRotaIncidents(on_incident=self.on_incident)
        _LOGGER.debug("Starting incidents listener")
        self.listener.start(url=self.url)


ws = FireService(wsurl)

while True:
    time.sleep(1)

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

pyfireservicerota-0.0.47.tar.gz (6.8 kB view details)

Uploaded Source

Built Distribution

pyfireservicerota-0.0.47-py3-none-any.whl (7.6 kB view details)

Uploaded Python 3

File details

Details for the file pyfireservicerota-0.0.47.tar.gz.

File metadata

  • Download URL: pyfireservicerota-0.0.47.tar.gz
  • Upload date:
  • Size: 6.8 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/4.0.2 CPython/3.11.2

File hashes

Hashes for pyfireservicerota-0.0.47.tar.gz
Algorithm Hash digest
SHA256 59d787f66181cd759cc30cee7b11f8833f420f54cc12539f6a05c7b189611e4d
MD5 3572e52075a106fac509ef4aeebf0ac3
BLAKE2b-256 717902bef58461719b06dee7e4ea160d450f2125e828c7cecd91b5859dbf76a3

See more details on using hashes here.

File details

Details for the file pyfireservicerota-0.0.47-py3-none-any.whl.

File metadata

File hashes

Hashes for pyfireservicerota-0.0.47-py3-none-any.whl
Algorithm Hash digest
SHA256 5c42826c952787bc980bf1a530ecb9c1c41c9e4738466c866b5b70aeb8f640ed
MD5 8aabf7c0a67bea3c296409ab729125e6
BLAKE2b-256 f3f8e71c9a489ca66d8302ec76b67f2a88d0320e84705b6d1b33eeaf45544f9a

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page