testing tune and correcting it push values

Author: SafiullahOmar

src/dt4acc/custom_tango/views/repeated_publishing_view.py

@@ -44,7 +44,7 @@ def process(self, orbit_data: Orbit):
 class LegacyBPMView(ViewInterface):
     def __init__(self, prefix):
         # For Tango, prefix is like "server_name/instance_name:bpm_pv_name"
-        # We need to extract the server/instance part and construct the device name
+       
         self.prefix = prefix
         self.counter = itertools.count()
         # Extract base prefix (server_name/instance_name) by splitting on ':'
@@ -144,6 +144,8 @@ async def push(self, data: TuneData):
             device_name = f"{self.prefix}/tune_device"
             device = DeviceProxy(device_name)
 
+            logger.debug(f"TuneView.push: Writing tune X={tune_x:.10f}, Y={tune_y:.10f} to {device_name}")
+            
             await asyncio.get_event_loop().run_in_executor(
                 None,
                 lambda: device.write_attribute("x", tune_x)
@@ -152,9 +154,11 @@ async def push(self, data: TuneData):
                 None,
                 lambda: device.write_attribute("y", tune_y)
             )
+            
+            logger.info(f"TuneView.push: >>>>>>>> Successfully wrote tune X={tune_x:.10f}, Y={tune_y:.10f}")
 
         except Exception as e:
-            logger.error(f"Error processing tune object data: {e}")
+            logger.error(f"TuneView.push: ❌ Error writing tune values: {e}")
 
 
 class RepeatedResultView:
@@ -178,15 +182,23 @@ def set_bpm_mimicry(self, bpm_mimicry):
     async def heart_beat(self):
         """
         Periodic heartbeat function to push default BPM data.
+        
+        Note: We do NOT republish tune here because tune values should only
+        be published when fresh calculations are complete. Republishing cached
+        tune values would overwrite fresh values with stale ones.
         """
-        logger.debug(f"{self.__class__.__name__} heartbeat {datetime.now()}, publishing bpm, orbit, twiss")
+        logger.debug(f"{self.__class__.__name__} heartbeat {datetime.now()}, publishing bpm, orbit")
         await self.orbit_object_publisher.publish()
-        await self.tune_publisher.publish()
+        # DO NOT publish tune here - it's published by push_twiss() with fresh values
+        # await self.tune_publisher.publish()  # ← REMOVED: This was overwriting fresh tune values with stale cache
         await self.legacy_bpm_publisher.publish()
-        logger.info(f"{self.__class__.__name__} view heartbeat {datetime.now()}, published bpm, orbit, twiss")
+        logger.info(f"{self.__class__.__name__} view heartbeat {datetime.now()}, published bpm, orbit")
 
     async def push_twiss(self, twiss_result: TwissWithAggregatedKValues):
-        self.tune_publisher.set_data(TuneData(x=twiss_result.x.tune, y=twiss_result.y.tune))
+        tune_x = float(twiss_result.x.tune)
+        tune_y = float(twiss_result.y.tune)
+        logger.info(f"RepeatedResultView.push_twiss: Publishing FRESH tune values X={tune_x:.10f}, Y={tune_y:.10f}")
+        self.tune_publisher.set_data(TuneData(x=tune_x, y=tune_y))
         await self.tune_publisher.publish()
 
     async def push_orbit(self, orbit_result: Orbit):

tests/test_small_current_changes.py

@@ -0,0 +1,113 @@
+#!/usr/bin/env python3
+"""
+Test script with SMALL, realistic current changes.
+
+This tests whether small current adjustments (±1%, ±5%, ±10%)
+produce corresponding small tune changes, which would be expected
+in normal accelerator operation.
+"""
+
+import time
+import sys
+from tango import DeviceProxy
+
+def test_small_changes():
+    """Test tune response to small current changes."""
+    
+    print("=" * 80)
+    print("SMALL CURRENT CHANGES TEST - Realistic operational adjustments")
+    print("=" * 80)
+    
+    try:
+        # Connect to devices
+        tune_device_name = "SimpleTangoServer/test/tune_device"
+        pc_device_name = "SimpleTangoServer/test/power_converter_Q3P2T6R"
+        
+        print(f"\nConnecting to devices...")
+        tune_device = DeviceProxy(tune_device_name)
+        pc_device = DeviceProxy(pc_device_name)
+        print(f"  ✓ Tune device: {tune_device_name}")
+        print(f"  ✓ Power converter: {pc_device_name}")
+        
+        # Get initial values
+        initial_current = pc_device.read_attribute("current_setpoint").value
+        initial_tune_x = tune_device.read_attribute("x").value
+        initial_tune_y = tune_device.read_attribute("y").value
+        
+        print(f"\n Initial State:")
+        print(f"  Current:  {initial_current:.6f} A")
+        print(f"  Tune X:   {initial_tune_x:.10f}")
+        print(f"  Tune Y:   {initial_tune_y:.10f}")
+        
+        print("\n" + "=" * 80)
+        print("TEST SEQUENCE: Small incremental changes")
+        print("=" * 80)
+        
+        # Test sequence with small changes
+        # Format: (percentage, description)
+        test_changes = [
+            (1.01, "+1% (very small)"),
+            (1.02, "+2%"),
+            (1.05, "+5%"),
+            (1.10, "+10%"),
+            (1.15, "+15%"),
+            (1.00, "Back to initial"),
+            (0.99, "-1%"),
+            (0.98, "-2%"),
+            (0.95, "-5%"),
+            (0.90, "-10%"),
+            (0.85, "-15%"),
+            (1.00, "Restore initial"),
+        ]
+        
+        results = []
+        tune_changed_count = 0
+        
+        for i, (factor, description) in enumerate(test_changes, 1):
+            test_current = initial_current * factor
+            
+            print(f"\n--- Test {i}/{len(test_changes)}: {description} → {test_current:.4f} A ---")
+            
+            # Set the new current
+            pc_device.write_attribute("current_setpoint", test_current)
+            
+            # Wait for calculations (longer wait to ensure completion)
+            wait_time = 2.0
+            time.sleep(wait_time)
+            
+            # Read values
+            readback_current = pc_device.read_attribute("current_readback").value
+            new_tune_x = tune_device.read_attribute("x").value
+            new_tune_y = tune_device.read_attribute("y").value
+            
+            
+        
+        return tune_changed_count > 0
+        
+    except Exception as e:
+        print(f"\n❌ ERROR: {e}")
+        import traceback
+        traceback.print_exc()
+        return False
+    
+    finally:
+        # Restore initial current
+        try:
+            if pc_device and initial_current is not None:
+                print(f"\n Restoring initial current: {initial_current:.6f} A")
+                pc_device.write_attribute("current_setpoint", initial_current)
+                time.sleep(2.0)
+                print("✓ Current restored")
+        except:
+            pass
+
+
+if __name__ == "__main__":
+    print("\n" + "" * 40)
+    print("SMALL CURRENT CHANGES → TUNE RESPONSE TEST")
+    print("" * 40 + "\n")
+    
+    success = test_small_changes()
+    
+  
+