Monitor your training model using any service (Telegram/SMS/Pushover and more...)

(Vijayabhaskar J) #1

I’m new to creating fastai callbacks, appreciate any feedback. I created this callback which uses the awesome Apprise library to send notifications at each epoch end.
You can use almost any combination of Instant Messaging, SMS, Email, Push Notification service to monitor your training model.
Callback Code:

class NotifierCallback(Callback):
    "Notifies you the losses and the metrics"
    def __init__(self, learn, service_addrs):
        self.learn = learn
        self.service_addrs = listify(service_addrs)
        try:
            import apprise
        except:
            raise ModuleNotFoundError("Apprise module not found, Install it to use this Callback.")
        self.apobj = apprise.Apprise()
        for addrs in self.service_addrs:
            self.apobj.add(addrs)
    def _get_values(self):
        "Store Values to be Notified"
        if len(self.learn.recorder.losses) == 0: return None
        self.values = defaultdict(list)
        self.values['train_loss']=self.learn.recorder.losses[-1].cpu().numpy()
        if self.learn.recorder.val_losses[-1]:
            self.values['valid_loss']=self.learn.recorder.val_losses[-1]
        if self.learn.recorder.metrics:
            for m, n in zip(self.learn.recorder.metrics[-1],self.learn.recorder.names[3:-1]):
                self.values["metrics"].append((n,m.cpu().numpy()))
        return self.values
    
    def on_epoch_end(self, epoch, **kwargs):
        self.values = self._get_values()
        msg_body = []
        self.epoch = epoch
        if self.values:
            title = f"Epoch {self.epoch}:"
            val = self.values.get("valid_loss",None)
            if val: msg_body.append(f"Valid Loss: {val}")
            val = self.values.get("train_loss",None)
            if val: msg_body.append(f"Train Loss: {val}")
            metrics = self.values.get("metrics",None)
            if metrics:
                for metric in metrics:
                    msg_body.append(f"{metric[0]}: {metric[1]}")
            try:
                self.apobj.notify(title=title,body="\n".join(msg_body))
            except:
                pass # Skip incase of Notifiaction error to continue training

Example usage:

telegram_addrs = f'"tgram://{bot_token}/{chat_id}" #Telegram Notification
windows = "windows://"  #Windows Desktop Notification
service_addrs=[telegram_addrs,windows]
learn.fit_one_cycle(25, max_lr=1e-3,callbacks=[NotifierCallback(learn,services_addrs)])

There is a huge list of services supported by Apprise which you can view at Apprise Github Page

2 Likes