Feuersoftware Connect API

connectapi.py

#!/usr/bin/env python3
#-*- coding: utf-8 -*-


import os
import click
import requests
import json
from datetime import datetime as dt
import coloredlogs, logging

logger = logging.getLogger("ConnectAPI")
logger.propagate = False
coloredlogs.install(level='INFO', fmt='%(asctime)s %(levelname)s [%(name)s] %(message)s')

class ConnectAPI:

    def __init__(self, dump=False):
        token = open(os.path.join(os.path.dirname(os.path.realpath(__file__)), 'apitoken'),'r').read().strip()
        self.headers = {
            'authorization': f'bearer {token}',
            'accept': 'application/json',
            'content-type': 'application/json',
        }
        self.body = None
        self.url = None
        self.dump = dump

    def einsatz(self, start, keyword, address=None, position=None, facts=None, ric=None, properties=None):
        self.url = "https://connectapi.feuersoftware.com/interfaces/public/operation"
        self.body = {
            'start': start,
            'keyword': keyword
        }
        if address:
           self.body['address'] = address
        if position:
           self.body['position'] =  {'latitude':position[0], 'longitude': position[1]}
        if facts:
           self.body['facts'] = facts
        if ric:
           self.body['ric'] = ric
        if properties:
           self.body['properties'] = [{'key':k,'value':v} for k,v in properties]
        self._send()


    def fahrzeugstatus(self, radio, status, position=None):
        self.url = f"https://connectapi.feuersoftware.com/interfaces/public/vehicle/{radio}/status"
        self.body = {
            'status': status
        }
        if position:
            self.body['position'] = {'latitude':position[0], 'longitude': position[1]}
        self._send()


    def _send(self):
        r = requests.post(self.url, data=json.dumps(self.body), headers=self.headers)
        if r.status_code != 200:
            logger.warning(f"Fehler beim senden des Alarms {r.status_code}, \"{r.text}\"")
        else:
            logger.info(f"Alarms erfolgreich gesendet!")
        if self.dump:
            print(self.url)
            print(self.headers)
            print(self.body)


@click.group()
def cli():
    pass

@cli.command()
@click.option('--start', required=True, default=dt.now().isoformat())
@click.option('--keyword', required=True, default="Test")
@click.option('--address', required=True, default="Strasse 1, 12345 Testort")
@click.option('--position', type=(float, float), required=True, default=(0.0, 0.0))
@click.option('--facts', required=True, default="Testalarm via Connect Schnittstelle")
@click.option('--ric', required=True, default="Testric")
@click.option('--properties', multiple=True, type=(str, str))
@click.option('--dump', is_flag=True, default=False)
def einsatz(start, keyword, address, position, facts, ric, properties, dump):
    api = ConnectAPI(dump)
    api.einsatz(start, keyword, address, position, facts, ric, list(properties))

@cli.command()
@click.option('--radio', type=int, required=True, default=12345678)
@click.option('--status', type=int, required=True)
@click.option('--lat', type=float, required=True, default=0.0)
@click.option('--lng', type=float, required=True, default=0.0)
@click.option('--dump', is_flag=True, default=False)
def fahrzeug(radio, status, lat, lng, dump):
    api = ConnectAPI(dump)
    api.fahrzeugstatus(radio, status, (lat, lng))

if __name__ == "__main__":
    cli()