added attenuator library in realm.py

py-json/lf_attenmod.py

@@ -0,0 +1,70 @@
+#!/usr/bin/env python3
+from LANforge.lfcli_base import LFCliBase
+from LANforge import LFRequest
+from LANforge import LFUtils
+import time
+
+
+class ATTENUATORProfile(LFCliBase):
+    def __init__(self, lfclient_host, lfclient_port, local_realm, debug_=False):
+        super().__init__(lfclient_host, lfclient_port, debug_)
+        self.local_realm = local_realm
+        self.lfclient_host = lfclient_host
+        self.COMMANDS = ["show_attenuators", "set_attenuator"]
+        self.atten_serno = ""
+        self.atten_idx = ""
+        self.atten_val = ""
+        self.atten_data = {
+            "shelf": 1,
+            "resource": 1,
+            "serno": None,
+            "atten_idx": None,
+            "val": None,
+            "mode": None,
+            "pulse_width_us5": None,
+            "pulse_interval_ms": None,
+            "pulse_count": None,
+            "pulse_time_ms": None
+        }
+
+    def set_command_param(self, command_name, param_name, param_value):
+        # we have to check what the param name is
+        if (command_name is None) or (command_name == ""):
+            return
+        if (param_name is None) or (param_name == ""):
+            return
+        if command_name not in self.COMMANDS:
+            raise ValueError("Command name name [%s] not defined in %s" % (command_name, self.COMMANDS))
+        if command_name == "set_attenuator":
+            self.atten_data[param_name] = param_value
+
+    def show(self, debug=False):
+        print("Show Attenuators.........")
+        response = self.json_get("/attenuators/")
+        time.sleep(0.01)
+        if response is None:
+            print(response)
+            raise ValueError("Cannot find any endpoints")
+        else:
+            attenuator_resp = response["attenuator"]
+            for key, val in attenuator_resp.items():
+                if key == "entity id":
+                    serial_num = val.split(".")
+                    print("Serial-num : %s" % serial_num[-1])
+                print("%s : %s" % (key, val))
+        print("\n")
+
+    def create(self, debug=False):
+        if len(self.atten_serno) == 0 or len(self.atten_idx) == 0 or len(self.atten_val) == 0:
+            print("ERROR:  Must specify atten_serno, atten_idx, and atten_val when setting attenuator.\n")
+        print("Creating Attenuator...")
+        self.set_command_param("set_attenuator", "serno", self.atten_serno)
+        self.set_command_param("set_attenuator", "atten_idx", self.atten_idx)
+        self.set_command_param("set_attenuator", "val", self.atten_val)
+        set_attenuators = LFRequest.LFRequest(self.lfclient_url + "/cli-json/set_attenuator", debug_=debug)
+        set_attenuators.addPostData(self.atten_data)
+        time.sleep(0.01)
+        json_response = set_attenuators.jsonPost(debug)
+        time.sleep(10)
+        print("\n")
+

py-json/realm.py

@@ -19,6 +19,7 @@ from LANforge.lfcli_base import LFCliBase
 from l3_cxprofile import L3CXProfile
 from l3_cxprofile2 import L3CXProfile2
 from l4_cxprofile import L4CXProfile
+from lf_attenmod import ATTENUATORProfile
 from multicast_profile import MULTICASTProfile
 from http_profile import HTTPProfile
 from station_profile import StationProfile
@@ -869,7 +870,10 @@ class Realm(LFCliBase):
             #                   debug_=self.debug,
             #                   report_timer_=3000)
         return cx_prof
-
+    def new_attenuator_profile(self, ver=1):
+        if ver == 1:
+            atten_prof = ATTENUATORProfile(self.lfclient_host, self.lfclient_port, local_realm=self, debug_=self.debug)
+        return  atten_prof
     def new_generic_endp_profile(self, ver=1):
         if ver == 1 :
             endp_prof = GenCXProfile(self.lfclient_host, self.lfclient_port, local_realm=self, debug_=self.debug)

py-scripts/lf_atten_mod_test.py

@@ -0,0 +1,83 @@
+#!/usr/bin/env python3
+
+"""
+NAME: lf_atten_mod_test.py
+
+PURPOSE:
+This program is used to modify the LANforge attenuator (through the LANforge manager/server processes) by using create() method.
+You can check the attenuator details and serial number by using show() method.
+
+EXAMPLE:
+Run with all serial number and module: python3 lf_atten_mod_test.py -hst 192.168.200.12 -port 8080 -atten_serno all --atten_idx all --atten_val 220
+Run with particular serial number(2222) and module(2): python3 lf_atten_mod_test.py -hst 192.168.200.12 -port 8080 -atten_serno 2222 --atten_idx 3 --atten_val 220
+
+"atten_serno" = serial number
+"atten_idx" = module name
+
+
+Use './lf_atten_mod_test.py --help' to see command line usage and options
+Copyright 2021 Candela Technologies Inc
+License: Free to distribute and modify. LANforge systems must be licensed.
+"""
+
+import sys
+
+if sys.version_info[0] != 3:
+    print("This script requires Python 3")
+    exit(1)
+
+if 'py-json' not in sys.path:
+    sys.path.append('../py-json')
+
+import argparse
+from LANforge.lfcli_base import LFCliBase
+from LANforge.LFUtils import *
+from LANforge import LFUtils
+import realm
+import time
+
+class CreateAttenuator(LFCliBase):
+    def __init__(self, host, port, serno, idx, val,
+                 _debug_on=False,
+                 _exit_on_error=False,
+                 _exit_on_fail=False):
+        super().__init__(host, port, _local_realm=realm.Realm(host, port), _debug=_debug_on, _exit_on_fail=_exit_on_fail)
+        self.host = host
+        self.port = port
+        self.serno = serno
+        self.idx = idx
+        self.val = val
+        self.attenuator_profile = self.local_realm.new_attenuator_profile()
+        self.attenuator_profile.atten_idx = self.idx
+        self.attenuator_profile.atten_val = self.val
+        self.attenuator_profile.atten_serno = self.serno
+
+    def build(self):
+        self.attenuator_profile.create()
+        self.attenuator_profile.show()
+
+def main():
+    parser = LFCliBase.create_basic_argparse(
+        prog='lf_atten_mod_test.py',
+        formatter_class=argparse.RawTextHelpFormatter,
+        epilog=None,
+        description='''\
+            lf_atten_mod_test.py
+            --------------------
+        set and show Attenuator:
+        python3 lf_atten_mod_test.py -hst 192.168.200.12 -port 8080 -atten_serno all --atten_idx all --atten_val 220
+                ''')
+
+    parser.add_argument('-hst', '--host', help='host name', default='192.168.200.12')
+    parser.add_argument('-port', '--port', help='port name', default=8080)
+    parser.add_argument('-atten_serno', '--atten_serno', help='Serial number for requested Attenuator, or \'all\'', default=2222)
+    parser.add_argument('-atten_idx', '--atten_idx', help='Attenuator index eg. For module 1 = 0,module 2 = 1', default='all')
+    parser.add_argument('-atten_val', '--atten_val', help='Requested attenution in 1/10ths of dB (ddB).', default=550)
+    args = parser.parse_args()
+
+    atten_mod_test = CreateAttenuator(host=args.host, port=args.port, serno=args.atten_serno, idx=args.atten_idx, val=args.atten_val)
+    atten_mod_test.build()
+
+
+if __name__ == "__main__":
+    main()