From 35f16f7dcdaf5419c59d21be13fa82ebe4279439 Mon Sep 17 00:00:00 2001 From: ryantan Date: Mon, 14 Apr 2025 10:43:08 +0200 Subject: [PATCH] removed AttocubePowerboxScript.py --- AttocubePowerboxScript.py | 935 -------------------------------------- 1 file changed, 935 deletions(-) delete mode 100644 AttocubePowerboxScript.py diff --git a/AttocubePowerboxScript.py b/AttocubePowerboxScript.py deleted file mode 100644 index b3715e9..0000000 --- a/AttocubePowerboxScript.py +++ /dev/null @@ -1,935 +0,0 @@ -# -*- coding: utf-8 -*- -""" -Created on Fri Dec 22 15:10:10 2023 -Lightfield + Positioner -@author: Serdar, adjusted by Lukas -""" -############################################ -# Packages from Ryan -import re -import math -import threading -import pyvisa -# from pyvisa import ResourceManager, constants - -# B Field Limits (in T) -BX_MAX = 1.7 -BY_MAX = 1.7 -BZ_MAX = 4.0 -############################################ - -import AMC -import csv -import time -import clr -import sys -import os -import spe2py as spe -import spe_loader as sl -import pandas as pd -import time -from System.IO import * -from System import String -import numpy as np -import matplotlib.pyplot as plt -import datetime -from typing import Union - - -#First choose your controller -IP_AMC300 = "192.168.1.1" -IP_AMC100 = "192.168.71.100" - -# IP = "192.168.1.1" -IP = IP_AMC100 - - -# Import os module -import os, glob, string - -# Import System.IO for saving and opening files -from System.IO import * - -from System.Threading import AutoResetEvent - -# Import C compatible List and String -from System import String -from System.Collections.Generic import List - -# Add needed dll references -sys.path.append(os.environ['LIGHTFIELD_ROOT']) -sys.path.append(os.environ['LIGHTFIELD_ROOT']+"\\AddInViews") -sys.path.append(r'C:\Program Files\Princeton Instruments\LightField\AddInViews') #I added them by hand -serdar -sys.path.append(r'C:\Program Files\Princeton Instruments\LightField') #this one also -clr.AddReference('PrincetonInstruments.LightFieldViewV5') -clr.AddReference('PrincetonInstruments.LightField.AutomationV5') -clr.AddReference('PrincetonInstruments.LightFieldAddInSupportServices') -os.environ['LIGHTFIELD_ROOT'] = r'C:\Program Files\Princeton Instruments\LightField' -# PI imports -from PrincetonInstruments.LightField.Automation import Automation -from PrincetonInstruments.LightField.AddIns import ExperimentSettings -from PrincetonInstruments.LightField.AddIns import CameraSettings -#from PrincetonInstruments.LightField.AddIns import DeviceType -from PrincetonInstruments.LightField.AddIns import SpectrometerSettings -from PrincetonInstruments.LightField.AddIns import RegionOfInterest - -######################################################################################################### code begins from here ############################################# - -def set_custom_ROI(): - - # Get device full dimensions - dimensions = experiment.FullSensorRegion() - - regions = [] - - # Add two ROI to regions - regions.append( - RegionOfInterest( - int(dimensions.X), int(dimensions.Y), - int(dimensions.Width), int(dimensions.Height//4), # Use // for integer division - int(dimensions.XBinning), int(dimensions.Height//4))) - - - - # Set both ROI - experiment.SetCustomRegions(regions) - -def experiment_completed(sender, event_args): #callback function which is hooked to event completed, this is the listener - print("... Acquisition Complete!") - acquireCompleted.Set() #set the event. This puts the autoresetevent false.(look at .NET library for furher info) - -def InitializerFilenameParams(): - experiment.SetValue(ExperimentSettings.FileNameGenerationAttachIncrement, False) - experiment.SetValue(ExperimentSettings.FileNameGenerationIncrementNumber, 1.0) - experiment.SetValue(ExperimentSettings.FileNameGenerationIncrementMinimumDigits, 2.0) - experiment.SetValue(ExperimentSettings.FileNameGenerationAttachDate, False) - experiment.SetValue(ExperimentSettings.FileNameGenerationAttachTime, False) - -def AcquireAndLock(name): - print("Acquiring...", end = "") - # name += 'Exp{0:06.2f}ms.CWL{1:07.2f}nm'.format(\ - # experiment.GetValue(CameraSettings.ShutterTimingExposureTime)\ - # ,experiment.GetValue(SpectrometerSettings.GratingCenterWavelength)) - - experiment.SetValue(ExperimentSettings.FileNameGenerationBaseFileName, name) #this creates .spe file with the name - experiment.Acquire() # this is an ashynrchronus func. - acquireCompleted.WaitOne() - -def calculate_distance(x1, y1, x2, y2): - return np.sqrt((x2 - x1)**2 + (y2 - y1)**2) - -def generate_scan_positions(center, range_val, resolution): - positive_range = np.arange(center, center + range_val + resolution, resolution) - return positive_range - -def save_as_csv(filename, position_x, position_y): - file_existance = os.path.isfile(filename) - - with open(filename, 'a', newline = '') as csvfile: - writer = csv.writer(csvfile) - - if not file_existance: - writer.writerow(['x_coordinates','y_coordinates']) - - writer.writerow([position_x, position_y]) - -def move_axis(axis, target): - """ - This function moves an axis to the specified target and stop moving after it is in the really closed - vicinity (+- 25nm) of the target (listener hooked to it). - """ - amc.move.setControlTargetPosition(axis, target) - amc.control.setControlMove(axis, True) - while not (target - 25) < amc.move.getPosition(axis) < (target + 25): - time.sleep(0.1) - time.sleep(0.15) - while not (target - 25) < amc.move.getPosition(axis) < (target + 25): - time.sleep(0.1) - amc.control.setControlMove(axis, False) - -def move_xy(target_x, target_y): # moving in x and y direction closed to desired position - amc.move.setControlTargetPosition(0, target_x) - amc.control.setControlMove(0, True) - amc.move.setControlTargetPosition(1, target_y) - amc.control.setControlMove(1, True) - while not (target_x - 25) < amc.move.getPosition(0) < (target_x + 25) and (target_y - 25) < amc.move.getPosition(1) < (target_y + 25): - time.sleep(0.1) - time.sleep(0.15) - while not (target_x - 25) < amc.move.getPosition(0) < (target_x + 25) and (target_y - 25) < amc.move.getPosition(1) < (target_y + 25): - time.sleep(0.1) - - amc.control.setControlOutput(0, False) - amc.control.setControlOutput(1, False) - - -# intensity_data = [] # To store data from each scan -# data_list = [] - -def move_scan_xy(range_x, range_y, resolution, Settings, baseFileName): - """ - This function moves the positioners to scan the sample with desired ranges and resolution in 2 dimensions. - At the end it saves a csv file - - Parameters - ---------- - range_x : integer in nm. max value is 5um - Scan range in x direction. - range_y : integer in nm. max value is 5um - Scan range in y direction. - resolution : integer in nm. - Room temprature max res is 50nm. In cyrostat (4K) it is 10nm (check the Attocube manual) - baseFileName: string. At the end the saved file will be: baseFileName_scan_data.csv and it will be saved to the current directory - - Returns - ------- - None. - - """ - start_time = time.time() - axis_x = 0 #first axis - axis_y = 1 #second axis - center_x = amc.move.getPosition(axis_x) - center_y = amc.move.getPosition(axis_y) - # #check if the intput range is reasonable - # if amc.move.getPosition(axis_x) + range_x >= 5000 or amc.move.getPosition(axis_x)- range_x <= 0 or amc.move.getPosition(axis_y) + range_y >=5000 or amc.move.getPosition(axis_y) - range_y <= 5000 : - # print("scan range is out of range!") - # return - # +- range from current positions for x and y directions - - - array_x = generate_scan_positions(center_x, range_x, resolution) - array_y = generate_scan_positions(center_y, range_y, resolution) - total_points = len(array_x)*len(array_y) - len_y = len(array_y) - intensity_data = [] # To store data from each scan - data_list = [] - cwd = os.getcwd() # save original directory - - #This gives a directory, in which the script will save the spectrum of each spot as spe - #However, it will open the spectrum, convert it to txt, add it to the intensity_data and delete the spe file - temp_folder_path = "C:/Users/localadmin/Desktop/Users/Lukas/2024_02_08_Map_test" - - #scanning loop - for i, x_positions in enumerate(array_x): - move_axis(axis_x, x_positions) - y = False - for j, y_positions in enumerate(array_y): - move_axis(axis_y, y_positions) - #each time when the positioner comes to the beggining of a new line - #this if will make the positioner wait a bit longer to really go to the target. - if y == False: - move_axis(axis_y, y_positions) - y = True - - #we acquire with the LF - acquire_name_spe = f'{baseFileName}_X{x_positions}_Y{y_positions}' - AcquireAndLock(acquire_name_spe) #this creates a .spe file with the scan name. - - #read the .spe file and get the data as loaded_files - cwd = os.getcwd() # save original directory - os.chdir(temp_folder_path) #change directory - loaded_files = sl.load_from_files([acquire_name_spe + '.spe']) # get the .spe file as a variable - os.chdir(cwd) # go back to original directory - - # Delete the created .spe file from acquiring after getting necessary info - spe_file_path = os.path.join(temp_folder_path, acquire_name_spe + '.spe') - os.remove(spe_file_path) - - distance = calculate_distance(x_positions, y_positions,amc.move.getPosition(axis_x), amc.move.getPosition(axis_y)) - - points_left = total_points - (i * len_y + (j+1)) + 1 - print('Points left in the scan: ', points_left) - - #append the intensity data as it is (so after every #of_wl_points, the spectrum of the next point begins) - intensity_data.append(loaded_files.data[0][0][0]) - - data_list.append({ - 'position_x': x_positions, - 'position_y': y_positions, - 'actual_x': amc.move.getPosition(axis_x), - 'actual_y': amc.move.getPosition(axis_y), - 'distance': distance, - }) - - #moves back to starting position - move_axis(axis_x, center_x) - move_axis(axis_y, center_y) - - #prints total time the mapping lasted - end_time = time.time() - elapsed_time = (end_time - start_time) / 60 - print('Scan time: ', elapsed_time, 'minutes') - - # df = pd.DataFrame(data_list) - - #save intensity & WL data as .txt - os.chdir('C:/Users/localadmin/Desktop/Users/Lukas') - # creates new folder for MAP data - new_folder_name = "Test_Map_" + f"{datetime.datetime.now().strftime('%Y_%m_%d_%H.%M')}" - os.mkdir(new_folder_name) - # Here the things will be saved in a new folder under user Lukas ! - # IMPORTANT last / has to be there, otherwise data cannot be saved and will be lost!!!!!!!!!!!!!!!! - os.chdir('C:/Users/localadmin/Desktop/Users/Lukas/'+ new_folder_name) - - intensity_data = np.array(intensity_data) - np.savetxt(Settings + str(center_x) + '_' + str(center_y) + experiment_name +'.txt', intensity_data) - - wl = np.array(loaded_files.wavelength) - np.savetxt("Wavelength.txt", wl) - -################################################################# RYAN'S FUNCTIONS HERE ########################################################################################## - -def sep_num_from_units(powerbox_output :str)->list: - ''' - Receives a string as input and separates the numberic value and unit and returns it as a list. - - Parameters - ---------- - powerbox_output : str - string output from the attocube powerbox, e.g. 1.35325kG - - Returns - ------- - list - list of float value and string (b value and it's units). If string is purely alphabets, then return a single element list - - ''' - match = re.match(r'\s*([+-]?\d*\.?\d+)([A-Za-z]+)', powerbox_output) - if match: - numeric_part = float(match.group(1)) # Convert the numeric part to a float - alphabetic_part = match.group(2) # Get the alphabetic part - return [numeric_part, alphabetic_part] - else: - return [powerbox_output,] - - -def query_no_echo(instr:pyvisa.resources.Resource, command:str, sleeptime=0)->str: - """helper function for the Attocube APS100 that queries a function to the device, removing the echo. - - Args: - instr (pyvisa.resources.Resource): - command (str): commands, can be stringed in series with ; between commands - sleeptime (float, optional): delay time between commands. Defaults to 0.01. - - Returns: - str: _description_ - """ '''''' - try: - print(f"Sending command: {command}") - instr.write(command) - time.sleep(sleeptime) - echo_response = instr.read() # Read and discard the echo - # print(f"Echo response: {echo_response}") - actual_response = instr.read() # Read the actual response - print(f"Actual response: {actual_response}") - return actual_response - except pyvisa.VisaIOError as e: - print(f"Error communicating with instrument: {e}") - return None - - -def write_no_echo(instr:pyvisa.resources.Resource, command:str, sleeptime=0)->str: - """helper function for the Attocube APS100 that writes a function to the device, removing the echo. - - Args: - instr (pyvisa.resources.Resource): - command (str): commands, can be stringed in series with ; between commands - sleeptime (float, optional): delay time between commands. Defaults to 0.01. - - Returns: - str: _description_ - """ '''''' - try: - print(f"Sending command: {command}") - instr.write(command) - time.sleep(sleeptime) # Give the device some time to process - try: - while True: - echo_response = instr.read() # Read and discard the echo - # print(f"Echo response: {echo_response}") - except pyvisa.VisaIOError as e: - # Expected timeout after all echoed responses are read - if e.error_code != pyvisa.constants.VI_ERROR_TMO: - raise - except pyvisa.VisaIOError as e: - print(f"Error communicating with instrument: {e}") - - -# receive values in units of T, rescale in kg to talk with the power supplyy. 1T = 10kG -# NOTE: removed singlepowersupply_bool, reading serial-nr. of the device instead. -# old save folder: "C:/Users/localadmin/Desktop/Users/Lukas/2024_02_08_Map_test" -def sweep_b_val(instr:pyvisa.resources.Resource, min_bval:float, max_bval:float, - res:float, magnet_coil:str, Settings:str, base_file_name='', - reversescan_bool=False, zerowhenfin_bool=False, loopscan_bool=False)->None: - # TODO: update docs in the end - """ this function performs a sweep of the B field of the chosen magnet coil. It creates a list o B values from the given min and max values, - with the given resolution. For each value, a measurement of the spectrum of the probe in the cryostat is made, using the LightField spectrometer. - - Args: - instr (pyvisa.resources.Resource): chosen power supply device to connect to - min_bval (float): min B value of the scan (please input in units of Tesla) - max_bval (float): max B value of the scan (please input in units of Tesla) - res (float): resolution of the list of B values (please input in units of Tesla) - magnet_coil (str): select magnet coil to be used. String should be 'x-axis','y-axis' or 'z-axis'. - Settings (str): experiment settings, included in file name. - base_file_name (str, optional): base file name. Defaults to ''. - singlepowersupply_bool (bool, optional): _description_. Defaults to False. - reversescan_bool (bool, optional): _description_. Defaults to False. - zerowhenfin_bool (bool, optional): _description_. Defaults to False. - - Raises: - ValueError: when By limit is exceeded. - ValueError: when Bz limit is exceeded. - ValueError: when Bx limit is exceeded. - ConnectionError: when no device is connected. - """ '''''' - - def pyramid_list(lst) -> Union[list, np.ndarray]: - """reverses the list and removes the first element of reversed list. Then, this is appended to - the end of the original list and returned as the 'pyramid' list. - - Args: - lst (list or np.ndarray): - Raises: - TypeError: if the input object isn't a list or np.ndarray - Returns: - Union[list, np.ndarray]: the pyramid list - """ '''''' - if isinstance(lst, list): - return lst + lst[-2::-1] - elif isinstance(lst, np.ndarray): - return np.append(lst, lst[-2::-1]) - else: - raise TypeError('Please input a list!') - - # defines the folder, in which the data from the spectrometer is temporarily stored in - temp_folder_path = "C:/Users/localadmin/Desktop/Users/Lukas/2024_02_08_Map_test" - - # if path_save =='': - # path_save = datetime.datetime.now().strftime("%Y_%m_%d_%H%M_hrs_") - - if base_file_name =='': - base_file_name = datetime.datetime.now().strftime('%Y_%m_%d_%H.%M') - - start_time = time.time() # start of the scan function - - instr_info = query_no_echo(instr, '*IDN?') - - instr_bsettings = list(sep_num_from_units(el) for el in query_no_echo(instr, 'UNITS?;LLIM?;ULIM?').split(';')) # deliver a 3 element list of lists containing the set unit, llim and ulim - - if instr_bsettings[0][0] == 'T': - instr_bsettings[1][0] = instr_bsettings[1][0]*0.1 # rescale kG to T, device accepts values only in kG or A, eventho we set it to T - instr_bsettings[2][0] = instr_bsettings[2][0]*0.1 - - # if singlepowersupply_bool: # checks limits of Bx or By - # if (min_bval< -BY_MAX) or (max_bval > BY_MAX): - # raise ValueError('Input limits exceed that of the magnet By! Please input smaller limits.') - # elif '1' in query_no_echo(instr, 'CHAN?'): # check if its the coils for Bz - # if (min_bval < -BZ_MAX) or (max_bval > BZ_MAX): - # raise ValueError('Input limits exceed that of the magnet (Bz)! Please input smaller limits.') - # else: # checks limits of Bx - # if (min_bval< -BX_MAX) or (max_bval > BX_MAX): - # raise ValueError('Input limits exceed that of the magnet Bx! Please input smaller limits.') - - if '2101014' in instr_info and (magnet_coil=='y-axis'): # single power supply - if (min_bval< -BY_MAX) or (max_bval > BY_MAX): - raise ValueError('Input limits exceed that of the magnet By! Please input smaller limits.') - elif '2301034' in instr_info: # dual power supply - if magnet_coil=='z-axis': # check if its the coils for Bz - if (min_bval < -BZ_MAX) or (max_bval > BZ_MAX): - raise ValueError('Input limits exceed that of the magnet (Bz)! Please input smaller limits.') - write_no_echo(instr, 'CHAN 1') - elif magnet_coil=='x-axis': # checks limits of Bx - if (min_bval< -BX_MAX) or (max_bval > BX_MAX): - raise ValueError('Input limits exceed that of the magnet Bx! Please input smaller limits.') - write_no_echo(instr, 'CHAN 2') - else: - raise ConnectionError('Device is not connected!') - - write_no_echo(instr, f'LLIM {min_bval*10};ULIM {max_bval*10}') # sets the given limits, must convert to kG for the device to read - bval_lst = np.arange(min_bval, max_bval + res, res) # creates list of B values to measure at, with given resolution, in T - - # TODO: unused, see if can remove - # init_bval = sep_num_from_units(query_no_echo(instr, 'IMAG?'))[0]*0.1 # queries the initial B value of the coil, rescale from kG to T - - init_lim, subsequent_lim = 'LLIM', 'ULIM' - init_sweep, subsequent_sweep = 'DOWN', 'UP' - - #################################################### - # TODO: decide whether to start at min b val or max b val, depending on which one is nearer, IMPLEMENT THIS LATER - # nearest_bval = (abs(init_bval - min_bval), abs(init_bval - max_bval)) - # if nearest_bval[0] <= nearest_bval[1]: - # reversescan_bool = True - #################################################### - - # if reverse scan, then flip the values in the b list, and swap the initial limit and sweep conditions - if reversescan_bool: - bval_lst = bval_lst[::-1] - init_lim, subsequent_lim = subsequent_lim, init_lim - init_sweep, subsequent_sweep = subsequent_sweep, init_sweep - - # creates the pyramid list of B vals if one were to perform a hysteresis measurement - if loopscan_bool: - bval_lst = pyramid_list(bval_lst) - - total_points = len(bval_lst) - middle_index_bval_lst = total_points // 2 - intensity_data = [] # To store data from each scan - cwd = os.getcwd() # save original directory - - # NOTE: helper function for the scanning loop - def helper_scan_func(idx, bval, instr=instr, init_lim=init_lim, init_sweep=init_sweep, - subsequent_lim=subsequent_lim, subsequent_sweep=subsequent_sweep, sleep=5): - if idx == 0: # for first iteration, sweep to one of the limits - write_no_echo(instr, f'{init_lim} {bval*10}') # convert back to kG - write_no_echo(instr, f'SWEEP {init_sweep}') - else: - write_no_echo(instr, f'{subsequent_lim} {bval*10}') # convert back to kG - write_no_echo(instr, f'SWEEP {subsequent_sweep}') - - actual_bval = sep_num_from_units(query_no_echo(instr, 'IMAG?'))[0]*0.1 # convert kG to T - print(f'Actual magnet strength: {actual_bval} T,', f'Target magnet strength: {bval} T') - - while abs(actual_bval - bval) > 0.0001: - time.sleep(5) # little break - actual_bval = sep_num_from_units(query_no_echo(instr, 'IMAG?'))[0]*0.1 - # update the actual bval - print(f'Actual magnet strength: {actual_bval} T,', f'Target magnet strength: {bval} T') - - #scanning loop - for i, bval in enumerate(bval_lst): - # if init_bval == bval: - # # if initial bval is equal to the element of the given iteration from the bval_lst, then commence measuring the spectrum - # pass - # else: - - # NOTE: original code without the loop scan - ################################################ - # if i == 0: # for first iteration, sweep to one of the limits - # write_no_echo(instr, f'{init_lim} {bval*10}') # convert back to kG - # write_no_echo(instr, f'SWEEP {init_sweep}') - # else: - # write_no_echo(instr, f'{subsequent_lim} {bval*10}') # convert back to kG - # write_no_echo(instr, f'SWEEP {subsequent_sweep}') - - # actual_bval = sep_num_from_units(query_no_echo(instr, 'IMAG?'))[0]*0.1 # convert kG to T - # print(f'Actual magnet strength: {actual_bval} T,', f'Target magnet strength: {bval} T') - - # while abs(actual_bval - bval) > 0.0001: - # time.sleep(5) # little break - # actual_bval = sep_num_from_units(query_no_echo(instr, 'IMAG?'))[0]*0.1 - # # update the actual bval - # print(f'Actual magnet strength: {actual_bval} T,', f'Target magnet strength: {bval} T') - ############################################### - if not loopscan_bool: - helper_scan_func(i, bval) - else: - if i <= middle_index_bval_lst: - helper_scan_func(i, bval) - else: - helper_scan_func(i, bval, instr=instr, init_lim=subsequent_lim, init_sweep=subsequent_sweep, - subsequent_lim=init_lim, subsequent_sweep=init_sweep, sleep=5) - - time.sleep(5) - # we acquire with the LF - acquire_name_spe = f'{base_file_name}_{bval}T' - AcquireAndLock(acquire_name_spe) #this creates a .spe file with the scan name. - - # read the .spe file and get the data as loaded_files - cwd = os.getcwd() # save original directory - os.chdir(temp_folder_path) #change directory - loaded_files = sl.load_from_files([acquire_name_spe + '.spe']) # get the .spe file as a variable - os.chdir(cwd) # go back to original directory - - # Delete the created .spe file from acquiring after getting necessary info - spe_file_path = os.path.join(temp_folder_path, acquire_name_spe + '.spe') - os.remove(spe_file_path) - - points_left = total_points - i - 1 # TODO: SEE IF THIS IS CORRECT - print('Points left in the scan: ', points_left) - - #append the intensity data as it is (so after every #of_wl_points, the spectrum of the next point begins) - intensity_data.append(loaded_files.data[0][0][0]) - - #prints total time the mapping lasted - end_time = time.time() - elapsed_time = (end_time - start_time) / 60 - print('Scan time: ', elapsed_time, 'minutes') - - write_no_echo(instr, f'LLIM {instr_bsettings[1][0]*10};ULIM {instr_bsettings[2][0]*10}') # reset the initial limits of the device after the scan - - if zerowhenfin_bool: - write_no_echo(instr, 'SWEEP ZERO') # if switched on, discharges the magnet after performing the measurement loop above - - #save intensity & WL data as .txt - os.chdir('C:/Users/localadmin/Desktop/Users/Lukas') - # creates new folder for MAP data - new_folder_name = "Test_Map_" + f"{datetime.datetime.now().strftime('%Y_%m_%d_%H.%M')}" - os.mkdir(new_folder_name) - # Here the things will be saved in a new folder under user Lukas ! - # IMPORTANT last / has to be there, otherwise data cannot be saved and will be lost!!!!!!!!!!!!!!!! - os.chdir('C:/Users/localadmin/Desktop/Users/Lukas/'+ new_folder_name) - - intensity_data = np.array(intensity_data) - np.savetxt(Settings + str(min_bval) + 'T_to_' + str(max_bval) + 'T' + experiment_name +'.txt', intensity_data) - - wl = np.array(loaded_files.wavelength) - np.savetxt("Wavelength.txt", wl) - - -def polar_to_cartesian(radius, start_angle, end_angle, step_size, clockwise=True): - # TODO: DOCS - """Creates a list of discrete cartesian coordinates (x,y), given the radius, start- and end angles, the angle step size, and the direction of rotation. - Function then returns a list of two lists: list of angles and list of cartesian coordinates (x,y coordinates in a tuple). - - Args: - radius (_type_): _description_ - start_angle (_type_): _description_ - end_angle (_type_): _description_ - step_size (_type_): _description_ - clockwise (bool, optional): _description_. Defaults to True. - - Returns: - _type_: _description_ - """ """""" - # Initialize lists to hold angles and (x, y) pairs - angles = [] - coordinates = [] - - # Normalize angles to the range [0, 360) - start_angle = start_angle % 360 - end_angle = end_angle % 360 - - if clockwise: - # Clockwise rotation - current_angle = start_angle - while True: - # Append the current angle to the angles list - angles.append(current_angle % 360) - - # Convert the current angle to radians - current_angle_rad = math.radians(current_angle % 360) - - # Convert polar to Cartesian coordinates - x = radius * math.cos(current_angle_rad) - y = radius * math.sin(current_angle_rad) - - # Append the (x, y) pair to the list - coordinates.append((x, y)) - - # Check if we've reached the end_angle (handling wrap-around) (current_angle - step_size) % 360 == end_angle or - if current_angle % 360 == end_angle: - break - - # Decrement the current angle by the step size - current_angle -= step_size - if current_angle < 0: - current_angle += 360 - else: - # Counterclockwise rotation - current_angle = start_angle - while True: - # Append the current angle to the angles list - angles.append(current_angle % 360) - - # Convert the current angle to radians - current_angle_rad = math.radians(current_angle % 360) - - # Convert polar to Cartesian coordinates - x = radius * math.cos(current_angle_rad) - y = radius * math.sin(current_angle_rad) - - # Append the (x, y) pair to the list - coordinates.append((x, y)) - - # Check if we've reached the end_angle (handling wrap-around) (current_angle + step_size) % 360 == end_angle or - if current_angle % 360 == end_angle: - break - - # Increment the current angle by the step size - current_angle += step_size - if current_angle >= 360: - current_angle -= 360 - - return [angles, coordinates] - - -def b_field_rotation(instr1:pyvisa.resources.Resource, instr2:pyvisa.resources.Resource, - Babs:float, startangle:float, endangle:float, angle_stepsize:float, Settings:str, clockwise=True, base_file_name='', zerowhenfin_bool=False)->None: - # TODO: update docs - """Rotation of the b-field in discrete steps, spectrum is measured at each discrete step in the rotation. Scan angle is - defined as the angle between the x-axis and the current B-field vector, i.e., in the anticlockwise direction. - - Args: - instr1 (pyvisa.resources.Resource): _description_ - instr2 (pyvisa.resources.Resource): _description_ - Babs (float): absolute B-field value in T - startangle (float): start angle in degrees - endangle (float): end angle in degrees - angle_stepsize (float): angle step size in degrees - clockwise (bool): determines the direction of rotation of the B-field. Defaults to True. - zerowhenfin_bool (bool, optional): after finishing the rotation, both B-field components should be set to 0 T. Defaults to False. - """ - - # TODO: possibly rename instr1 and instr2 to the dual and single power supplies respectively?? - - # defines the folder, in which the data from the spectrometer is temporarily stored in - temp_folder_path = "C:/Users/localadmin/Desktop/Users/Lukas/2024_02_08_Map_test" - - if base_file_name =='': - base_file_name = datetime.datetime.now().strftime('%Y_%m_%d_%H.%M') - - start_time = time.time() # start of the scan function - - startangle = startangle % 360 - endangle = endangle % 360 # ensures that the angles are within [0,360) - - idnstr1 = query_no_echo(instr1, '*IDN?') - idnstr2 = query_no_echo(instr1, '*IDN?') - - intensity_data = [] # To store data from each scan - cwd = os.getcwd() # save original directory - - # find which one is the dual power supply, then, ramp B_x to Babs value - if '2301034' in idnstr1: # serial no. the dual power supply - pass - elif '2101034' in idnstr2: - # swap instruments, instr 1 to be the dual power supply (^= x-axis) - instr1, instr2 = instr2, instr1 - - # save initial low and high sweep limits of each device, and set them back after the rotation - instr1_bsettings = list(sep_num_from_units(el) for el in query_no_echo(instr1, 'UNITS?;LLIM?;ULIM?').split(';')) # deliver a 3 element tuple of tuples containing the set unit, llim and ulim - instr2_bsettings = list(sep_num_from_units(el) for el in query_no_echo(instr2, 'UNITS?;LLIM?;ULIM?').split(';')) # deliver a 3 element tuple of tuples containing the set unit, llim and ulim - if instr1_bsettings[0][0] == 'T': - instr1_bsettings[1][0] = instr1_bsettings[1][0]*0.1 # rescale kG to T, device accepts values only in kG or A, eventho we set it to T - instr1_bsettings[2][0] = instr1_bsettings[2][0]*0.1 - if instr2_bsettings[0][0] == 'T': - instr2_bsettings[1][0] = instr2_bsettings[1][0]*0.1 # rescale kG to T, device accepts values only in kG or A, eventho we set it to T - instr2_bsettings[2][0] = instr2_bsettings[2][0]*0.1 - - # initialise the sweep angle list as well as the sweep limits and directions for each instrument - instr1_lim, instr2_lim = 'LLIM', 'ULIM' - instr1_sweep, instr2_sweep = 'DOWN', 'UP' - - # create lists of angles and discrete Cartesian coordinates - angles, cartesian_coords = polar_to_cartesian(Babs, startangle, endangle, angle_stepsize, clockwise=clockwise) - - if clockwise: # NOTE: old conditional was: startangle > endangle see if this works.... - # reverse sweep limits and directions for the clockwise rotation - instr1_lim, instr2_lim = instr2_lim, instr1_lim - instr1_sweep, instr2_sweep = instr2_sweep, instr1_sweep - - # list of rates (with units) for diff ranges of each device, only up to Range 1 for single power supply as that is already - # the max recommended current. - init_range_lst1 = list(sep_num_from_units(el) for el in query_no_echo(instr1, 'RATE? 0;RATE? 1;RATE? 2').split(';')) - init_range_lst2 = list(sep_num_from_units(el) for el in query_no_echo(instr2, 'RATE? 0;RATE? 1').split(';')) - - min_range_lst = [min(el1[0], el2[0]) for el1,el2 in zip(init_range_lst1, init_range_lst2)] # min rates for each given range - - # set both devices to the min rates - write_no_echo(instr1, f'RATE 0 {min_range_lst[0]};RATE 1 {min_range_lst[1]}') - write_no_echo(instr2, f'RATE 0 {min_range_lst[0]};RATE 1 {min_range_lst[1]}') - - write_no_echo(instr1, f'CHAN 2;ULIM {Babs*10};SWEEP UP') # sets to B_x, the B_x upper limit and sweeps the magnet field to the upper limit - print(f'SWEEPING B-X TO {Babs} T NOW') - - # wait for Babs to be reached by the Bx field - actual_bval = sep_num_from_units(query_no_echo(instr1, 'IMAG?'))[0]*0.1 # convert kG to T - print(f'Actual magnet strength (Bx): {actual_bval} T,', f'Target magnet strength: {Babs} T') - while abs(actual_bval - Babs) > 0.0001: - time.sleep(5) # little break - actual_bval = sep_num_from_units(query_no_echo(instr1, 'IMAG?'))[0]*0.1 - print(f'Actual magnet strength (Bx): {actual_bval} T,', f'Target magnet strength: {Babs} T') - - # NOTE: implement PID control, possibly best option to manage the b field DO THIS LATER ON, WE DO DISCRETE B VALUES RN - # Helper function that listens to a device - def listen_to_device(device_id, target_value, shared_values, lock, all_targets_met_event): - while not all_targets_met_event.is_set(): # Loop until the event is set - # value = 0 # Simulate receiving a float from the device INSERT QUERY NO ECHO HERE TO ASK FOR DEVICE IMAG - if '2301034' in device_id: - value = sep_num_from_units(query_no_echo(instr1, 'IMAG?'))[0]*0.1 # convert kG to T - elif '2101014' in device_id: - value = sep_num_from_units(query_no_echo(instr2, 'IMAG?'))[0]*0.1 # convert kG to T - print(f"Device {device_id} reports value: {value} T") - - with lock: - shared_values[device_id] = value - # Check if both devices have met their targets - if all(shared_values.get(device) is not None and abs(shared_values[device] - target_value[device]) <= 0.0001 - for device in shared_values): - print(f"Both devices reached their target values: {shared_values}") - all_targets_met_event.set() # Signal that both targets are met - - # time.sleep(1) # Simulate periodic data checking - - # Main function to manage threads and iterate over target values - def monitor_devices(device_target_values, angles_lst, intensity_data=intensity_data): - for iteration, target in enumerate(device_target_values): - print(f"\nStarting iteration {iteration+1} for target values: {target}") - # Shared dictionary to store values from devices - shared_values = {device: None for device in target.keys()} - # Event to signal when both target values are reached - all_targets_met_event = threading.Event() - - # Lock to synchronize access to shared_values - lock = threading.Lock() - - # Create and start threads for each device - threads = [] - for device_id in target.keys(): - thread = threading.Thread(target=listen_to_device, args=(device_id, target, shared_values, lock, all_targets_met_event)) - threads.append(thread) - thread.start() - - # Wait until both devices meet their target values - all_targets_met_event.wait() - print(f"Both target values for iteration {iteration+1} met. Performing action...") - - # Perform some action after both targets are met - # we acquire with the LF - acquire_name_spe = f'{base_file_name}_{angles_lst[iteration]}°' # NOTE: save each intensity file with the given angle - AcquireAndLock(acquire_name_spe) #this creates a .spe file with the scan name. - - # read the .spe file and get the data as loaded_files - cwd = os.getcwd() # save original directory - os.chdir(temp_folder_path) #change directory - loaded_files = sl.load_from_files([acquire_name_spe + '.spe']) # get the .spe file as a variable - os.chdir(cwd) # go back to original directory - - # Delete the created .spe file from acquiring after getting necessary info - spe_file_path = os.path.join(temp_folder_path, acquire_name_spe + '.spe') - os.remove(spe_file_path) - - # points_left = total_points - i - 1 # TODO: SEE IF THIS IS CORRECT - # print('Points left in the scan: ', points_left) - - #append the intensity data as it is (so after every #of_wl_points, the spectrum of the next point begins) - intensity_data.append(loaded_files.data[0][0][0]) - - # Clean up threads - for thread in threads: - thread.join() - - print(f"Threads for iteration {iteration+1} closed.\n") - - #prints total time the mapping lasted - end_time = time.time() - elapsed_time = (end_time - start_time) / 60 - print('Scan time: ', elapsed_time, 'minutes') - - # reset both devices to original sweep limits - write_no_echo(instr1, f'LLIM {instr1_bsettings[1][0]*10};ULIM {instr1_bsettings[2][0]*10}') # reset the initial limits of the device after the scan - write_no_echo(instr2, f'LLIM {instr2_bsettings[1][0]*10};ULIM {instr2_bsettings[2][0]*10}') # reset the initial limits of the device after the scan - # reset both devices' initial rates for each range - write_no_echo(instr1, f'RANGE 0 {init_range_lst1[0][0]};RANGE 1 {init_range_lst1[1][0]};RANGE 2 {init_range_lst1[2][0]}') # reset the initial limits of the device after the scan - write_no_echo(instr2, f'RANGE 0 {init_range_lst2[0][0]};RANGE 1 {init_range_lst2[1][0]}') # reset the initial limits of the device after the scan - - if zerowhenfin_bool: - write_no_echo(instr1, 'SWEEP ZERO') # if switched on, discharges the magnet after performing the measurement loop above - write_no_echo(instr2, 'SWEEP ZERO') - - #save intensity & WL data as .txt - os.chdir('C:/Users/localadmin/Desktop/Users/Lukas') - # creates new folder for MAP data - new_folder_name = "Test_Map_" + f"{datetime.datetime.now().strftime('%Y_%m_%d_%H.%M')}" - os.mkdir(new_folder_name) - # Here the things will be saved in a new folder under user Lukas ! - # IMPORTANT last / has to be there, otherwise data cannot be saved and will be lost!!!!!!!!!!!!!!!! - os.chdir('C:/Users/localadmin/Desktop/Users/Lukas/'+ new_folder_name) - - intensity_data = np.array(intensity_data) - np.savetxt(Settings + f'{angles[0]}°_to_{angles[-1]}°' + experiment_name +'.txt', intensity_data) - # TODO: remove/edit experiment_name in line above, as well in sweep_b_val func, rn takes a global variable below - - wl = np.array(loaded_files.wavelength) - np.savetxt("Wavelength.txt", wl) - - # modify cartesian_coords to suite the required data struct in monitor_devices - cartesian_coords = [{'2301034': t[0], '2101014': t[1]} for t in cartesian_coords] - - # call the helper function to carry out the rotation/measurement of spectrum - monitor_devices(cartesian_coords, angles, intensity_data) - - -################################################################# END OF FUNCTION DEFS ########################################################################################### - -# NOTE: RYAN INTRODUCED SOME FUNCTIONS HERE TO PERFORM THE SCAN - -# Initialise PYVISA ResourceManager -rm = pyvisa.ResourceManager() -# print(rm.list_resources()) -# 'ASRL8::INSTR' for dual power supply, 'ASRL9::INSTR' for single power supply (online PC) -# 'ASRL10::INSTR' for dual power supply, 'ASRL12::INSTR' for single power supply (offline PC) - - -# Open the connection with the APS100 dual power supply -powerbox_dualsupply = rm.open_resource('ASRL10::INSTR', - baud_rate=9600, - data_bits=8, - parity= pyvisa.constants.Parity.none, - stop_bits= pyvisa.constants.StopBits.one, - timeout=100)# 5000 ms timeout - -# Open the connection with the APS100 dual power supply -powerbox_singlesupply = rm.open_resource('ASRL12::INSTR', - baud_rate=9600, - data_bits=8, - parity= pyvisa.constants.Parity.none, - stop_bits= pyvisa.constants.StopBits.one, - timeout=100)# 5000 ms timeout - -write_no_echo(powerbox_dualsupply, 'REMOTE') # turn on the remote mode -write_no_echo(powerbox_singlesupply, 'REMOTE') # turn on the remote mode - -# TODO: test functionality of the magnet_coil param later on, should work... as this code below is basically implemented inside the scan func. -# select axis for the dual supply, either z-axis(CHAN 1 ^= Supply A) or x-axis(CHAN 2 ^= Supply B) -# write_no_echo(powerbox_dualsupply, 'CHAN 1') - -# Setup connection to AMC -amc = AMC.Device(IP) -amc.connect() - -# Internally, axes are numbered 0 to 2 -amc.control.setControlOutput(0, True) -amc.control.setControlOutput(1, True) - - -auto = Automation(True, List[String]()) -experiment = auto.LightFieldApplication.Experiment -acquireCompleted = AutoResetEvent(False) - -experiment.Load("Lukas_experiment_2024_02_06") -experiment.ExperimentCompleted += experiment_completed # we are hooking a listener. -# experiment.SetValue(SpectrometerSettings.GratingSelected, '[750nm,1200][0][0]') -# InitializerFilenameParams() - - -#set scan range and resolution in nanometers -range_x = 20000 -range_y = 20000 -resolution = 1000 -# set B-field scan range and resolution (all in T) -set_llim_bval = -0.01 -set_ulim_bval = 0.01 -set_res_bval = 0.01 - -#Here you can specify the filename of the map e.g. put experiment type, exposure time, used filters, etc.... -experiment_settings = 'PL_SP_700_LP_700_HeNe_52muW_exp_2s_Start_' -#The program adds the range of the scan as well as the resolution and the date and time of the measurement -experiment_name = f"{set_llim_bval}T_to_{set_ulim_bval}T_{set_res_bval}T_{datetime.datetime.now().strftime('%Y_%m_%d_%H%M')}" - -# this moves the probe in xy-direction and measures spectrum there -# move_scan_xy(range_x, range_y, resolution, experiment_settings, experiment_name) - -# perform the B-field measurement for selected axis above -# sweep_b_val(powerbox_dualsupply, set_llim_bval, set_ulim_bval, set_res_bval, experiment_settings, experiment_name) -sweep_b_val(powerbox_dualsupply, set_llim_bval, set_ulim_bval, set_res_bval, 'z-axis', - experiment_settings, experiment_name, zerowhenfin_bool=True, reversescan_bool=False) - -# Internally, axes are numbered 0 to 2 - -write_no_echo(powerbox_dualsupply, 'LOCAL') # turn off the remote mode -write_no_echo(powerbox_singlesupply, 'LOCAL') # turn off the remote mode -# time.sleep(0.5) -powerbox_dualsupply.close() -powerbox_singlesupply.close() \ No newline at end of file