In this post you will discover how to send CISCO NSO netconf notifications. Basically you have to go through 3 steps:

Configure the netconf stream in the NSO ncs.conf config file.

Create YANG model in the NSO.

Produce a python code to send the notification.

Firstly, you should configure the ncs.conf. You have to set a Netconf notification stream in the ncs config file. Here is an example how to configure the stream in the NSO:

<notifications>
    <event-streams>
      <stream>
        <name>notif-stream</name>
        <description>EXAMPLE</description>
        <replay-support>true</replay-support>
        <builtin-replay-store>
          <enabled>true</enabled>
          <dir>./state</dir>
          <max-size>S10M</max-size>
          <max-files>50</max-files>
        </builtin-replay-store>
      </stream>
    </event-streams>
  </notifications>

Secondly, you have to create yang model with namespace in the NSO:

module netconf-notification{

  namespace "https://dmytro.shytyi.net/netconf-notification-example";
  prefix notif;

  description
    "Complete example of Netconf Notification";

  revision 2018-08-01 {
    description
      "Netconf notification example";
  }

  notification netconf-notif {
    leaf state {
      type string;
      mandatory true;
    }
  }
}

Don’t forget to compile YANG model and macke package-reload in the NSO.

The next step is to write python logic:

import ncs,_ncs
from ncs.application import Service
import ncs.experimental
from _ncs import dp
import socket
import datetime
from ncs import maapi

#maapi needs to get schemas, to make str2hash work.
maapi = maapi.Maapi()

mysocket = socket.socket()

try:
    ctx = dp.init_daemon('send-notif')
    # making a socket
    dp.connect(dx=ctx, sock=mysocket, type=dp.CONTROL_SOCKET, ip='127.0.0.1', port=_ncs.PORT)
    # getting all the required hashes
    ns_hash = _ncs.str2hash("https://dmytro.shytyi.net/netconf-notification-example")
    notif_id_hash = _ncs.str2hash('state')
    mynotif_hash = _ncs.str2hash('netconf-notif')

    # making the notification
    message = []
    message += [_ncs.TagValue(_ncs.XmlTag(ns_hash, mynotif_hash),
        _ncs.Value((mynotif_hash, ns_hash), _ncs.C_XMLBEGIN))]
    message += [_ncs.TagValue(ncs.XmlTag(ns_hash, notif_id_hash),
        _ncs.Value("ready"))]
    message += [_ncs.TagValue(_ncs.XmlTag(ns_hash, mynotif_hash),
        _ncs.Value((mynotif_hash, ns_hash), _ncs.C_XMLEND))]
    # registering the stream
    livectx = dp.register_notification_stream(ctx, None, mysocket, "notif-stream")
    # time
    now = datetime.datetime.now()
    time = _ncs.DateTime(now.year, now.month, now.day, now.hour, now.minute,
            now.second, now.microsecond, now.timetz().hour,
            now.timetz().minute)
    # sending the notification
    dp.notification_send(livectx, time, message)
except Exception as e:
    print (e)
finally:
    mysocket.close()