initial

.gitignore

@@ -0,0 +1 @@
+.venv

requirements.txt

@@ -0,0 +1 @@
+pyagentx3==0.1.4

snippets/agentx/test.py

@@ -0,0 +1,62 @@
+import pyagentx3
+import threading
+import time
+
+
+# --------------------------------------------
+import logging
+class NullHandler(logging.Handler):
+    def emit(self, record):
+        pass
+logger = logging.getLogger('pyagentx3.main')
+logger.addHandler(NullHandler())
+# --------------------------------------------
+
+
+
+# Updater class that set OID values
+class TestData(pyagentx3.Updater):
+    def update(self):
+        self.set_INTEGER('1', self.data_store['cnt'])
+
+class TestAgent(pyagentx3.Agent):
+    def __init__(self, agent_id='TestAgent', socket_path=None, data_store=None):
+      self.data_store = data_store
+      super().__init__(agent_id, socket_path)
+
+    def setup(self):
+      self.register('1.3.6.1.4.1.9676.1', TestData, freq=1, data_store=self.data_store)
+
+class Data(threading.Thread):
+    def __init__(self):
+      self.data = {'cnt': 1}
+      self.stop_event = threading.Event()
+      threading.Thread.__init__(self)
+      
+    def run(self):
+      while not self.stop_event.is_set():
+        self.data['cnt'] += 1
+        logger.info('data updated')
+        time.sleep(1.0)
+      logger.info('Data collector stopping')
+
+    def stop(self):
+      self.stop_event.set()
+      self.join()
+
+
+# Main
+pyagentx3.setup_logging()
+try:
+    d = Data()
+    d.start()
+    a = TestAgent(data_store=d.data)
+    a.start()
+except Exception as e:
+    print( "Unhandled exception:", e)
+    a.stop()
+    d.stop()
+except KeyboardInterrupt:
+    a.stop()
+    d.stop()
+

snippets/ntp/test.py

@@ -0,0 +1,16 @@
+import ntp.packet
+
+session = ntp.packet.ControlSession()
+session.openhost('localhost')
+
+peers = session.readstat()
+
+l = session.readvar(0)
+print(f"{l=}")
+
+for p in peers:
+  l = session.readvar(p.associd)
+  print(f"{p.associd}: {dict(l)}")
+
+
+