Marvin's code checks

Author: mstep44

python-serial/src/thermoflex/network.py

@@ -34,11 +34,7 @@ def net_manager_thread():
         while NET_MANAGER_FLAG.is_set() == False:
 
             checklist = NodeNet.netlist.copy()
-            for net in checklist:
-                # for cmd in net.rec_cmd_buff:
-                #     net.disperse(cmd)
-                #     del net.rec_cmd_buff[cmd]
-                
+            for net in checklist:                
                 try:
                     net.update_network()
                 except TypeError:
@@ -106,7 +102,6 @@ def removeNode(self, id):
         D.debug(DEBUG_LEVELS['INFO'], self.debug_name, f"Removing node: {id}")
         for node in self.node_list:
             if node.id == id:
-                self.node_list.remove(node)
                 node.endself()
 
     def getDevice(self, id):

test/muscle-cmd-test.py

@@ -1,3 +1,96 @@
 '''
 Unit test for all muscle commands and status variables
-'''
+'''
+import thermoflex as tf
+import threading as th
+import time as t
+
+COMMAND_CHANGE_INTERVAL = 5
+STATUS_THREAD_INTERVAL =  10
+END_TEST_FLAG = th.Event()
+END_TEST_FLAG.clear()
+
+def threaded(func):
+    global threadlist
+    threadlist = []
+
+    def wrapper(*args, **kwargs):
+        thread = th.Thread(target=func, args=args, kwargs=kwargs)
+        threadlist.append(thread)
+        thread.start()
+        return thread
+    
+    return wrapper
+
+#Establish a connection to the testnet and testnode
+network = tf.discover()
+testnet = network[0]
+testnode = testnet.node_list[0]
+
+#Test the muscle command and status variables
+
+muscle1 = testnode.muscle0
+
+# Test the Muscle class methods
+@threaded
+def test_muscle_cmd(muscle):
+
+    # Set the mode of the muscle
+    muscle.setMode(conmode="amps")
+    print(f"Muscle Mode: {muscle.cmode}")
+    t.sleep(COMMAND_CHANGE_INTERVAL)
+
+    # Set a setpoint for the muscle
+    muscle.setSetpoint(conmode="percent", setpoint=0.75)
+    print("Setpoint set to 0.75 in percent mode.")
+    t.sleep(COMMAND_CHANGE_INTERVAL)
+
+    # Enable the muscle
+    muscle.setEnable(True)
+    print("Muscle enabled.")
+    t.sleep(COMMAND_CHANGE_INTERVAL)
+
+    # Disable the muscle
+    muscle.setEnable(False)
+    print("Muscle disabled.")
+    t.sleep(COMMAND_CHANGE_INTERVAL)
+
+@threaded
+def muscle_status_test(muscle):
+        
+    while not END_TEST_FLAG.is_set():
+        t.sleep(STATUS_THREAD_INTERVAL)
+        # Get the current setpoint of the muscle
+        setpoint = muscle.getSetpoint()
+        print(f"Muscle Setpoint: {setpoint}")
+
+        # Get the current mode of the muscle
+        mode = muscle.getMode()
+        print(f"Muscle Mode: {mode}")
+
+        # Get the current current of the muscle
+        current = muscle.getCurrent()
+        print(f"Muscle Current: {current}")
+
+        # Get the current voltage of the muscle
+        voltage = muscle.getVoltage()
+        print(f"Muscle Voltage: {voltage}")
+        # Get the muscle status
+        
+        status = muscle.muscleStatus()
+        print(f"Muscle Status: {status}")
+
+        # Get the resistance of the muscle
+        resistance = muscle.getResistance()
+        print(f"Muscle Resistance: {resistance}")
+        
+
+
+# Run the test
+muscle_status_test(muscle1)
+test_muscle_cmd(muscle1)
+    
+
+
+for thread  in threadlist:
+    thread.join()