plugin.telegram.BoswatchPlugin Class Reference

Description of the Plugin. More...

Public Member Functions
	__init__ (self, config)
	Do not change anything here!
	onLoad (self)
	Called by import of the plugin.
	setup (self)
	Called before alarm.
	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.
	teardown (self)
	Called after alarm.
	onUnload (self)
	Called by destruction of the plugin.
Public Member Functions inherited from plugin.pluginBase.PluginBase
	parseWildcards (self, msg)
	Return the message with parsed wildcards.

Data Fields
	sender
Data Fields inherited from plugin.pluginBase.PluginBase
	config

Protected Member Functions
	_ensure_sender (self)
	_has_sender (self)
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
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.telegram.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

_ensure_sender()

plugin.telegram.BoswatchPlugin._ensure_sender (   self )

protected

 checking with hasattr if self.sender already exists

    def _ensure_sender(self):
        r""" checking with hasattr if self.sender already exists"""
        if not hasattr(self, 'sender') or self.sender is None:
            token = self.config.get("botToken")
            ids = self.config.get("chatIds", default=[])
 
            if token and ids:
                self.sender = TelegramSender(
                    bot_token=token,
                    chat_ids=ids,
                    max_retries=self.config.get("max_retries"),
                    initial_delay=self.config.get("initial_delay"),
                    max_delay=self.config.get("max_delay"),
                    parse_mode=self.config.get("parse_mode")
                )
                logger.debug("TelegramSender initialized via Self-Healing")
            else:
                missing = []
                if not token:
                    missing.append("botToken")
                if not ids:
                    missing.append("chatIds")
                logger.error(f"Telegram configuration incomplete! Missing: {', '.join(missing)}. Plugin will not send messages.")
 

_has_sender()

plugin.telegram.BoswatchPlugin._has_sender (   self )

protected

Check if sender is available and properly initialized

    def _has_sender(self):
        r"""Check if sender is available and properly initialized"""
        return hasattr(self, 'sender') and self.sender is not None
 

onLoad()

plugin.telegram.BoswatchPlugin.onLoad

(

self

)

Called by import of the plugin.

Reimplemented from plugin.pluginBase.PluginBase.

    def onLoad(self):
        r"""!Called by import of the plugin"""
        self._ensure_sender()
        if not self._has_sender():
            logger.warning("Telegram plugin loaded but not configured.  Messages will be discarded.")
        else:
            startup = self.config.get("startup_message")
            if startup and startup.strip():
                self.sender.send_message(startup)
 

setup()

plugin.telegram.BoswatchPlugin.setup

(

self

)

Called before alarm.

Reimplemented from plugin.pluginBase.PluginBase.

    def setup(self):
        r"""!Called before alarm"""
        # ensure sender exists before alarms are processed
        self._ensure_sender()
 

fms()

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

Called on FMS alarm.

Reimplemented from plugin.pluginBase.PluginBase.

    def fms(self, bwPacket):
        r"""!Called on FMS alarm"""
        if self._has_sender():
            msg = self.parseWildcards(self.config.get("message_fms", default="{FMS}"))
            self.sender.send_message(msg)
 

pocsag()

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

Called on POCSAG alarm.

Reimplemented from plugin.pluginBase.PluginBase.

    def pocsag(self, bwPacket):
        r"""!Called on POCSAG alarm"""
        if self._has_sender():
            msg = self.parseWildcards(self.config.get("message_pocsag", default="{RIC}({SRIC})\n{MSG}"))
            self.sender.send_message(msg)
 
            if self.config.get("coordinates", default=False):
                lat = bwPacket.get("lat")
                lon = bwPacket.get("lon")
                if lat is not None and lon is not None:
                    self.sender.send_location(lat, lon)
 

zvei()

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

Called on ZVEI alarm.

Reimplemented from plugin.pluginBase.PluginBase.

    def zvei(self, bwPacket):
        r"""!Called on ZVEI alarm"""
        if self._has_sender():
            msg = self.parseWildcards(self.config.get("message_zvei", default="{TONE}"))
            self.sender.send_message(msg)
 

msg()

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

Called on MSG packet.

Reimplemented from plugin.pluginBase.PluginBase.

    def msg(self, bwPacket):
        r"""!Called on MSG packet"""
        if self._has_sender():
            msg = self.parseWildcards(self.config.get("message_msg", default="{MSG}"))
            self.sender.send_message(msg)
 

teardown()

plugin.telegram.BoswatchPlugin.teardown

(

self

)

Called after alarm.

Reimplemented from plugin.pluginBase.PluginBase.

    def teardown(self):
        r"""!Called after alarm"""
        pass
 

onUnload()

plugin.telegram.BoswatchPlugin.onUnload

(

self

)

Called by destruction of the plugin.

Reimplemented from plugin.pluginBase.PluginBase.

    def onUnload(self):
        r"""!Called by destruction of the plugin"""
        if self._has_sender():
            self.sender.shutdown()
            logger.info("Telegram plugin unloaded")

Field Documentation

sender

plugin.telegram.BoswatchPlugin.sender