diff --git a/examples/combined_video_mode_voltage_control.py b/examples/combined_video_mode_voltage_control.py index 28bbaab..f2e1bbf 100644 --- a/examples/combined_video_mode_voltage_control.py +++ b/examples/combined_video_mode_voltage_control.py @@ -21,6 +21,8 @@ def main(): y_axis = video_mode_component.data_acquirer.y_axis x_axis.offset_parameter = voltage_control_component.voltage_parameters[0] y_axis.offset_parameter = voltage_control_component.voltage_parameters[1] + #logger.debug(f"offset parameter x-axis: {x_axis.offset_parameter}") + #logger.debug(f"voltage parameter 0: {voltage_control_component.voltage_parameters[0].get()}") app = build_dashboard( components=[video_mode_component, voltage_control_component], diff --git a/examples/combined_video_mode_voltage_control_simulated.py b/examples/combined_video_mode_voltage_control_simulated.py new file mode 100644 index 0000000..5a87162 --- /dev/null +++ b/examples/combined_video_mode_voltage_control_simulated.py @@ -0,0 +1,192 @@ +""" +Example Script: +Video Mode with SimulatedDataAcquirer for a four dot transition with 2 sensors, and voltage control +No barrier gate is used in this example. +""" + +from qua_dashboards.core import build_dashboard +from qua_dashboards.utils import setup_logging +from qua_dashboards.video_mode import ( + SweepAxis, + SimulatedDataAcquirer, + VideoModeComponent, +) +from qua_dashboards.voltage_control import VoltageControlComponent +from qdarts.experiment import Experiment +import numpy as np + +def get_video_mode_component() -> VideoModeComponent: + """ + Creates and returns a VideoModeComponent instance configured with a SimulatedDataAcquirer. + + This function encapsulates the setup of the sweep axes and the data acquirer, + making it easy to reuse this configuration or import it into other scripts. + + Returns: + VideoModeComponent: The configured video mode component. + """ + # Define the system + + #All capacitances are given in aF + N = 6 #number of dots + C_DD=20* np.eye((N))/2 #The self-capacitance of each dot, NOTE: factor of 2 due to symmetrization + C_DD[0,1] = 10 #capacitance between dot 0 and dot 1 (Left double dot) + C_DD[2,3] = 7 #capacitance between dot 3 and dot 4 (Right double dot) + + C_DD[0,4] = 1.6 #capacitance between sensor dot 4 and dot 0 + C_DD[1,4] = 1.4 #capacitance between sensor dot 4 and dot 1 + C_DD[2,5] = 1.4 #capacitance between sensor dot 5 and dot 2 + C_DD[3,5] = 2 #capacitance between sensor dot 5 and dot 3 + C_DD[1,2] = 6 #capacitance between the middle dots 2 and dot 3 + C_DD = C_DD + C_DD.T + + C_DG=11*np.eye(N) #dot-to-gate capacitances + #cross-capacitances + C_DG[0,1] = 1.5 #dot 0 from dot 1 + C_DG[1,0] = 1.2 #dot 1 from dot 0 + C_DG[2,3] = 1.3 #dot 2 from dot 3 + C_DG[3,2] = 1.4 #dot 3 from dot 3 + + # Definition of the tunnel couplings in eV + # NOTE: we use the convention that tc is the energy gap at avoided crossing H = tc/2 sx + tunnel_couplings = np.zeros((N,N)) + tunnel_couplings[0,1] = 50*1e-6 + tunnel_couplings[1,0] = 50*1e-6 + tunnel_couplings[2,3] = 60*1e-6 + tunnel_couplings[3,2] = 60*1e-6 + + # Experiment configurations + capacitance_config = { + "C_DD" : C_DD, #dot-dot capacitance matrix + "C_Dg" : C_DG, #dot-gate capacitance matrix + "ks" : 4, #distortion of Coulomb peaks. NOTE: If None -> constant size of Coublomb peak + } + + tunneling_config = { + "tunnel_couplings": tunnel_couplings, #tunnel coupling matrix + "temperature": 0.1, #temperature in Kelvin + "energy_range_factor": 5, #energy scale for the Hamiltonian generation. NOTE: Smaller -> faster but less accurate computation + } + sensor_config = { + "sensor_dot_indices": [4,5], #Indices of the sensor dots + "sensor_detunings": [-0.0005,-0.0005], #Detuning of the sensor dots + "noise_amplitude": {"fast_noise": 0.5*1e-6, "slow_noise": 1e-8}, #Noise amplitude for the sensor dots in eV + "peak_width_multiplier": 15, #Width of the sensor peaks in the units of thermal broadening m *kB*T/0.61. + } + + # Set up experiment + experiment = Experiment(capacitance_config, tunneling_config, sensor_config) + + # Arguments for the function that renders the capacitance CSD + unit = 'mV' + factor_mV_to_V = 1e-3 + span_x = 20 + span_y = 20 + points_x = 50 + points_y = 50 + + P=np.zeros((6,2)) + P[0,0]=1 + P[1,1]=1 + state_hint_lower_left = [1,1,0,0,3,3] + + args_sensor_scan_2D = { + "P": P, + "minV": [-span_x/2.*factor_mV_to_V,-span_y/2.*factor_mV_to_V], + "maxV": [ span_x/2.*factor_mV_to_V, span_y/2.*factor_mV_to_V], + "resolution": [points_x,points_y], + "state_hint_lower_left": state_hint_lower_left, + "cache": True, + "insitu_axis": None, + } + + # Define the X-axis for the 2D scan. + x_axis = SweepAxis( + name="x", # Internal identifier for the axis. + label="X Coordinate", # Display label in the UI. + units=unit, # Physical units of the axis. + span=span_x, # The total range of the X-axis sweep. + points=points_x, # The number of points (pixels) along the X-axis. + ) + + # Define the Y-axis for the 2D scan. + y_axis = SweepAxis( + name="y", # Internal identifier for the axis. + label="Y Coordinate", # Display label in the UI. + units=unit, # Physical units of the axis. + span=span_y, # The total range of the Y-axis sweep. + points=points_y, # The number of points (pixels) along the Y-axis. + ) + + # Instantiate the SimulatedDataAcquirer. + # This acquirer simulates data fetching by generating random 2D arrays. + simulated_acquirer = SimulatedDataAcquirer( + component_id="simulated-data-acquirer", # Unique ID for Dash elements. + x_axis=x_axis, + y_axis=y_axis, + experiment = experiment, + args_rendering = args_sensor_scan_2D, + conversion_factor_unit_to_volt=factor_mV_to_V, + SNR=20, # Signal-to-noise ratio on simulated images + acquire_time=0.1, # Simulated delay (seconds) for acquiring one raw frame. + num_software_averages=5, # Number of raw frames to average for display. + acquisition_interval_s=0.5, # Target time (seconds) between acquiring raw frames. + sensor_number=1, + ) + + # Instantiate the VideoModeComponent. + # This is the main UI component that displays the live 2D plot and controls. + video_mode_component = VideoModeComponent( + component_id=VideoModeComponent.DEFAULT_COMPONENT_ID, # Uses a default ID. + data_acquirer=simulated_acquirer, # The source of the data. + # How often the frontend asks for new data (linked to acquirer's interval). + data_polling_interval_s=simulated_acquirer.acquisition_interval_s, + ) + return video_mode_component + +def get_voltage_control_component(video_mode_component, labels) -> VoltageControlComponent: + voltage_controller = video_mode_component.data_acquirer.get_voltage_control_component(labels=labels) + return voltage_controller + + +def main() -> None: + """ + Sets up logging, creates the VideoModeComponent, builds the dashboard, + and runs the Dash application server. + """ + # Configure logging for the application. + logger = setup_logging(__name__) + logger.info("Starting Video Mode application with SimulatedDataAcquirer (based on QDarts).") + + # Get the configured VideoModeComponent. + video_mode_component = get_video_mode_component() + logger.info( + f"VideoModeComponent instance created: {video_mode_component.component_id}" + ) + + # Get the VoltageControlComponent + voltage_control_component = get_voltage_control_component(video_mode_component, labels=None) + logger.info( + f"VoltageControlComponent instance created: {voltage_control_component.component_id}" + ) + + logger.info("Building the dashboard...") + # Use build_dashboard to create the Dash app layout. + app = build_dashboard( + components=[video_mode_component,voltage_control_component], # List of dashboard components to include. + title="Video Mode Simulation (QDarts)", # Browser window title. + ) + + logger.info("Dashboard built. Starting Dash server on http://localhost:8050") + # Run the Dash server. + app.run( + debug=True, # Enables helpful Dash debugging features. + host="0.0.0.0", # Makes the server accessible on your local network. + port=8050, # Sets the server port. + use_reloader=False, # Often recommended for stability with background threads. + ) + + +if __name__ == "__main__": + # This ensures that main() is called only when the script is executed directly. + main() \ No newline at end of file diff --git a/examples/combined_video_mode_voltage_control_simulated_two-dot-system.py b/examples/combined_video_mode_voltage_control_simulated_two-dot-system.py new file mode 100644 index 0000000..662a98b --- /dev/null +++ b/examples/combined_video_mode_voltage_control_simulated_two-dot-system.py @@ -0,0 +1,203 @@ +""" +Example Script: +Video Mode with SimulatedDataAcquirer for a two dot system with one sensor, and voltage control + +""" + +from qua_dashboards.core import build_dashboard +from qua_dashboards.utils import setup_logging +from qua_dashboards.video_mode import ( + SweepAxis, + SimulatedDataAcquirer, + VideoModeComponent, +) +from qdarts.experiment import Experiment +import numpy as np + +from qua_dashboards.voltage_control import VoltageControlComponent +from qua_dashboards.utils import BasicParameter + + +def get_video_mode_component() -> VideoModeComponent: + """ + Creates and returns a VideoModeComponent instance configured with a SimulatedDataAcquirer. + + This function encapsulates the setup of the sweep axes and the data acquirer, + making it easy to reuse this configuration or import it into other scripts. + + Returns: + VideoModeComponent: The configured video mode component. + """ + # Define the system + + #All capacitances are given in aF + N = 3 #number of dots + C_DD=20* np.eye((N))/ 2 #The self-capacitance of each dot, NOTE: factor of 2 due to symmetrization + C_DD[1,1] = 10 / 2 #NOTE: factor of 2 due to symmetrization + C_DD[0,1] = 5 #capacitance between dot 0 and dot 1 (Left double dot) + + C_DD[0,2] = 1.6/2 #capacitance between sensor dot 2 and dot 0 + C_DD[1,2] = 1.4/2 #capacitance between sensor dot 2 and dot 1 + C_DD = C_DD + C_DD.T + + C_DG=11*np.eye(N,N+1) #dot-to-gate capacitances, there is one barrier gate (index 3) + # cross-capacitances + C_DG[0,1] = 1.5 #dot 0 from gate 1 + C_DG[1,0] = 1.2 #dot 1 from gate 0 + C_DG[0,3] = 0 #dot 0 from barrier gate + C_DG[1,3] = 0 #dot 1 from barrier gate + + # Definition of the tunnel couplings in eV + # NOTE: we use the convention that tc is the energy gap at avoided crossing H = tc/2 sx + tunnel_couplings = np.zeros((N,N)) + tunnel_couplings[0,1] = 50*1e-6 + tunnel_couplings[1,0] = 50*1e-6 + + # Definition of barrier levels + barrier_levers = np.zeros((N,N,N+1)) # barrier between dot i and dot j is affected by gate k + barrier_levers[0,1,3] = 100*1e-6 + barrier_levers[1,0,3] = 100*1e-6 + barrier_levers = np.log(barrier_levers + 1.e-20) + + # Experiment configurations + capacitance_config = { + "C_DD" : C_DD, #dot-dot capacitance matrix + "C_Dg" : C_DG, #dot-gate capacitance matrix + "ks" : 4, #distortion of Coulomb peaks. NOTE: If None -> constant size of Coublomb peak + } + + tunneling_config = { + "tunnel_couplings": tunnel_couplings, #tunnel coupling matrix + "temperature": 0.1, #temperature in Kelvin + "energy_range_factor": 1, #energy scale for the Hamiltonian generation. NOTE: Smaller -> faster but less accurate computation + "barrier_levers": barrier_levers, #barrier levels matrix + } + + sensor_config = { + "sensor_dot_indices": [2], #Indices of the sensor dots + "sensor_detunings": [-0.0005], #Detuning of the sensor dots + "noise_amplitude": {"fast_noise": 0.5*1e-6, "slow_noise": 1e-8}, #Noise amplitude for the sensor dots in eV + "peak_width_multiplier": 15, #Width of the sensor peaks in the units of thermal broadening m *kB*T/0.61. + } + + # Set up experiment + experiment = Experiment(capacitance_config, tunneling_config, sensor_config) + + # Arguments for the function that renders the capacitance CSD + unit = 'mV' + factor_mV_to_V = 1e-3 + span_x = 20 + span_y = 20 + points_x = 50 + points_y = 50 + + P=np.zeros((N+1,2)) + P[0,0]=1 + P[1,1]=1 + state_hint_lower_left = [1,1,5] + + W = np.eye(2) + + args_sensor_scan_2D = { + "P": P, + "virtualisation_matrix": W, + "minV": [-span_x/2.*factor_mV_to_V,-span_y/2.*factor_mV_to_V], + "maxV": [ span_x/2.*factor_mV_to_V, span_y/2.*factor_mV_to_V], + "resolution": [points_x,points_y], + "state_hint_lower_left": state_hint_lower_left, + "cache": False, # Do not cache the results + "insitu_axis": None, + } + + # Define the X-axis for the 2D scan. + x_axis = SweepAxis( + name="x", # Internal identifier for the axis. + label="X Coordinate", # Display label in the UI. + units=unit, # Physical units of the axis. + span=span_x, # The total range of the X-axis sweep. + points=points_x, # The number of points (pixels) along the X-axis. + ) + + # Define the Y-axis for the 2D scan. + y_axis = SweepAxis( + name="y", # Internal identifier for the axis. + label="Y Coordinate", # Display label in the UI. + units=unit, # Physical units of the axis. + span=span_y, # The total range of the Y-axis sweep. + points=points_y, # The number of points (pixels) along the Y-axis. + ) + + # Instantiate the SimulatedDataAcquirer. + # This acquirer simulates data fetching by generating random 2D arrays. + simulated_acquirer = SimulatedDataAcquirer( + component_id="simulated-data-acquirer", # Unique ID for Dash elements. + x_axis=x_axis, + y_axis=y_axis, + experiment = experiment, + args_rendering = args_sensor_scan_2D, + conversion_factor_unit_to_volt=factor_mV_to_V, + SNR=20, # Signal-to-noise ratio on simulated images + acquire_time=0.1, # Simulated delay (seconds) for acquiring one raw frame. + num_software_averages=5, # Number of raw frames to average for display. + acquisition_interval_s=0.5, # Target time (seconds) between acquiring raw frames. + ) + + # Instantiate the VideoModeComponent. + # This is the main UI component that displays the live 2D plot and controls. + video_mode_component = VideoModeComponent( + component_id=VideoModeComponent.DEFAULT_COMPONENT_ID, # Uses a default ID. + data_acquirer=simulated_acquirer, # The source of the data. + # How often the frontend asks for new data (linked to acquirer's interval). + data_polling_interval_s=simulated_acquirer.acquisition_interval_s, + ) + return video_mode_component + + +def get_voltage_control_component(video_mode_component, labels) -> VoltageControlComponent: + voltage_controller = video_mode_component.data_acquirer.get_voltage_control_component(labels=labels) + return voltage_controller + + +def main() -> None: + """ + Sets up logging, creates the VideoModeComponent, builds the dashboard, + and runs the Dash application server. + """ + # Configure logging for the application. + logger = setup_logging(__name__) + logger.info("Starting Video Mode application with SimulatedDataAcquirer (based on QDarts).") + + # Get the configured VideoModeComponent. + video_mode_component = get_video_mode_component() + logger.info( + f"VideoModeComponent instance created: {video_mode_component.component_id}" + ) + + # Get the VoltageControlComponent + labels = ["Gate 1 (x)", "Gate 2 (y)", "Sensor Gate", "Barrier Gate"] + voltage_control_component = get_voltage_control_component(video_mode_component, labels) + logger.info( + f"VoltageControlComponent instance created: {voltage_control_component.component_id}" + ) + + logger.info("Building the dashboard...") + + # Use build_dashboard to create the Dash app layout. + app = build_dashboard( + components=[video_mode_component,voltage_control_component], # List of dashboard components to include. + title="Video Mode Simulation (QDarts)", # Browser window title. + ) + + logger.info("Dashboard built. Starting Dash server on http://localhost:8050") + # Run the Dash server. + app.run( + debug=True, # Enables helpful Dash debugging features. + host="0.0.0.0", # Makes the server accessible on your local network. + port=8050, # Sets the server port. + use_reloader=False, # Often recommended for stability with background threads. + ) + + +if __name__ == "__main__": + # This ensures that main() is called only when the script is executed directly. + main() \ No newline at end of file diff --git a/examples/video_mode/example_video_mode_simulated.py b/examples/video_mode/example_video_mode_simulated.py new file mode 100644 index 0000000..1772471 --- /dev/null +++ b/examples/video_mode/example_video_mode_simulated.py @@ -0,0 +1,211 @@ +""" +Example Script: Video Mode with SimulatedDataAcquirer for a four dot transition with 2 sensors + +This script demonstrates how to use the VideoModeComponent with a SimulatedDataAcquirer. +The data is simulated using the QDarts package. No barrier gate is used in this example. + +This setup is ideal for simulating and testing video mode dashboards without needing +a live connection to an OPX or other hardware. It allows you to understand the +dashboard's functionality, test UI interactions, and develop custom components +in a controlled environment. + +Core Components Used: +- SweepAxis: Defines the parameters for each axis in the 2D scan (name, label, + units, span, number of points). +- SimulatedDataAcquirer: A data acquirer that generates simulated data based on QDarts for the 2D scan, + simulating a real data acquisition process. +- VideoModeComponent: The main Dash component that orchestrates the video mode + display, taking a data acquirer as input and rendering + the live plot and controls. +- build_dashboard: A utility function from qua_dashboards.core to construct + a Dash application layout with the provided components. + +How to Run: +1. Ensure you have `qua-dashboards` and its dependencies installed. + (e.g., `pip install qua-dashboards`) +2. Save this script as a Python file (e.g., `run_random_video_mode.py`). +3. Run the script from your terminal: `python run_random_video_mode.py` +4. Open your web browser and navigate to `http://127.0.0.1:8050/` (or the + address shown in your terminal). + +You should see a dashboard titled "Video Mode Simulation (Simulated Data)" +displaying a 2D plot that updates with new simulated data periodically. You will +also have controls to adjust the parameters of the X and Y axes (Span and Points) +and the SimulatedDataAcquirer (Software Averages, Simulated Acquire Time). +""" + +from qua_dashboards.core import build_dashboard +from qua_dashboards.utils import setup_logging +from qua_dashboards.video_mode import ( + SweepAxis, + SimulatedDataAcquirer, + VideoModeComponent, +) +from qdarts.experiment import Experiment +import numpy as np + +def get_video_mode_component() -> VideoModeComponent: + """ + Creates and returns a VideoModeComponent instance configured with a SimulatedDataAcquirer. + + This function encapsulates the setup of the sweep axes and the data acquirer, + making it easy to reuse this configuration or import it into other scripts. + + Returns: + VideoModeComponent: The configured video mode component. + """ + # Define the system + + #All capacitances are given in aF + N = 6 #number of dots + C_DD=20* np.eye((N))/2 #The self-capacitance of each dot, NOTE: factor of 2 due to symmetrization + C_DD[0,1] = 10 #capacitance between dot 0 and dot 1 (Left double dot) + C_DD[2,3] = 7 #capacitance between dot 3 and dot 4 (Right double dot) + + C_DD[0,4] = 1.6 #capacitance between sensor dot 4 and dot 0 + C_DD[1,4] = 1.4 #capacitance between sensor dot 4 and dot 1 + C_DD[2,5] = 1.4 #capacitance between sensor dot 5 and dot 2 + C_DD[3,5] = 2 #capacitance between sensor dot 5 and dot 3 + C_DD[1,2] = 6 #capacitance between the middle dots 2 and dot 3 + C_DD = C_DD + C_DD.T + + C_DG=11*np.eye(N) #dot-to-gate capacitances + #cross-capacitances + C_DG[0,1] = 1.5 #dot 0 from dot 1 + C_DG[1,0] = 1.2 #dot 1 from dot 0 + C_DG[2,3] = 1.3 #dot 2 from dot 3 + C_DG[3,2] = 1.4 #dot 3 from dot 3 + + # Definition of the tunnel couplings in eV + # NOTE: we use the convention that tc is the energy gap at avoided crossing H = tc/2 sx + tunnel_couplings = np.zeros((N,N)) + tunnel_couplings[0,1] = 50*1e-6 + tunnel_couplings[1,0] = 50*1e-6 + tunnel_couplings[2,3] = 60*1e-6 + tunnel_couplings[3,2] = 60*1e-6 + + # Experiment configurations + capacitance_config = { + "C_DD" : C_DD, #dot-dot capacitance matrix + "C_Dg" : C_DG, #dot-gate capacitance matrix + "ks" : 4, #distortion of Coulomb peaks. NOTE: If None -> constant size of Coublomb peak + } + + tunneling_config = { + "tunnel_couplings": tunnel_couplings, #tunnel coupling matrix + "temperature": 0.1, #temperature in Kelvin + "energy_range_factor": 5, #energy scale for the Hamiltonian generation. NOTE: Smaller -> faster but less accurate computation + } + sensor_config = { + "sensor_dot_indices": [4,5], #Indices of the sensor dots + "sensor_detunings": [-0.0005,-0.0005], #Detuning of the sensor dots + "noise_amplitude": {"fast_noise": 0.5*1e-6, "slow_noise": 1e-8}, #Noise amplitude for the sensor dots in eV + "peak_width_multiplier": 15, #Width of the sensor peaks in the units of thermal broadening m *kB*T/0.61. + } + + # Set up experiment + experiment = Experiment(capacitance_config, tunneling_config, sensor_config) + + # Arguments for the function that renders the capacitance CSD + unit = 'mV' + factor_mV_to_V = 1e-3 + span_x = 20 + span_y = 20 + points_x = 50 + points_y = 50 + + P=np.zeros((6,2)) + P[0,0]=1 + P[1,1]=1 + state_hint_lower_left = [1,1,0,0,3,3] + + args_sensor_scan_2D = { + "P": P, + "minV": [-span_x/2.*factor_mV_to_V,-span_y/2.*factor_mV_to_V], + "maxV": [ span_x/2.*factor_mV_to_V, span_y/2.*factor_mV_to_V], + "resolution": [points_x,points_y], + "state_hint_lower_left": state_hint_lower_left, + "cache": True, + "insitu_axis": None, + } + + # Define the X-axis for the 2D scan. + x_axis = SweepAxis( + name="x", # Internal identifier for the axis. + label="X Coordinate", # Display label in the UI. + units=unit, # Physical units of the axis. + span=span_x, # The total range of the X-axis sweep. + points=points_x, # The number of points (pixels) along the X-axis. + ) + + # Define the Y-axis for the 2D scan. + y_axis = SweepAxis( + name="y", # Internal identifier for the axis. + label="Y Coordinate", # Display label in the UI. + units=unit, # Physical units of the axis. + span=span_y, # The total range of the Y-axis sweep. + points=points_y, # The number of points (pixels) along the Y-axis. + ) + + # Instantiate the SimulatedDataAcquirer. + # This acquirer simulates data fetching by generating random 2D arrays. + simulated_acquirer = SimulatedDataAcquirer( + component_id="simulated-data-acquirer", # Unique ID for Dash elements. + x_axis=x_axis, + y_axis=y_axis, + experiment = experiment, + args_rendering = args_sensor_scan_2D, + conversion_factor_unit_to_volt=factor_mV_to_V, + SNR=20, # Signal-to-noise ratio on simulated images + acquire_time=0.1, # Simulated delay (seconds) for acquiring one raw frame. + num_software_averages=5, # Number of raw frames to average for display. + acquisition_interval_s=0.5, # Target time (seconds) between acquiring raw frames. + sensor_number=0, # Pick the first of the two defined sensors (0 or 1). + ) + + # Instantiate the VideoModeComponent. + # This is the main UI component that displays the live 2D plot and controls. + video_mode_component = VideoModeComponent( + component_id=VideoModeComponent.DEFAULT_COMPONENT_ID, # Uses a default ID. + data_acquirer=simulated_acquirer, # The source of the data. + # How often the frontend asks for new data (linked to acquirer's interval). + data_polling_interval_s=simulated_acquirer.acquisition_interval_s, + ) + return video_mode_component + + +def main() -> None: + """ + Sets up logging, creates the VideoModeComponent, builds the dashboard, + and runs the Dash application server. + """ + # Configure logging for the application. + logger = setup_logging(__name__) + logger.info("Starting Video Mode application with SimulatedDataAcquirer (based on QDarts).") + + # Get the configured VideoModeComponent. + video_mode_component = get_video_mode_component() + logger.info( + f"VideoModeComponent instance created: {video_mode_component.component_id}" + ) + + logger.info("Building the dashboard...") + # Use build_dashboard to create the Dash app layout. + app = build_dashboard( + components=[video_mode_component], # List of dashboard components to include. + title="Video Mode Simulation (QDarts)", # Browser window title. + ) + + logger.info("Dashboard built. Starting Dash server on http://localhost:8050") + # Run the Dash server. + app.run( + debug=True, # Enables helpful Dash debugging features. + host="0.0.0.0", # Makes the server accessible on your local network. + port=8050, # Sets the server port. + use_reloader=False, # Often recommended for stability with background threads. + ) + + +if __name__ == "__main__": + # This ensures that main() is called only when the script is executed directly. + main() \ No newline at end of file diff --git a/examples/video_mode/example_video_mode_simulated_two-dot-system.py b/examples/video_mode/example_video_mode_simulated_two-dot-system.py new file mode 100644 index 0000000..af4ff8f --- /dev/null +++ b/examples/video_mode/example_video_mode_simulated_two-dot-system.py @@ -0,0 +1,215 @@ +""" +Example Script: Video Mode with SimulatedDataAcquirer for a two dot system with one sensor + +This script demonstrates how to use the VideoModeComponent with a SimulatedDataAcquirer. +The data is simulated using the QDarts package. +This setup is ideal for simulating and testing video mode dashboards without needing +a live connection to an OPX or other hardware. It allows you to understand the +dashboard's functionality, test UI interactions, and develop custom components +in a controlled environment. + +Core Components Used: +- SweepAxis: Defines the parameters for each axis in the 2D scan (name, label, + units, span, number of points). +- SimulatedDataAcquirer: A data acquirer that generates simulated data based on QDarts for the 2D scan, + simulating a real data acquisition process. +- VideoModeComponent: The main Dash component that orchestrates the video mode + display, taking a data acquirer as input and rendering + the live plot and controls. +- build_dashboard: A utility function from qua_dashboards.core to construct + a Dash application layout with the provided components. + +How to Run: +1. Ensure you have `qua-dashboards` and its dependencies installed. + (e.g., `pip install qua-dashboards`) +2. Save this script as a Python file (e.g., `run_random_video_mode.py`). +3. Run the script from your terminal: `python run_random_video_mode.py` +4. Open your web browser and navigate to `http://127.0.0.1:8050/` (or the + address shown in your terminal). + +You should see a dashboard titled "Video Mode Simulation (Simulated Data)" +displaying a 2D plot that updates with new simulated data periodically. You will +also have controls to adjust the parameters of the X and Y axes (Span and Points) +and the SimulatedDataAcquirer (Software Averages, Simulated Acquire Time). +""" + +from qua_dashboards.core import build_dashboard +from qua_dashboards.utils import setup_logging +from qua_dashboards.video_mode import ( + SweepAxis, + SimulatedDataAcquirer, + VideoModeComponent, +) +from qdarts.experiment import Experiment +import numpy as np + +def get_video_mode_component() -> VideoModeComponent: + """ + Creates and returns a VideoModeComponent instance configured with a SimulatedDataAcquirer. + + This function encapsulates the setup of the sweep axes and the data acquirer, + making it easy to reuse this configuration or import it into other scripts. + + Returns: + VideoModeComponent: The configured video mode component. + """ + # Define the system + + #All capacitances are given in aF + N = 3 #number of dots + C_DD=20* np.eye((N))/ 2 #The self-capacitance of each dot, NOTE: factor of 2 due to symmetrization + C_DD[1,1] = 10 / 2 #NOTE: factor of 2 due to symmetrization + C_DD[0,1] = 5 #capacitance between dot 0 and dot 1 (Left double dot) + + C_DD[0,2] = 1.6/2 #capacitance between sensor dot 2 and dot 0 + C_DD[1,2] = 1.4/2 #capacitance between sensor dot 2 and dot 1 + C_DD = C_DD + C_DD.T + + C_DG=11*np.eye(N,N+1) #dot-to-gate capacitances, there is one barrier gate (index 3) + # cross-capacitances + C_DG[0,1] = 1.5 #dot 0 from gate 1 + C_DG[1,0] = 1.2 #dot 1 from gate 0 + C_DG[0,3] = 0 #dot 0 from barrier gate + C_DG[1,3] = 0 #dot 1 from barrier gate + + # Definition of the tunnel couplings in eV + # NOTE: we use the convention that tc is the energy gap at avoided crossing H = tc/2 sx + tunnel_couplings = np.zeros((N,N)) + tunnel_couplings[0,1] = 50*1e-6 + tunnel_couplings[1,0] = 50*1e-6 + + # Definition of barrier levels + barrier_levers = np.zeros((N,N,N+1)) # barrier between dot i and dot j is affected by gate k + barrier_levers[0,1,3] = 100*1e-6 + barrier_levers[1,0,3] = 100*1e-6 + barrier_levers = np.log(barrier_levers + 1.e-20) + + # Experiment configurations + capacitance_config = { + "C_DD" : C_DD, #dot-dot capacitance matrix + "C_Dg" : C_DG, #dot-gate capacitance matrix + "ks" : 4, #distortion of Coulomb peaks. NOTE: If None -> constant size of Coublomb peak + } + + tunneling_config = { + "tunnel_couplings": tunnel_couplings, #tunnel coupling matrix + "temperature": 0.1, #temperature in Kelvin + "energy_range_factor": 1, #energy scale for the Hamiltonian generation. NOTE: Smaller -> faster but less accurate computation + "barrier_levers": barrier_levers, #barrier levels matrix + } + + sensor_config = { + "sensor_dot_indices": [2], #Indices of the sensor dots + "sensor_detunings": [-0.0005], #Detuning of the sensor dots + "noise_amplitude": {"fast_noise": 0.5*1e-6, "slow_noise": 1e-8}, #Noise amplitude for the sensor dots in eV + "peak_width_multiplier": 15, #Width of the sensor peaks in the units of thermal broadening m *kB*T/0.61. + } + + # Set up experiment + experiment = Experiment(capacitance_config, tunneling_config, sensor_config) + + # Arguments for the function that renders the capacitance CSD + unit = 'mV' + factor_mV_to_V = 1e-3 + span_x = 20 + span_y = 20 + points_x = 50 + points_y = 50 + + P=np.zeros((N+1,2)) + P[0,0]=1 + P[1,1]=1 + state_hint_lower_left = [1,1,5] + + W = np.eye(2) + + args_sensor_scan_2D = { + "P": P, + "virtualisation_matrix": W, + "minV": [-span_x/2.*factor_mV_to_V,-span_y/2.*factor_mV_to_V], + "maxV": [ span_x/2.*factor_mV_to_V, span_y/2.*factor_mV_to_V], + "resolution": [points_x,points_y], + "state_hint_lower_left": state_hint_lower_left, + "cache": False, # Do not cache the results + "insitu_axis": None, + } + + # Define the X-axis for the 2D scan. + x_axis = SweepAxis( + name="x", # Internal identifier for the axis. + label="X Coordinate", # Display label in the UI. + units=unit, # Physical units of the axis. + span=span_x, # The total range of the X-axis sweep. + points=points_x, # The number of points (pixels) along the X-axis. + ) + + # Define the Y-axis for the 2D scan. + y_axis = SweepAxis( + name="y", # Internal identifier for the axis. + label="Y Coordinate", # Display label in the UI. + units=unit, # Physical units of the axis. + span=span_y, # The total range of the Y-axis sweep. + points=points_y, # The number of points (pixels) along the Y-axis. + ) + + # Instantiate the SimulatedDataAcquirer. + # This acquirer simulates data fetching by generating random 2D arrays. + simulated_acquirer = SimulatedDataAcquirer( + component_id="simulated-data-acquirer", # Unique ID for Dash elements. + x_axis=x_axis, + y_axis=y_axis, + experiment = experiment, + args_rendering = args_sensor_scan_2D, + conversion_factor_unit_to_volt=factor_mV_to_V, + SNR=20, # Signal-to-noise ratio on simulated images + acquire_time=0.1, # Simulated delay (seconds) for acquiring one raw frame. + num_software_averages=5, # Number of raw frames to average for display. + acquisition_interval_s=0.5, # Target time (seconds) between acquiring raw frames. + ) + + # Instantiate the VideoModeComponent. + # This is the main UI component that displays the live 2D plot and controls. + video_mode_component = VideoModeComponent( + component_id=VideoModeComponent.DEFAULT_COMPONENT_ID, # Uses a default ID. + data_acquirer=simulated_acquirer, # The source of the data. + # How often the frontend asks for new data (linked to acquirer's interval). + data_polling_interval_s=simulated_acquirer.acquisition_interval_s, + ) + return video_mode_component + + +def main() -> None: + """ + Sets up logging, creates the VideoModeComponent, builds the dashboard, + and runs the Dash application server. + """ + # Configure logging for the application. + logger = setup_logging(__name__) + logger.info("Starting Video Mode application with SimulatedDataAcquirer (based on QDarts).") + + # Get the configured VideoModeComponent. + video_mode_component = get_video_mode_component() + logger.info( + f"VideoModeComponent instance created: {video_mode_component.component_id}" + ) + + logger.info("Building the dashboard...") + # Use build_dashboard to create the Dash app layout. + app = build_dashboard( + components=[video_mode_component], # List of dashboard components to include. + title="Video Mode Simulation (QDarts)", # Browser window title. + ) + + logger.info("Dashboard built. Starting Dash server on http://localhost:8050") + # Run the Dash server. + app.run( + debug=True, # Enables helpful Dash debugging features. + host="0.0.0.0", # Makes the server accessible on your local network. + port=8050, # Sets the server port. + use_reloader=False, # Often recommended for stability with background threads. + ) + + +if __name__ == "__main__": + # This ensures that main() is called only when the script is executed directly. + main() \ No newline at end of file diff --git a/examples/video_mode/qua-dashboards.code-workspace b/examples/video_mode/qua-dashboards.code-workspace new file mode 100644 index 0000000..0e847bc --- /dev/null +++ b/examples/video_mode/qua-dashboards.code-workspace @@ -0,0 +1,21 @@ +{ + "folders": [ + { + "path": "../.." + }, + { + "path": "../../../../qua-dashboards-main" + } + ], + "settings": { + "terminal.integrated.env.osx": { + "PYTHONPATH": "${workspaceFolder}/src" + }, + "terminal.integrated.env.linux": { + "PYTHONPATH": "${workspaceFolder}/src" + }, + "terminal.integrated.env.windows": { + "PYTHONPATH": "${workspaceFolder}/src" + } + } +} \ No newline at end of file diff --git a/examples/voltage_control/example_voltage_control_simulated.py b/examples/voltage_control/example_voltage_control_simulated.py index 75b6207..4493d77 100644 --- a/examples/voltage_control/example_voltage_control_simulated.py +++ b/examples/voltage_control/example_voltage_control_simulated.py @@ -16,20 +16,23 @@ def define_gates_simple(): ] -def get_voltage_control_component(): +def get_voltage_control_component(callback=None): """Returns a VoltageControlComponent with demo gates.""" voltage_parameters = define_gates_simple() voltage_controller = VoltageControlComponent( component_id="v_ctrl", voltage_parameters=voltage_parameters, + callback_on_param_change = callback ) return voltage_controller +def example_callback(parameter, previous_value): + print("Parameter", parameter.name, "was changed from value ",previous_value," to value", parameter.get_latest()) def main(): logger = setup_logging(__name__) logger.info("Starting Voltage Control dashboard (Simple demo mode)") - component = get_voltage_control_component() + component = get_voltage_control_component(callback=example_callback) app = build_dashboard( components=[component], title="Voltage Control Dashboard (Simple Demo)", diff --git a/src/qua_dashboards/core/base_updatable_component.py b/src/qua_dashboards/core/base_updatable_component.py index 5c17c9f..596f3d3 100644 --- a/src/qua_dashboards/core/base_updatable_component.py +++ b/src/qua_dashboards/core/base_updatable_component.py @@ -17,6 +17,7 @@ class ModifiedFlags(Flag): PARAMETERS_MODIFIED = auto() PROGRAM_MODIFIED = auto() CONFIG_MODIFIED = auto() + PLOT_PARAMETERS_MODIFIED = auto() class BaseUpdatableComponent(BaseComponent, ABC): diff --git a/src/qua_dashboards/core/parameter_protocol.py b/src/qua_dashboards/core/parameter_protocol.py index 07420cc..0b485a1 100644 --- a/src/qua_dashboards/core/parameter_protocol.py +++ b/src/qua_dashboards/core/parameter_protocol.py @@ -11,6 +11,9 @@ class ParameterProtocol(Protocol): """ A protocol defining the expected interface for a generic parameter. + + This protocol follows exactly the semantics and names of qcodes.parameters.Parameter and therefore + any function expecting a class of this protocol is compatible with QCodes Parameters. """ @property @@ -24,7 +27,7 @@ def label(self) -> str: ... @property - def units(self) -> str: + def unit(self) -> str: """ A string indicating the physical units of the parameter (e.g., "V", "mV", "Hz"). @@ -33,7 +36,12 @@ def units(self) -> str: def get_latest(self) -> float: """ - Returns the latest known or actual value of the parameter. + Returns the latest known chached, or actual value of the parameter. + """ + ... + def get(self) -> float: + """ + Queries the current value of the parameter. """ ... diff --git a/src/qua_dashboards/utils/__init__.py b/src/qua_dashboards/utils/__init__.py index 95129ff..5ef6ffe 100644 --- a/src/qua_dashboards/utils/__init__.py +++ b/src/qua_dashboards/utils/__init__.py @@ -3,6 +3,7 @@ from .general_utils import * from .log_utils import * from .basic_parameter import * +from .callback_parameter import * __all__ = [ *data_utils.__all__, @@ -10,4 +11,5 @@ *general_utils.__all__, *log_utils.__all__, *basic_parameter.__all__, + *callback_parameter.__all__, ] diff --git a/src/qua_dashboards/utils/basic_parameter.py b/src/qua_dashboards/utils/basic_parameter.py index c9f8a78..ca8aa3e 100644 --- a/src/qua_dashboards/utils/basic_parameter.py +++ b/src/qua_dashboards/utils/basic_parameter.py @@ -1,34 +1,40 @@ import logging import time - - +from typing import Callable, Optional, TYPE_CHECKING +from qua_dashboards.core import ParameterProtocol __all__ = ["BasicParameter"] class BasicParameter: - def __init__(self, name, label=None, units="V", initial_value=0.0): + def __init__(self, name:str, label:Optional[str]=None, unit:str="V", initial_value:float=0.0, units:Optional[str]=None): self.name = name - self.label = label + self.label = label if label is not None else name self.latest_value = initial_value self._value = initial_value - self.units = units + self.unit = unit + if units is not None: + logging.warning("The use of parameter 'units' is deprecated. Use 'unit' instead.") + self.unit = units logging.debug( - f"{self.name} initialized with value {self.latest_value} {self.units}" + f"{self.name} initialized with value {self.latest_value} {self.unit}" ) def get(self): - time.sleep(0.2) # Simulate a 200ms delay self.latest_value = self._value - logging.debug(f"Getting {self.name}: {self.latest_value} {self.units}") + logging.debug(f"Getting {self.name}: {self.latest_value} {self.unit}") return self.latest_value def set(self, new_value): self._value = new_value updated_value = self.get() # Return the value after setting logging.debug( - f"Setting {self.name} to {new_value}: Actual value is {updated_value} {self.units}" + f"Setting {self.name} to {new_value}: Actual value is {updated_value} {self.unit}" ) return updated_value def get_latest(self): return self.latest_value + +if TYPE_CHECKING: + def check(name: str): + _:ParameterProtocol = BasicParameter(name) \ No newline at end of file diff --git a/src/qua_dashboards/utils/callback_parameter.py b/src/qua_dashboards/utils/callback_parameter.py new file mode 100644 index 0000000..3def626 --- /dev/null +++ b/src/qua_dashboards/utils/callback_parameter.py @@ -0,0 +1,47 @@ + +from qua_dashboards.core import ParameterProtocol +from typing import Callable, TYPE_CHECKING + +__all__ = ["CallbackParameter"] + + +class CallbackParameter: + """ Calls a callback function when the value of this parameter is changed. The callback takes the parameter changed as argument.""" + + def __init__(self, parameter: ParameterProtocol, callback_on_set: Callable[[ParameterProtocol, float],None]): + """ Wraps the given parameter with a class that calls a given callback when set() is called. + The callback is run after parameter.set() is called. + + Parameters: + parameter: the parameter to which to add the callback + callback_on_set: A callback that takes as argument parameter and the old value + """ + self.parameter = parameter + self.callback = callback_on_set + + @property + def name(self) -> str: + return self.parameter.name + + @property + def label(self) -> str: + return self.parameter.label + + @property + def unit(self) -> str: + return self.parameter.unit + + def get_latest(self) -> float: + return self.parameter.get_latest() + + def get(self) -> float: + return self.parameter.get() + + def set(self, value: float) -> None: + old_value = self.parameter.get_latest() + self.parameter.set(value) + self.callback(self.parameter,old_value) # Calls the callback with the parameter and the old value every time set() is called! + +if TYPE_CHECKING: + def check(parameter: ParameterProtocol, callback_on_set: Callable[[ParameterProtocol], None]): + _: ParameterProtocol = CallbackParameter(parameter, callback_on_set) \ No newline at end of file diff --git a/src/qua_dashboards/video_mode/data_acquirers/__init__.py b/src/qua_dashboards/video_mode/data_acquirers/__init__.py index d288b53..3a97045 100644 --- a/src/qua_dashboards/video_mode/data_acquirers/__init__.py +++ b/src/qua_dashboards/video_mode/data_acquirers/__init__.py @@ -3,10 +3,12 @@ RandomDataAcquirer, ) from qua_dashboards.video_mode.data_acquirers.opx_data_acquirer import OPXDataAcquirer +from qua_dashboards.video_mode.data_acquirers.simulated_data_acquirer import SimulatedDataAcquirer __all__ = [ "BaseDataAcquirer", "RandomDataAcquirer", "OPXDataAcquirer", + "SimulatedDataAcquirer", ] diff --git a/src/qua_dashboards/video_mode/data_acquirers/base_data_acquirer.py b/src/qua_dashboards/video_mode/data_acquirers/base_data_acquirer.py index cb2b8de..511aafe 100644 --- a/src/qua_dashboards/video_mode/data_acquirers/base_data_acquirer.py +++ b/src/qua_dashboards/video_mode/data_acquirers/base_data_acquirer.py @@ -261,6 +261,13 @@ def update_parameters(self, parameters: Dict[str, Dict[str, Any]]) -> ModifiedFl if component.component_id in parameters: flags |= component.update_parameters(parameters) + if flags & ModifiedFlags.PLOT_PARAMETERS_MODIFIED: + logger.debug("PLOT parameters were modified!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!") + # Clear history as plotting parameters change + with self._data_lock: + self._data_history_raw.clear() + self._latest_processed_data = None + return flags def get_dash_components(self, include_subcomponents: bool = True) -> List[Any]: diff --git a/src/qua_dashboards/video_mode/data_acquirers/simulated_data_acquirer.py b/src/qua_dashboards/video_mode/data_acquirers/simulated_data_acquirer.py new file mode 100644 index 0000000..a7f721d --- /dev/null +++ b/src/qua_dashboards/video_mode/data_acquirers/simulated_data_acquirer.py @@ -0,0 +1,345 @@ +import logging +import copy +from time import sleep +from typing import Any, Dict, List, Sequence + +import numpy as np +from dash import html + +from qua_dashboards.core import ModifiedFlags, ParameterProtocol +from qua_dashboards.utils.dash_utils import create_input_field +from qua_dashboards.video_mode.data_acquirers.base_2d_data_acquirer import ( + Base2DDataAcquirer, +) +from qua_dashboards.video_mode.sweep_axis import SweepAxis +from qdarts.experiment import Experiment + +from qua_dashboards.voltage_control import VoltageControlComponent +from qua_dashboards.utils import BasicParameter + +logger = logging.getLogger(__name__) + +__all__ = ["SimulatedDataAcquirer"] + + +class SimulatedDataAcquirer(Base2DDataAcquirer): + """Data acquirer that generates simulated 2D data using the simulation package QDarts. + + Inherits from Base2DDataAcquirer and simulates a delay for data acquisition. + """ + + def __init__( + self, + *, + x_axis: SweepAxis, + y_axis: SweepAxis, + experiment: Experiment, + args_rendering: dict[str, Any], + conversion_factor_unit_to_volt: float, + SNR: float = 20, # signal-to-noise ratio + component_id: str = "simulated-data-acquirer", + acquire_time: float = 0.05, # Simulate 50ms acquisition time per frame + sensor_number: int = 0, # Default to first sensor, can only be different from 0 if there are multiple sensors + # Other parameters like num_software_averages, acquisition_interval_s + # are passed via **kwargs to Base2DDataAcquirer and then to BaseDataAcquirer. + **kwargs: Any, + ) -> None: + """Initializes the SimulatedDataAcquirer. + + Args: + component_id: Unique ID for Dash elements. + x_axis: The X sweep axis. + y_axis: The Y sweep axis. + args_rendering: The arguments used for the rendering function of the QDarts simulator. + SNR: Signal-to-noise ratio used for adding noise to the simulated image. + acquire_time: Simulated time in seconds to 'acquire' one raw data frame. + **kwargs: Additional arguments for Base2DDataAcquirer, including + num_software_averages and acquisition_interval_s for + BaseDataAcquirer. + """ + self.acquire_time: float = acquire_time + self.experiment: Experiment = experiment + self.args_rendering: dict[str, Any] = args_rendering + self.conversion_factor_unit_to_volt: float = conversion_factor_unit_to_volt + self.SNR: float = SNR + self.simulated_image: np.ndarray = None + self.voltage_parameters: Sequence[ParameterProtocol] = None + self._last_voltage_parameters: Sequence[ParameterProtocol] = None + self.m = None + self._first_acquisition: bool = True + self._plot_parameters_changed: bool = False + self._voltage_control_component: bool = False + if self.experiment.sensor_config["sensor_dot_indices"] is not None and 0 <= sensor_number < len(self.experiment.sensor_config["sensor_dot_indices"]): + self.sensor_number = sensor_number + else: + raise ValueError(f"Invalid sensor number {sensor_number}.") + logger.debug( + f"Initializing SimulatedDataAcquirer (ID: {component_id}) with " + f"acquire_time: {self.acquire_time}s" + ) + super().__init__( + component_id=component_id, x_axis=x_axis, y_axis=y_axis, **kwargs + ) + + + def _initialize_m(self) -> None: + """Initializes the m parameter and adds it to the rendering arguments.""" + self.m = copy.deepcopy(self.experiment.tunneling_sim.boundaries(self.args_rendering["state_hint_lower_left"]).point_inside) # Deepcopy needed that self.m does not have the same reference as polytope.point_inside + logger.debug(f"initial m: {self.m}") + + + def get_voltage_control_component(self, labels = None) -> VoltageControlComponent: + """ Creates and returns a voltage control component for the simulated data acquirer. + The voltage values are set to the initial guess for m. + + Arguments: + Labels: They are used to set the names of the voltage parameters. + Default is None, which means that the default labels are used. + """ + logger.debug(f"Creating VoltageControlComponent with labels: {labels}") + + self._voltage_control_component = True + self._initialize_m() + + if labels is not None and len(labels) is not self.args_rendering["P"].shape[0]: + raise ValueError( + f"Number of labels ({len(labels)}) does not match number of voltage parameters " + f"({self.args_rendering['P'].shape[0]})." + ) + + if labels is not None: + voltage_parameters = [ + BasicParameter(f"vg{i+1}", label, "mV", initial_value=self.m[i] * 1./self.conversion_factor_unit_to_volt) + for i, label in enumerate(labels) + ] + else: + # If no labels are provided, use default labels + logger.info(f"No labels are provided: default labels are used.") + voltage_parameters = [ + BasicParameter(f"vg{i+1}", f"vg{i+1}", "mV", initial_value=self.m[i] * 1./self.conversion_factor_unit_to_volt) + for i in range(self.args_rendering["P"].shape[0]) + ] + + self.voltage_parameters = voltage_parameters + self._last_voltage_parameters = copy.deepcopy(voltage_parameters) # Store the initial voltage parameters + + def callback(parameter, previous_value): + logger.info(f"Parameter {parameter.name} was changed from value {previous_value} to value {parameter.get_latest()}") + + # Update m + self._last_voltage_parameters = copy.deepcopy(self.voltage_parameters) # Update the last voltage parameters + for i in range(len(self.voltage_parameters)): + self.m[i] = self.voltage_parameters[i].get_latest() * self.conversion_factor_unit_to_volt + logger.info(f"Voltage parameters changed, m updated: {self.m}") + # Clear the data history + with self._data_lock: + self._data_history_raw.clear() + self._latest_processed_data = None + + # Set the flag that the plot parameters were changed + self._plot_parameters_changed = True # This will trigger a new simulated image in perform_actual_acquisition() + + # Get the VoltageControlComponent + voltage_controller = VoltageControlComponent( + component_id="voltage_control", + voltage_parameters=self.voltage_parameters, + callback_on_param_change=callback, # Callback function to handle parameter changes + ) + + return voltage_controller + + + def set_virtualisation_matrix(self, new_W): + self._plot_parameters_changed = True + self.args_rendering["virtualisation_matrix"] = new_W + + + def get_virtualisation_matrix(self): + return self.args_rendering["virtualisation_matrix"] + + + def perform_actual_acquisition(self) -> np.ndarray: + """Simulates data acquisition by sleeping and returning simulated data. + + This method is called by the background thread in BaseDataAcquirer. + + Returns: + A 2D numpy array of simulated float values between 0 and 1, with + dimensions (y_axis.points, x_axis.points). + """ + + def generate_simulated_image() -> np.ndarray: + logger.debug( + f"SimulatedDataAcquirer (ID: {self.component_id}): " + f"Generating simulated data for {self.y_axis.points}x{self.x_axis.points}" + ) + + # Ensure y_axis.points and x_axis.points are positive integers + if self.y_axis.points <= 0 or self.x_axis.points <= 0: + logger.warning( + f"SimulatedDataAcquirer (ID: {self.component_id}): Invalid points " + f"({self.y_axis.points}x{self.x_axis.points}). Returning empty array." + ) + return np.array([[]]) # Return a 2D empty array to avoid downstream errors + + # Generate the simulated image + logger.info("Generating simulated data") + state = self.experiment.tunneling_sim.poly_sim.find_state_of_voltage(v = self.m, + state_hint = self.args_rendering["state_hint_lower_left"]) + sliced_sim = self.experiment.tunneling_sim.slice(P = self.args_rendering["P"]@self.args_rendering["virtualisation_matrix"].T, m = self.m) + sensor_signalexp = sliced_sim.sensor_scan_2D(P = np.eye(2), + m = np.zeros(2), + minV = self.args_rendering["minV"], + maxV = self.args_rendering["maxV"], + resolution = self.args_rendering["resolution"], + state_hint_lower_left = state) + + # In case of several sensors, pick the sensor with the given sensor_number (default is 0) + if sensor_signalexp.ndim == 3: + sensor_values = sensor_signalexp[:,:,self.sensor_number].T + else: + sensor_values = sensor_signalexp.T + + self.simulated_image = sensor_values + + # First acquisition + if self._first_acquisition: + logger.info("First acquisition, generating simulated image") + self._first_acquisition = False + if not self._voltage_control_component: + self._initialize_m() + logger.info(f"First acquisition, initial rendering arguments: {self.args_rendering}") + generate_simulated_image() + + # Plot parameters changed (points, span, voltage parameters, etc.) + if self._plot_parameters_changed: + self._plot_parameters_changed = False # Do this here to PREVENT race conditions! It takes long to simulate the new image, and every change of plotting parameters would lead to race conditions. + generate_simulated_image() + + else: + sleep(self.acquire_time) # Simulate acquisition time + + # Add uniformly distributed noise to the simulated image + logger.info(f"Add noise to the image (SNR = {self.SNR})") + mean_signal = np.mean(self.simulated_image) + factor = np.sqrt(12)*mean_signal/self.SNR + noise = np.random.rand(self.y_axis.points, self.x_axis.points)*factor + + # Check if shape of axes still match the shape of the image + if self.simulated_image.shape[0] != self.y_axis.points or self.simulated_image.shape[1] != self.x_axis.points: # race conditions + logger.warning("Axes resolution changed during rendering !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!") + logger.info(f"signal shape: {self.simulated_image.shape}") + logger.info(f"(y_points x x_points): ({self.y_axis.points}x{self.x_axis.points}). Returning empty array.") + # logger.info(f"y_points: {self.y_axis.points}") + # logger.info(f"x_points: {self.x_axis.points}") + return np.empty((self.y_axis.points, self.x_axis.points)) # Return a 2D empty array with the correct shape to avoid downstream errors + else: + return self.simulated_image + noise + + + def get_dash_components(self, include_subcomponents: bool = True) -> List[html.Div]: + """Returns Dash UI components for configuring SimulatedDataAcquirer. + + Extends the components from Base2DDataAcquirer (which includes + BaseDataAcquirer's components) with a field for 'acquire_time'. + + Args: + include_subcomponents: Whether to include UI components from + parent classes. Defaults to True. + + Returns: + A list of Dash html.Div components. + """ + dash_components = super().get_dash_components( + include_subcomponents=include_subcomponents + ) + + # UI for acquire_time + dash_components.append( + html.Div( + create_input_field( + # Use self._get_id() for namespaced ID + id=self._get_id("acquire-time"), + label="Simulated Acquire Time", + value=self.acquire_time, + min=0.0, + max=10.0, # Max 10 seconds simulation + step=0.01, + units="s", + debounce=True, # Useful for number inputs + type="number", # Explicitly set type + ) + ) + ) + return dash_components + + def ramp(self, v_start, v_end, N): + """ + Generates a linear ramp from v_start to v_end with the specified resolution. + + Parameters: + v_start (np.array): Starting voltage vector. + v_end (np.array): Ending voltage vector. + state_hint (list): State hint for the tunneling simulation. + N (int): Number of points in the ramp (resolution). + + Returns: + 1D numpy array of voltage vectors along the ramp. + """ + state_hint = self.args_rendering["state_hint_lower_left"] + sensor_signal = self.experiment.tunneling_sim.sensor_scan( + v_start = v_start, + v_end = v_end, + resolution = N, + v_start_state_hint = state_hint, + ) + logger.info(f"!!!!!!!!!!!!!!!!!!!!!!!!!!!! Ramp from {v_start} to {v_end} with {N} points generated.") + #logger.info(f"!!!!!!!!!!!!!!!!!!!!!!!!!!!! Sensor signal: {sensor_signal[:, 0]}") + return sensor_signal[:, 0] # Return the sensor signal as a 1D numpy array ### DISCUSS: Transform coordinates ??? + + def update_parameters(self, parameters: Dict[str, Dict[str, Any]]) -> ModifiedFlags: + """Updates SimulatedDataAcquirer parameters based on UI input. + + Args: + parameters: A dictionary where keys are component IDs and values are + dictionaries of parameter names to their new values. + + Returns: + ModifiedFlags indicating what aspects of the component were changed. + """ + flags = super().update_parameters(parameters) + + if flags & ModifiedFlags.PLOT_PARAMETERS_MODIFIED: + # Update the rendering arguments + logger.info("Updating the rendering arguments") + span_x = parameters["x-axis"]["span"] + points_x = parameters["x-axis"]["points"] + span_y = parameters["y-axis"]["span"] + points_y = parameters["y-axis"]["points"] + self.args_rendering["minV"] = [-span_x/2.*self.conversion_factor_unit_to_volt,-span_y/2.*self.conversion_factor_unit_to_volt] + self.args_rendering["maxV"] = [ span_x/2.*self.conversion_factor_unit_to_volt, span_y/2.*self.conversion_factor_unit_to_volt] + self.args_rendering["resolution"] = [points_x,points_y] + + self._plot_parameters_changed = True # Does not really matter whether here or above... + # Here: plot parameters are changed, perform_actual_acquisition() might be called before _plot_parameters_changed is set to True --> an empty array will be returned + # Above: _plot_parameters_changed is set to True, perform_actual_acquisition() might be called before the plot parameters were changed --> new simulated image base on the old plot parameters, _plot_parameters_changed set to False --> empty arrays returned afterwards until a plot parameter is changed + + # Check if parameters for this specific component_id are present + if self.component_id in parameters: + params = parameters[self.component_id] + logger.debug(f"params: {params}") + if "acquire-time" in params: + new_acquire_time = float(params["acquire-time"]) + if self.acquire_time != new_acquire_time and new_acquire_time >= 0: + self.acquire_time = new_acquire_time + flags |= ModifiedFlags.PARAMETERS_MODIFIED + logger.debug( + f"SimulatedDataAcquirer (ID: {self.component_id}): " + f"Updated acquire_time to {self.acquire_time}s" + ) + elif new_acquire_time < 0: + logger.warning( + f"SimulatedDataAcquirer (ID: {self.component_id}): " + f"Invalid acquire_time ({new_acquire_time}s) received. Not updated." + ) + return flags diff --git a/src/qua_dashboards/video_mode/sweep_axis.py b/src/qua_dashboards/video_mode/sweep_axis.py index 25466c3..0a2d47f 100644 --- a/src/qua_dashboards/video_mode/sweep_axis.py +++ b/src/qua_dashboards/video_mode/sweep_axis.py @@ -1,3 +1,4 @@ +import logging from typing import Optional, Dict, Any from dash import Dash @@ -9,6 +10,8 @@ from qua_dashboards.utils.basic_parameter import BasicParameter from qua_dashboards.utils.dash_utils import create_input_field +logger = logging.getLogger(__name__) + __all__ = ["SweepAxis"] @@ -140,7 +143,9 @@ def update_parameters(self, parameters: Dict[str, Dict[str, Any]]) -> ModifiedFl if "span" in params and self.span != params["span"]: self.span = params["span"] flags |= ModifiedFlags.PARAMETERS_MODIFIED | ModifiedFlags.PROGRAM_MODIFIED + flags |= ModifiedFlags.PLOT_PARAMETERS_MODIFIED if "points" in params and self.points != params["points"]: self.points = params["points"] flags |= ModifiedFlags.PARAMETERS_MODIFIED | ModifiedFlags.PROGRAM_MODIFIED + flags |= ModifiedFlags.PLOT_PARAMETERS_MODIFIED return flags diff --git a/src/qua_dashboards/video_mode/tab_controllers/annotation_tab_controller.py b/src/qua_dashboards/video_mode/tab_controllers/annotation_tab_controller.py index 1a1981e..be3ca07 100644 --- a/src/qua_dashboards/video_mode/tab_controllers/annotation_tab_controller.py +++ b/src/qua_dashboards/video_mode/tab_controllers/annotation_tab_controller.py @@ -2,12 +2,13 @@ import logging from typing import Dict, Any, List, Optional, Tuple import re +import copy import dash import dash_bootstrap_components as dbc import numpy as np import xarray as xr -from dash import Dash, Input, Output, State, html, ctx +from dash import Dash, Input, Output, State, html, ctx #, dcc from dash.exceptions import PreventUpdate from plotly import graph_objects as go @@ -19,9 +20,15 @@ from qua_dashboards.video_mode.utils.dash_utils import xarray_to_plotly from qua_dashboards.video_mode.utils.annotation_utils import ( calculate_slopes, + calculate_normals, find_closest_line_id, + compute_transformation_matrix_from_image_gradients, + compute_transformation_matrix, + warp_image_with_normals, ) from qua_dashboards.video_mode.utils.data_utils import load_data +from qua_dashboards.video_mode.data_acquirers.simulated_data_acquirer import SimulatedDataAcquirer +from qua_dashboards.video_mode.utils.config import TransformationMatrixConfig logger = logging.getLogger(__name__) @@ -76,10 +83,14 @@ class AnnotationTabController(BaseTabController): _LOAD_FROM_DISK_BUTTON_SUFFIX = "load-from-disk-button" _LOAD_FROM_DISK_INPUT_SUFFIX = "load-from-disk-input" _SHOW_LABELS_CHECKLIST_SUFFIX = "show-labels-checklist" + _GRADIENT_COMPUTATION_BUTTON_SUFFIX = "gradient-computation-button" + _GRADIENT_COMPUTATION_RESULTS_SUFFIX = "gradient-computation-results" + #_STORE_LIVE_FRAME_SUFFIX = "trigger-input-live-frame" def __init__( self, component_id: str = "annotation-tab-controller", + data_acquirer: Optional[Any] = None, point_select_tolerance: float = 0.025, # Relative **kwargs: Any, ) -> None: @@ -92,6 +103,7 @@ def __init__( **kwargs: Additional keyword arguments passed to BaseComponent. """ super().__init__(component_id=component_id, **kwargs) + self.data_acquirer = data_acquirer self._relative_click_tolerance = max(0.001, point_select_tolerance) self._absolute_click_tolerance: float = 1.0 # Default, updated with figure @@ -242,7 +254,7 @@ def get_layout(self) -> html.Div: html.Hr(), html.H6("Analysis"), dbc.Button( - "Calculate Slopes", + "Calculate Slopes of Added Lines", id=self._get_id(self._CALCULATE_BUTTON_SUFFIX), color="info", size="sm", @@ -257,6 +269,23 @@ def get_layout(self) -> html.Div: "fontSize": "0.8em", }, ), + dbc.Button( + "Compute Transformation Matrix", + id=self._get_id(self._GRADIENT_COMPUTATION_BUTTON_SUFFIX), + color="danger", + size="sm", + className="mb-2", + ), + html.Pre( + id=self._get_id(self._GRADIENT_COMPUTATION_RESULTS_SUFFIX), + className="border rounded analysis-results-dark p-1 mb-3", + style={ + "maxHeight": "100px", + "overflowY": "auto", + "fontSize": "0.8em", + }, + ), + # dcc.Store(id=self._get_id(self._STORE_LIVE_FRAME_SUFFIX), data=False), ] ) @@ -422,7 +451,71 @@ def register_callbacks( self._register_mode_change( app, viewer_ui_state_store_id ) + self._register_compute_transformation_matrix_from_gradients( + app, viewer_data_store_id + ) + + def _register_compute_transformation_matrix_from_gradients( + self, + app:Dash, + viewer_data_store_id: Dict[str, str] + ) -> None: + """ + Callback to compute the transformation matrix (compensated --> original coordinates) using Gaussian Mixture Models (GMM) on the gradients. + In case data is acquired by the SimulatedDataAcquirer, the simulators virtualisation matrix is updated, such that new data is acquired in the transformed coordinates. + """ + + @app.callback( + Output(self._get_id(self._GRADIENT_COMPUTATION_RESULTS_SUFFIX), "children"), +# Output(self._get_id(self._STORE_LIVE_FRAME_SUFFIX),"data"), + Input(self._get_id(self._GRADIENT_COMPUTATION_BUTTON_SUFFIX), "n_clicks"), + State(viewer_data_store_id, "data"), +# State(self._get_id(self._STORE_LIVE_FRAME_SUFFIX), "data"), + prevent_initial_call=True, + ) + def _compute_transformation_matrix_from_gradients(n_clicks: int, current_viewer_data_ref: Optional[Dict[str, str]]): #, trigger: bool): + logger.info(f"{self._get_id(self._GRADIENT_COMPUTATION_BUTTON_SUFFIX)}: Compute transformation matrix clicked.") + + # Parameters from config.py in utils + try: + cfg = TransformationMatrixConfig() + except ValueError as e: + return f"Value error: {e}" + logger.info(f"Config: {cfg}") + + # Get current base image + if ( + not current_viewer_data_ref + or current_viewer_data_ref.get("key") != data_registry.STATIC_DATA_KEY + ): + return "Analysis requires static data to be active." + static_data_object = data_registry.get_data(data_registry.STATIC_DATA_KEY) + if not static_data_object or static_data_object.get("base_image_data") is None: + return "No image data found in static data for analysis." + + image_data = static_data_object.get("base_image_data") + + # Compute the normals + p1, p2, m1, m2 = compute_transformation_matrix_from_image_gradients(image_data.values, cfg) + # Warp the image --> Creates a plot to check whether the transformation matrix is computed correctly + warp_image_with_normals(image_data.values,p1,p2) + + # Compute the transformation matrix + A_inv, A = compute_transformation_matrix(p1,p2) + W_new = self.data_acquirer.get_virtualisation_matrix() @ A_inv # old virtualisation matrix W @ new transformation matrix A_inv + W_new_formatted = np.array2string(W_new, precision=4, suppress_small=True) + output = f"Transformation matrix (compensated --> original coords):\n {W_new_formatted}" + + # Apply transformation to the simulated data (does not work for other data acquirers yet) + # Note: The live data is NOT automatically imported in the annotation tab! + # Note: The axes are NOT transformed. + if isinstance(self.data_acquirer, SimulatedDataAcquirer): + logger.info(f"Set virtualisation matrix in simulated data acquirer to {W_new}") + self.data_acquirer.set_virtualisation_matrix(W_new) + + return output #, not trigger + def _register_mode_change( self, app:Dash, @@ -461,12 +554,17 @@ def _register_import_live_frame_callback( Output(viewer_data_store_id, "data"), Output(viewer_ui_state_store_id, "data"), Input(self._get_id(self._IMPORT_LIVE_FRAME_BUTTON_SUFFIX), "n_clicks"), + # Input(self._get_id(self._STORE_LIVE_FRAME_SUFFIX),"data"), State(latest_processed_data_store_id, "data"), prevent_initial_call=True, ) + # def _import_live_frame( + # n_clicks: int, trigger: bool, live_data_ref: Optional[Dict[str, Any]] + # ) -> Tuple[Dict[str, Any],Dict[str, Any]]: def _import_live_frame( n_clicks: int, live_data_ref: Optional[Dict[str, Any]] ) -> Tuple[Dict[str, Any],Dict[str, Any]]: + if live_data_ref is None: logger.warning("Import Live Frame: No live data reference found.") raise PreventUpdate @@ -890,7 +988,10 @@ def _clear_annotations( def _register_analysis_callback( self, app: Dash, viewer_data_store_id: Dict[str, str] ) -> None: - """Callback for performing analysis (e.g., calculating slopes).""" + """ + Callback for performing analysis (e.g., calculating slopes). + In the slope analysis (from annotated lines), the transformation matrix (compensated --> original coords) is computed in case of exactly 2 annotated lines. + """ @app.callback( Output(self._get_id(self._ANALYSIS_RESULTS_SUFFIX), "children"), @@ -913,8 +1014,22 @@ def _run_slope_analysis( annotations_data = static_data_object["annotations"] slopes = calculate_slopes(annotations_data) + normals = calculate_normals(annotations_data) if slopes: - return json.dumps(slopes, indent=2) + # Exactly 2 annotated lines: Compute also the transformation matrix + if normals and len(normals)==2: + # order slopes and normals in the same way + keys = list(slopes.keys()) + s_vals = [slopes[k] for k in keys] + n_vals = [normals[k] for k in keys] + # ensure that first normal is aligned with x-axis, and second normal with y-axis --> in order to be consistent with the button Compute Transformation Matrix + if abs(s_vals[0]) < abs(s_vals[1]): + n_vals = [n_vals[1], n_vals[0]] + A_inv,A = compute_transformation_matrix(*n_vals) + return f"Transformation matrix (compensated --> original coords):\n {np.array2string(A_inv, precision=4, suppress_small=True)} \n\nSlopes:\n" + json.dumps(slopes, indent=2) + # Fewer or more than 2 annotated lines + else: + return json.dumps(slopes, indent=2) return "No lines to analyze or error in calculation." def _register_load_from_disk_callback( diff --git a/src/qua_dashboards/video_mode/utils/annotation_utils.py b/src/qua_dashboards/video_mode/utils/annotation_utils.py index a1c2c28..bbbbd60 100644 --- a/src/qua_dashboards/video_mode/utils/annotation_utils.py +++ b/src/qua_dashboards/video_mode/utils/annotation_utils.py @@ -13,6 +13,23 @@ import numpy as np import plotly.graph_objects as go +from dataclasses import dataclass # dataclass decorator: automatically generates special methods (e.g. __init__, __repr__, __eq__, etc.) for classes primarily used to store data +from typing import Callable +import autograd.numpy as anp +import autograd.scipy.stats as stat +from autograd import value_and_grad +import scipy.optimize +from scipy.interpolate import interpn +from scipy.optimize import minimize +from scipy.ndimage import gaussian_filter1d +import matplotlib.pyplot as plt +import cv2 as cv +import autograd.numpy as anp +from autograd import value_and_grad +import matplotlib.pyplot as plt +from scipy.interpolate import interpn + + logger = logging.getLogger(__name__) __all__ = [ @@ -21,6 +38,10 @@ "find_closest_point_id", "find_closest_line_id", "calculate_slopes", + "calculate_normals", + "compute_transformation_matrix_from_image_gradients", + "compute_transformation_matrix", + "warp_image_with_normals" ] @@ -319,3 +340,350 @@ def calculate_slopes( logger.info(f"Calculated slopes for {len(slopes)} lines.") return slopes + + +def calculate_normals(annotations_data: Dict[str, List[Dict[str, Any]]], +) -> Dict[str, np.ndarray]: + """ + Calculate lines of annotated lines. + + Args: + annotations_data The main annotations data structure. + Expected: {"points": [...], "lines": [...]} + + Returns: + normals A dictionary mapping line ID (string) to its normal (np.array). + """ + normals: Dict[str, np.ndarray] = {} + lines = annotations_data.get("lines", []) + + for line in lines: + line_id = line["id"] + coords1 = get_point_coords_by_id(annotations_data, line["start_point_id"]) + coords2 = get_point_coords_by_id(annotations_data, line["end_point_id"]) + + x1, y1 = coords1 + x2, y2 = coords2 + delta_x = x2 - x1 + delta_y = y2 - y1 + n = np.array([-delta_y,delta_x], dtype=float) + n = n/np.linalg.norm(n) + normals[line_id] = n + + logger.info(f"Calculated normals for {len(normals)} lines.") + return normals + + +def subtract_low_norm_mean(img, frac=0.7): + """ + Subtract the mean of the pixels with smallest norm from the input image + + Args: + img Input image + frac Fraction of pixels to use for mean computation (i.e. with smallest norm) + + Return: + Centered image Input image minus mean of pixels with smallest norm + """ + pixels = img.reshape(-1,1) + norm = np.linalg.norm(pixels,axis=1) + threshold = np.quantile(norm, frac) + mask = norm <= threshold # mask: the fraction of pixels used for the mean computation + mean_value = pixels[mask].mean() + return img - mean_value + + +def image_gradients(img, cfg): + """ + Compute the gradients in x- and y-direction of an image using Sobel filters. + Apply Gaussian blurring before computing the gradients. Subtract the mean of pixels with small norm from the gradient images. + + Args: + img Input image I of shape (H,W) + cfg Config parameters + sigmaX_blur Standard deviation of the Gaussian in x direction used for blurring the image + sigmaY_blur Standard deviation of the Gaussian in y direction used for blurring the image + ksize_sobelX Kernel size for computing the gradients in x-direction. Note: Has to be ODD number. + ksize_sobelY Kernel size for computing the gradients in x-direction. Note: Has to be ODD number. + frac Fraction of pixels to use for mean computation + + Returns: + sobel_x Image Ix of gradients in x-direction of shape (H,W) + sobel_y Image Iy of gradients in y-direction of shape (H,W) + """ + + # Apply Gaussian blur to the image + blurred_img = cv.GaussianBlur(img, (0, 0), cfg.sigmaX_blur, cfg.sigmaY_blur) # (0,0) means that the kernel size is automatically computed based on sigmaX and sigmaY; sigmaX=sigmaY + + # Apply Sobel filter to compute gradients + sobel_x = cv.Sobel(blurred_img, cv.CV_64F, 1, 0, ksize=cfg.ksize_sobelX) # Output image type is np.float64 + sobel_y = cv.Sobel(blurred_img, cv.CV_64F, 0, 1, ksize=cfg.ksize_sobelY) + + # Subtract mean of pixels with small norm + sobel_x = subtract_low_norm_mean(sobel_x, frac=cfg.frac) + sobel_y = subtract_low_norm_mean(sobel_y, frac=cfg.frac) + + return sobel_x, sobel_y + + +def generate_2D_gradient_vector(img, cfg): + """ + Generates a dataset of 2D gradient vectors from the input image using Sobel filters. Only one channel is considered. + + Input: + img Input image I of shape (H,W) + cfg Config parameters for computing the image gradients + + Args: + points A 2D array where each row is a gradient vector [Gx, Gy] for each pixel of the input image, i.e. rows are observations (pixels), columns are variables (sobel_x, sobel_y). + """ + print("Generating 2D gradient vectors using Sobel filters") + + # Convert the image to uint8, i.e. range [0,255] (standard OpenCV format) + if img.dtype != np.uint8: + img = (255 * (img - img.min()) / (img.max() - img.min())).astype(np.uint8) + + # Compute gradients + sobel_x, sobel_y = image_gradients(img, cfg) + + # Generate 2D gradient vector + points = np.stack([sobel_x,sobel_y],axis=-1).reshape(-1,2) # stack along the last axis -> shape (H,W,2), then reshape to (H*W,2), i.e. rows are observations (pixels), columns are variables (sobel_x, sobel_y) + + return points + + +def scale_data(points, scale): + """ + Scale the 2D data points. Either with the overall standard deviation (scale=="overall") or separately for each dimension (scale=="per-dimension"). + + Input: + points 2D data points + + Args: + data Scaled data points + data_std Scaling factor for each dimension + """ + print(f"Scaling data using method: {scale}") + if scale == "overall": + std_x = std_y = anp.std(points) + elif scale == "per-dimension": + std_x = anp.std(points[:,0]) + std_y = anp.std(points[:,1]) + data = points.copy() + data[:,0] = data[:,0]/std_x + data[:,1] = data[:,1]/std_y + data_std = {'x': std_x, 'y': std_y} + + return data, data_std + + +def make_covariance_matrices(params): + """ + Compute covariance matrices + + Input: + params Given parameters: p1x, p1y, p2x, p2y, tau + + Args: + Sigma0, Sigma1, Sigma2 Covariance matrices for the Gaussian components + """ + p1 = params[:2] + p2 = params[2:4] + sigma = anp.exp(params[4]) + 1.e-6*anp.linalg.norm(p1) + 1.e-6*anp.linalg.norm(p2) # positive! + I = anp.eye(2) + + Sigma0 = sigma**2 * I + Sigma1 = anp.outer(p1, p1) + sigma**2 * I # anp.outer computes the outer product: 2D matrix here + Sigma2 = anp.outer(p2, p2) + sigma**2 * I + + return Sigma0, Sigma1, Sigma2 + + +def normal_distribution_2D_vectorized(X, cov): + """ + Multivariate normal distribution PDF for 2D data with zero mean. + + Input: + X array of shape (N,2) where each row is a 2D data point + cov 2x2 covariance matrix + + Args: + pdf array of shape (N,1) + """ + det = anp.linalg.det(cov) + inv_cov = anp.linalg.inv(cov) + quad_form = anp.sum((X @ inv_cov) * X, axis=1) + pdf = 1 / (2 * anp.pi * anp.sqrt(det)) * anp.exp(-0.5 * quad_form) + + return pdf.reshape(-1,1) # Otherwise, shape (N,). Alternatively, use [:,None] to convert to column vector with shape (N,1) + + +# f: function to minimize, i.e. negative log-likelihood of the Gaussian Mixture Model +def gmm_log_likelihood(params,data,w): + """ + Compute the negative log-likelihood of the Gaussian Mixture Model. + This function is used to compute the normals of two lines by fitting a Gaussian Mixture Model on image gradients in the function compute_transformation_matrix_from_image_gradients(). + """ + Sigma0, Sigma1, Sigma2 = make_covariance_matrices(params) + likelihood = (w[0] * normal_distribution_2D_vectorized(data, Sigma0) + + w[1] * normal_distribution_2D_vectorized(data, Sigma1) + + w[2] * normal_distribution_2D_vectorized(data, Sigma2)) + log_likelihood = anp.sum(anp.log(likelihood + 1e-10)) # Add small constant to avoid log(0) + + return -log_likelihood + + +def gmm_log_likelihood_reg(params,data,w,reg_param): + """ + Compute the negative log-likelihood of the Gaussian Mixture Model and adds a regularization term to ensure that the normals are aligned with the correct axes. + This function is used to compute the normals of two lines by fitting a Gaussian Mixture Model on image gradients in the function compute_transformation_matrix_from_image_gradients(). + """ + p1 = params[0:2] + p2 = params[2:4] + Sigma0, Sigma1, Sigma2 = make_covariance_matrices(params) + likelihood = (w[0] * normal_distribution_2D_vectorized(data, Sigma0) + + w[1] * normal_distribution_2D_vectorized(data, Sigma1) + + w[2] * normal_distribution_2D_vectorized(data, Sigma2)) + log_likelihood = anp.sum(anp.log(likelihood + 1e-10)) # Add small constant to avoid log(0) + reg = reg_param * (p2[0]**2/(anp.linalg.norm(p2))**2 + p1[1]**2/(anp.linalg.norm(p1))**2) # ensure that the normal p1 is aligned with x-axis, and the normal p2 with y-axis + + return -log_likelihood + reg + + +def compute_transformation_matrix(n1,n2): + """ + Compute the transformation matrix between two coordinate systems. + In the original coordinate system, there are two non-orthogonal lines with the normals n1 and n2. + In the new coordinate system, these lines are orthogonal. + + Args: + n1, n2 The normals of the (non-orthogonal) lines. + + Returns: + A_inv Transformation matrix: new --> original coordinate system, diagonal elements are 1 + A Transformation matrix: original --> new coordinate system, diagonal elements are 1 + """ + B = np.column_stack([n1, n2]) # columns are normals + B /=np.diag(B)[None,:] # divide each column by its diagonal element; np.diag(B)[None,:] is the row vector [n11,n22] of shape (1,2) + A = B.T # transformation: original coordinates -> new coordinates (orthogonal lines) + A_inv = np.linalg.inv(A) # transformation: new -> original + A_inv = A_inv @ np.diag(np.diag(A_inv)**(-1)) + A = np.linalg.inv(A_inv) + return A_inv, A + + +def warp_image_with_normals(img, n1, n2, fill_value=1e-6): + """ + Warp an image defined on a rectangular grid (xs, ys) so that lines with normals n1, n2 + become orthogonal in the transformed coordinates. Plot the warped image. + + Args: + img Input image (2D xarray) of shape (H,W) + n1, n2 Normalized line normals: (2,) arrays + fill_value Value for points outside the input domain + """ + H, W = img.shape # height, width of image + xs = np.arange(W) # original x-grid (cols) + ys = np.arange(H) # original y-grid (rows) + + A_inv, A = compute_transformation_matrix(n1,n2) + + # Output grid + nx = W # output image size as the original image + ny = H + + XX, YY = np.meshgrid(xs, ys, indexing="xy") # original grid, 2D arrays of all x and y coordinates (Cartesian) + pts = np.column_stack([XX.ravel(), YY.ravel()]) # list of pixel coordinates (points), shape (H*W,2), i.e. one row per pixel + new_pts = (A @ pts.T).T # transformed coordinates of all original pixels, shape (H*W,2), i.e. one row per pixel + + x_min, x_max = new_pts[:,0].min(), new_pts[:,0].max() # determine bounding box in transformed space (both in x- and y-direction) --> how big the new image grid needs to be! + y_min, y_max = new_pts[:,1].min(), new_pts[:,1].max() + new_xs = np.linspace(x_min, x_max, nx) # define a rectangular grid in transformed space + new_ys = np.linspace(y_min, y_max, ny) + grid_x, grid_y = np.meshgrid(new_xs, new_ys, indexing="xy") # grid for the output image (covers the whole transformed image) + + # Map new coords back to original coords + Yq = np.column_stack([grid_x.ravel(), grid_y.ravel()]) # list of output pixel coordinates (points) + Xq = (A_inv @ Yq.T).T # output pixel coordinates mapped back to original image coordinates --> need to be able to interpolate the image values at the new positions + + # Interpolate at these mapped coordinates Xq + warped = interpn( + points=(xs, ys), # original coordinates (xs, ys), pixels + values=img.T, # image values at these coordinates (first index is x, second index is y) + xi=Xq, # mapped coordinates, where we want to know the values + method="linear", + bounds_error=False, + fill_value=fill_value, + ).reshape(ny, nx) # converts it back to a 2D image + + fig, axs = plt.subplots(1, 2, figsize=(12, 6)) + axs[0].imshow(img, origin="lower") + axs[0].set_title("Original") + axs[1].imshow(warped, origin="lower", extent=(new_xs.min(), new_xs.max(), new_ys.min(), new_ys.max())) + axs[1].set_title("Orthogonalized") + fig.suptitle(f"Normals: n1 = {np.array2string(n1, precision=4)}, n2 = {np.array2string(n2, precision=4)}") + plt.show() + + +def compute_transformation_matrix_from_image_gradients(img, cfg): + """ + This function computes the normals p1, p2 and slopes m1, m2 of lines with 2 distinct directions in the input image. + This is done by fitting a Gaussian Mixture Model on the image gradients in x- and y-direction. + The unknown parameters are given by θ = [p1_x, p1_y, p2_x, p2_y, τ] with σ_ε = exp(τ) > 0. + The log-likelihood function is given by L(θ) = Σ_(i=1)^N log(Σ_(k=0)^2 w_k N(v_i | 0,Σ_k(θ))) with Σ_0=σ_ε^2*I, Σ_k=p_k*p_k^T + σ_ε^2*I for k=1,2, w is fixed. + + Args: + img Input image (xarray) + cfg Config parameters (for computing the image gradients, for the model, and for the optimization) + + Returns: + p1, p2 Normalized normals (p1 aligned with x-axis, p2 aligned with y-axis) + m1, m2 Slopes of the lines with 2 distinct directions (abs(m1) > abs(m2)) + """ + # Turn tuples into arrays (they do not change values during optimization) + w = np.array(cfg.model.w).reshape(-1, 1) # values have shape (3,1) for broadcasting + init_params = np.array(cfg.model.init_params, dtype=float) + + # Opencv assumes that the origin is at upper left. Apply vertical flip to the image to convert origin at lower left to origin at upper left. + img = np.flipud(img) + + # Image gradients + data = generate_2D_gradient_vector(img, cfg.gradient) + + # Scale the image gradients + data, data_std = scale_data(data, cfg.model.scale) + + # Data fitting (GMM on gradients) + logger.info(f"Optimizing GMM") + if cfg.model.likelihood == "without-reg": + logger.info(f"Initial log-likelihood: {gmm_log_likelihood(init_params,data,w)}") + problem = value_and_grad(lambda params: gmm_log_likelihood(params,data,w)) + if cfg.model.likelihood == "with-reg": + logger.info(f"Initial log-likelihood: {gmm_log_likelihood_reg(init_params,data,w,cfg.model.reg_param)}") + problem = value_and_grad(lambda params: gmm_log_likelihood_reg(params,data,w,cfg.model.reg_param)) + result = minimize(problem, init_params, method="L-BFGS-B", jac=True, tol=0.0, options={'maxiter':cfg.optimization.max_iterations, 'gtol': cfg.optimization.epsilon}) + + p1 = result.x[:2] # normal aligned with x-axis + p2 = result.x[2:4] # normal aligned with y-axis + # Ensure that the normals point in positive x- and y-direction (in the origin at lower left coordinate system) NOTE: THIS ONLY WORKS IF P1 ALIGNED WITH X-AXIS, AND P2 WITH Y-AXIS!!! + if np.abs(p1[0]) > np.abs(p1[1]) and p1[0] < 0: # check that p1 is aligned with x-axis, and ensure that it points in the positive x-direction + p2 = -p2 + if np.abs(p2[1]) > np.abs(p2[0]) and p2[1] > 0: # check that p2 is aligned with y-axis, and ensure that it points in the negative y-direction (flipped vertically afterwards!) + p2 = -p2 + p1_rescaled = np.array([p1[0]*data_std['x'],p1[1]*data_std['y']]) # Scale back + p2_rescaled = np.array([p2[0]*data_std['x'],p2[1]*data_std['y']]) + p1_rescaled = p1_rescaled / np.linalg.norm(p1_rescaled) # Normalize + p2_rescaled = p2_rescaled / np.linalg.norm(p2_rescaled) + m1 = -p1[0]*data_std['x']/(p1[1]*data_std['y']) + m2 = -p2[0]*data_std['x']/(p2[1]*data_std['y']) + + # Convert results back to origin at lower left, i.e. apply vertical flip. + p1[1] = -p1[1] + p2[1] = -p2[1] + p1_rescaled[1] = -p1_rescaled[1] + p2_rescaled[1] = -p2_rescaled[1] + m1 = -m1 + m2 = -m2 + + return p1_rescaled, p2_rescaled, m1, m2 + diff --git a/src/qua_dashboards/video_mode/utils/config.py b/src/qua_dashboards/video_mode/utils/config.py new file mode 100644 index 0000000..2fe487e --- /dev/null +++ b/src/qua_dashboards/video_mode/utils/config.py @@ -0,0 +1,63 @@ +from dataclasses import dataclass, field +import numpy as np + +__all__ = ["TransformationMatrixConfig"] + +@dataclass +class GradientConfig: + sigmaX_blur: float = 1.0 # >=0 + sigmaY_blur: float = 1.0 # >=0 + ksize_sobelX: int = 5 # >0, odd number + ksize_sobelY: int = 5 # >0, odd number + frac: float = 0.7 # between 0 and 1 + + def __post_init__(self): + if self.sigmaX_blur < 0: + raise ValueError(f"Config: Invalid sigmaX_blur value: {self.sigmaX_blur}") + if self.sigmaY_blur < 0: + raise ValueError(f"Config: Invalid sigmaY_blur value: {self.sigmaY_blur}") + if self.ksize_sobelX <= 0 or self.ksize_sobelX % 2 == 0: + raise ValueError(f"Config: Invalid ksize_sobelX value: {self.ksize_sobelX}") + if self.ksize_sobelY <= 0 or self.ksize_sobelY % 2 == 0: + raise ValueError(f"Config: Invalid ksize_sobelY value: {self.ksize_sobelY}") + if self.frac > 1 or self.frac < 0: + raise ValueError(f"Config: Invalid frac value: {self.frac}") + +@dataclass +class OptimizationConfig: + max_iterations: int = 1000000 # >0 + epsilon: float = 1e-4 # tolerance: pick small value >0 + + def __post_init__(self): + if self.max_iterations <= 0: + raise ValueError(f"Config: Invalid max_iterations value: {self.max_iterations}") + if self.epsilon <= 0: + raise ValueError(f"Config: Invalid epsilon (tolerance) value: {self.epsilon}") + + +@dataclass +class ModelConfig: + scale: str = "per-dimension" # "overall" or "per-dimension" + likelihood: str = "with-reg" # "with-reg" or "without-reg", i.e. with or without regularization term that ensures alignment of the normals with the correct axes + w: tuple = (0.8, 0.1, 0.1) # Weights for the Gaussian components (mixing coefficients), sum w_i = 1 + init_params: tuple = (0.1, 0.0, 0.0, 0.1, 0.0) # Initial guess for GMM: p1, p2 horizontal and vertical, tau = log_sigma = log(1.0) = 0.0 + reg_param: float = 1000.0 # Regularization parameter >0 + + def __post_init__(self): + if self.scale not in ("overall","per-dimension"): + raise ValueError(f"Config: Invalid scale parameter: {self.scale}") + if self.likelihood not in ("with-reg","without-reg"): + raise ValueError(f"Config: Invalid likelihood parameter: {self.likelihood}") + if len(self.w) != 3 or not abs(sum(self.w) - 1.0) < 1e-8: + raise ValueError(f"Config: Invalid weights w: {self.w}") + if len(self.init_params) != 5: + raise ValueError(f"Config: Invalid init_params: {self.init_params}") + if self.reg_param <= 0: + raise ValueError(f"Config: Invalid reg_param: {self.reg_param}") + + +@dataclass +class TransformationMatrixConfig: + gradient: GradientConfig = field(default_factory=GradientConfig) + optimization: OptimizationConfig = field(default_factory=OptimizationConfig) + model: ModelConfig = field(default_factory=ModelConfig) \ No newline at end of file diff --git a/src/qua_dashboards/video_mode/video_mode_component.py b/src/qua_dashboards/video_mode/video_mode_component.py index 71d9a0a..bec1f16 100644 --- a/src/qua_dashboards/video_mode/video_mode_component.py +++ b/src/qua_dashboards/video_mode/video_mode_component.py @@ -120,7 +120,7 @@ def __init__( data_acquirer=self.data_acquirer, # Pass data_acquirer here ) annotation_tab = AnnotationTabController( - component_id=f"{self.component_id}-annotation-tab", + component_id=f"{self.component_id}-annotation-tab", data_acquirer=self.data_acquirer, ) self.tab_controllers: List[BaseTabController] = [ live_view_tab, @@ -242,7 +242,8 @@ def get_layout(self) -> Component: data = {}, ), dcc.Store( - id=self._get_store_id(self.VIEWER_LAYOUT_CONFIG_STORE_SUFFIX), data={} + id=self._get_store_id(self.VIEWER_LAYOUT_CONFIG_STORE_SUFFIX), + data={}, ), ] diff --git a/src/qua_dashboards/voltage_control/voltage_control_component.py b/src/qua_dashboards/voltage_control/voltage_control_component.py index d574621..026beb1 100644 --- a/src/qua_dashboards/voltage_control/voltage_control_component.py +++ b/src/qua_dashboards/voltage_control/voltage_control_component.py @@ -14,6 +14,7 @@ ) from qua_dashboards.core import BaseComponent, ParameterProtocol +from qua_dashboards.utils import CallbackParameter from .voltage_control_row import VoltageControlRow, format_voltage @@ -36,9 +37,14 @@ def __init__( voltage_parameters: Sequence[ParameterProtocol], update_interval_ms: int = 1000, layout_columns: int = 3, + callback_on_param_change = None + ): super().__init__(component_id=component_id) self.voltage_parameters = voltage_parameters + #wrap parameters to trigger callbacks on change. + if callback_on_param_change is not None: + self.voltage_parameters = [CallbackParameter(param, callback_on_param_change) for param in self.voltage_parameters] self.update_interval_ms = update_interval_ms self._initial_values_loaded = False # To ensure first update populates values diff --git a/src/qua_dashboards/voltage_control/voltage_control_row.py b/src/qua_dashboards/voltage_control/voltage_control_row.py index 1e495f4..3273567 100644 --- a/src/qua_dashboards/voltage_control/voltage_control_row.py +++ b/src/qua_dashboards/voltage_control/voltage_control_row.py @@ -103,7 +103,7 @@ def get_layout(self) -> dbc.Row: ), dbc.Col( dbc.Label( - children=self.param.units, style={"whiteSpace": "nowrap"} + children=self.param.unit, style={"whiteSpace": "nowrap"} ), style={ "width": UNITS_WIDTH,