plugin.divera.BoswatchPlugin Class Reference

Description of the Plugin. More...

Public Member Functions
	__init__ (self, config)
	Do not change anything here!
	fms (self, bwPacket)
	Called on FMS alarm.
	pocsag (self, bwPacket)
	Called on POCSAG alarm.
	zvei (self, bwPacket)
	Called on ZVEI alarm.
	msg (self, bwPacket)
	Called on MSG packet.
Public Member Functions inherited from plugin.pluginBase.PluginBase
	onLoad (self)
	Called by import of the plugin can be inherited.
	setup (self)
	Called before alarm can be inherited.
	teardown (self)
	Called after alarm can be inherited.
	onUnload (self)
	Called on shutdown of boswatch can be inherited.
	parseWildcards (self, msg)
	Return the message with parsed wildcards.

Protected Member Functions
	_makeRequests (self, apipath, apicall)
	_asyncRequests (self, url)
	_fetch (self, url, session)
Protected Member Functions inherited from plugin.pluginBase.PluginBase
	_cleanup (self)
	Cleanup routine calls onUnload() directly.
	_run (self, bwPacket)
	start an complete running turn of an plugin.
	_getStatistics (self)
	Returns statistical information's from last plugin run.

Additional Inherited Members
Data Fields inherited from plugin.pluginBase.PluginBase
	config
Protected Attributes inherited from plugin.pluginBase.PluginBase
	_pluginName
	_bwPacket
	_sumTime
	_cumTime
	_setupTime
	_alarmTime
	_teardownTime
	_runCount
	_setupErrorCount
	_alarmErrorCount
	_teardownErrorCount
Static Protected Attributes inherited from plugin.pluginBase.PluginBase
list	_pluginsActive = []

Detailed Description

Description of the Plugin.

Constructor & Destructor Documentation

__init__()

plugin.divera.BoswatchPlugin.__init__	(		self,
			config
	)

Do not change anything here!

Reimplemented from plugin.pluginBase.PluginBase.

    def __init__(self, config):
        r"""!Do not change anything here!"""
        super().__init__(__name__, config)  # you can access the config class on 'self.config'
 

Member Function Documentation

fms()

plugin.divera.BoswatchPlugin.fms	(		self,
			bwPacket
	)

Called on FMS alarm.

Parameters

bwPacket

bwPacket instance Remove if not implemented

Reimplemented from plugin.pluginBase.PluginBase.

    def fms(self, bwPacket):
        r"""!Called on FMS alarm
 
        @param bwPacket: bwPacket instance
        Remove if not implemented"""
        fms_data = self.config.get("fms")
        apicall = urllib.parse.urlencode({
            "accesskey": self.config.get("accesskey", default=""),
            "vehicle_ric": self.parseWildcards(fms_data.get("vehicle", default="")),
            "status_id": bwPacket.get("status"),
            "status_note": bwPacket.get("directionText"),
            "title": self.parseWildcards(fms_data.get("title", default="{FMS}")),
            "text": self.parseWildcards(fms_data.get("message", default="{FMS}")),
            "priority": fms_data.get("priority", default="false"),
        })
        apipath = "/api/fms"
        self._makeRequests(apipath, apicall)
 

pocsag()

plugin.divera.BoswatchPlugin.pocsag	(		self,
			bwPacket
	)

Called on POCSAG alarm.

Parameters

bwPacket

bwPacket instance Remove if not implemented

Reimplemented from plugin.pluginBase.PluginBase.

    def pocsag(self, bwPacket):
        r"""!Called on POCSAG alarm
 
        @param bwPacket: bwPacket instance
        Remove if not implemented"""
        poc_data = self.config.get("pocsag")
        apicall = urllib.parse.urlencode({
            "accesskey": self.config.get("accesskey", default=""),
            "title": self.parseWildcards(poc_data.get("title", default="{RIC}({SRIC})\n{MSG}")),
            "ric": self.parseWildcards(poc_data.get("ric", default="")),
            "text": self.parseWildcards(poc_data.get("message", default="{MSG}")),
            "priority": poc_data.get("priority", default="false"),
        })
        apipath = "/api/alarm"
        self._makeRequests(apipath, apicall)
 

zvei()

plugin.divera.BoswatchPlugin.zvei	(		self,
			bwPacket
	)

Called on ZVEI alarm.

Parameters

bwPacket

bwPacket instance Remove if not implemented

Reimplemented from plugin.pluginBase.PluginBase.

    def zvei(self, bwPacket):
        r"""!Called on ZVEI alarm
 
        @param bwPacket: bwPacket instance
        Remove if not implemented"""
        zvei_data = self.config.get("zvei")
        apicall = urllib.parse.urlencode({
            "accesskey": self.config.get("accesskey", default=""),
            "title": self.parseWildcards(zvei_data.get("title", default="{TONE}")),
            "ric": self.parseWildcards(zvei_data.get("ric", default="{TONE}")),
            "text": self.parseWildcards(zvei_data.get("message", default="{TONE}")),
            "priority": zvei_data.get("priority", default="false"),
        })
        apipath = "/api/alarm"
        self._makeRequests(apipath, apicall)
 

msg()

plugin.divera.BoswatchPlugin.msg	(		self,
			bwPacket
	)

Called on MSG packet.

Parameters

bwPacket

bwPacket instance Remove if not implemented

Reimplemented from plugin.pluginBase.PluginBase.

    def msg(self, bwPacket):
        r"""!Called on MSG packet
 
        @param bwPacket: bwPacket instance
        Remove if not implemented"""
        msg_data = self.config.get("msg")
        apicall = urllib.parse.urlencode({
            "accesskey": self.config.get("accesskey", default=""),
            "title": self.parseWildcards(msg_data.get("title", default="{MSG}")),
            "ric": self.parseWildcards(msg_data.get("ric", default="")),
            "text": self.parseWildcards(msg_data.get("message", default="{MSG}")),
            "priority": msg_data.get("priority", default="false"),
        })
        apipath = "/api/alarm"
        self._makeRequests(apipath, apicall)
 

_makeRequests()

plugin.divera.BoswatchPlugin._makeRequests (   self,   apipath,   apicall  )

protected

Parses wildcard urls and handles asynchronus requests

@param urls: array of urls

    def _makeRequests(self, apipath, apicall):
        """Parses wildcard urls and handles asynchronus requests
 
        @param urls: array of urls"""
        url = "https://www.divera247.com"
        request = url + apipath + "?" + apicall
 
        loop = asyncio.get_event_loop()
 
        future = asyncio.ensure_future(self._asyncRequests(request))
        loop.run_until_complete(future)
 

_asyncRequests()

plugin.divera.BoswatchPlugin._asyncRequests (   self,   url  )

protected

Handles asynchronus requests

@param urls: array of urls to send requests to

    async def _asyncRequests(self, url):
        """Handles asynchronus requests
 
        @param urls: array of urls to send requests to"""
        tasks = []
 
        async with ClientSession() as session:
            logging.debug("Generated URL: [{}]".format(url))
            task = asyncio.ensure_future(self._fetch(url, session))
            tasks.append(task)
 
            responses = asyncio.gather(*tasks)
            await responses
 

_fetch()

plugin.divera.BoswatchPlugin._fetch (   self,   url,   session  )

protected

Fetches requests

@param url: url

@param session: Clientsession instance

    async def _fetch(self, url, session):
        """Fetches requests
 
        @param url: url
 
        @param session: Clientsession instance"""
        logging.debug("Post URL: [{}]".format(url))
        async with session.post(url) as response:
            logging.info("{} returned [{}]".format(response.url, response.status))
            return await response.read()