bec/phoenix_bec
0
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

refactor: misc changes; playing around

Browse Source
This commit is contained in:
gac-x07mb
2024-08-28 14:18:18 +02:00
parent 1f55dc8736
commit 6c8d5ee202
2 changed files with 117 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
phoenix_bec/local_scripts/PhoenixTemplate.py
View File

@ -1,18 +1,20 @@
"""
Scritpt to be developed as template for phoenic scritps
"""
#from unittest import mock
# from unittest import mock
import numpy as np
#import pandas
#import pytest
#from bec_lib import messages
#import device_server
#from ophyd importPhoenixTemplate.pyitioner import PVPositionerComparator
#from ophyd.status import DeviceStatus, SubscriptionStatus
# import pandas
# import pytest
# from bec_lib import messages
# import device_server
# from ophyd importPhoenixTemplate.pyitioner import PVPositionerComparator
# from ophyd.status import DeviceStatus, SubscriptionStatus
import time
#import ophyd
# import ophyd
import os
import sys
import importlib
@ -21,39 +23,38 @@ import ophyd
#
phoenix.add_phoenix_config()
#bec.config.update_session_with_file('./ConfigPHOENIX/device_config/phoenix_devices.yaml')
# bec.config.update_session_with_file('./ConfigPHOENIX/device_config/phoenix_devices.yaml')
time.sleep(1)
w1.
s1=scans.line_scan(dev.ScanX,0,0.1,steps=4,exp_time=.2,relative=False,delay=2)
s1 = scans.line_scan(dev.ScanX, 0, 0.1, steps=4, exp_time=0.2, relative=False, delay=2)
s2=scans.phoenix_line_scan(dev.ScanX,0,0.1,steps=4,exp_time=.2,relative=False,delay=2)
s2 = scans.phoenix_line_scan(dev.ScanX, 0, 0.1, steps=4, exp_time=0.2, relative=False, delay=2)
res1 = s1.scan.to_pandas()
re1 = res1.to_numpy()
w1=PH.PhGroup('Bec Linescan')
w1 = PH.PhGroup("Bec Linescan")
w1.linescan2group(s1)
print('res1')
print("res1")
print(res1)
print('as numpy')
print('re1')
print("as numpy")
print("re1")
res2 = s2.scan.to_pandas()
re2 = res2.to_numpy()
w2=PH.PhGroup('PHOENIX Linescan')
w2 = PH.PhGroup("PHOENIX Linescan")
w2.linescan2group(s2)
print('res2')
print("res2")
print(res2)
print('as numpy')
print('re2')
print("as numpy")
print("re2")
print (s1)
print('---------------------------------')
print(s1)
print("---------------------------------")
"""
# scan will not diode

176
phoenix_bec/scripts/phoenix.py
View File

@ -1,43 +1,46 @@
#from unittest import mock
# from unittest import mock
import os
import sys
import time
import numpy as np
#import pandas
#import pytest
#from bec_lib import messages
#import device_server
#from ophyd import Component as Cpt
# import pandas
# import pytest
# from bec_lib import messages
# import device_server
# from ophyd import Component as Cpt
from ophyd import Device, EpicsMotor, EpicsSignal, EpicsSignalRO
#from ophyd import FormattedComponent as FCpt
#from ophyd import Kind, PVPositioner, Signal
#from ophyd.flyers import FlyerInterface
#from ophyd.pv_positioner import PVPositionerComparator
#from ophyd.status import DeviceStatus, SubscriptionStatus
# from ophyd import FormattedComponent as FCpt
# from ophyd import Kind, PVPositioner, Signal
# from ophyd.flyers import FlyerInterface
# from ophyd.pv_positioner import PVPositionerComparator
# from ophyd.status import DeviceStatus, SubscriptionStatus
from bec_lib.config_helper import ConfigHelper
from bec_lib.logger import bec_logger
logger = bec_logger.logger
#import ophyd
#logger = bec_logger.logger
# import ophyd
# logger = bec_logger.logger
# load simulation
#bec.config.load_demo_config()
# bec.config.load_demo_config()
# .. define base path for directory with scripts
class PhoenixBL():
class PhoenixBL:
"""
#
# General class for PHOENIX beamline located in phoenix_bec/phoenic_bec/scripts
#
"""
t0=time.time()
t0 = time.time()
def __init__(self):
"""
init PhoenixBL() in phoenix_bec/scripts
@ -45,50 +48,60 @@ class PhoenixBL():
"""
import os
print('..... init PhoenixBL from phoenix_bec/scripts/phoenix.py')
print("..... init PhoenixBL from phoenix_bec/scripts/phoenix.py")
# load local configuration
self.path_scripts_local = '/data/test/x07mb-test-bec/bec_deployment/phoenix_bec/phoenix_bec/local_scripts/'
self.path_config_local = self.path_scripts_local + 'TEST_ConfigPhoenix/' # base dir for local configurations
self.path_devices_local = self.path_config_local + 'Local_device_config/' # local yamal file
self.file_devices_file_local = self.path_devices_local + 'phoenix_devices.yaml'
self.path_phoenix_bec ='/data/test/x07mb-test-bec/bec_deployment/phoenix_bec/'
self.path_devices = self.path_phoenix_bec + 'phoenix_bec/device_configs/' # local yamal file
self.file_devices_file = self.path_phoenix_bec + 'phoenix_bec/device_configs/phoenix_devices.yaml' # local yamal file
self.t0=time.time()
self.path_scripts_local = (
"/data/test/x07mb-test-bec/bec_deployment/phoenix_bec/phoenix_bec/local_scripts/"
)
self.path_config_local = (
self.path_scripts_local + "TEST_ConfigPhoenix/"
) # base dir for local configurations
self.path_devices_local = (
self.path_config_local + "Local_device_config/"
) # local yamal file
self.file_devices_file_local = self.path_devices_local + "phoenix_devices.yaml"
self.path_phoenix_bec = "/data/test/x07mb-test-bec/bec_deployment/phoenix_bec/"
self.path_devices = (
self.path_phoenix_bec + "phoenix_bec/device_configs/"
) # local yamal file
self.file_devices_file = (
self.path_phoenix_bec + "phoenix_bec/device_configs/phoenix_devices.yaml"
) # local yamal file
self.t0 = time.time()
def read_local_phoenix_config(self):
print('read file ')
print("read file ")
print(self.file_phoenix_devices_file)
bec.config.update_session_with_file(self.file_devices_file_local)
def add_phoenix_config(self):
print('add_phoenix_config ')
print('self.file_devices_file')
print("add_phoenix_config ")
print("self.file_devices_file")
bec.config.update_session_with_file(self.file_devices_file)
def add_xmap(self):
print('add xmap ')
print(self.path_devices+'phoenix_xmap.yaml')
print("add xmap ")
print(self.path_devices + "phoenix_xmap.yaml")
bec.config.update_session_with_file(self.path_devices+'phoenix_xmap.yaml')#,timeout=100)
bec.config.update_session_with_file(
self.path_devices + "phoenix_xmap.yaml"
) # ,timeout=100)
def add_falcon(self):
print('add_xmap')
print(self.path_devices+'/phoenix_falcon.yaml')
bec.config.wait_for_config_reply()
bec.config.update_session_with_file(self.path_devices+'/phoenix_falcon.yaml')
print("add_xmap")
print(self.path_devices + "/phoenix_falcon.yaml")
# bec.config.wait_for_config_reply()
bec.config.update_session_with_file(self.path_devices + "/phoenix_falcon.yaml")
def show_phoenix_setup(self):
print(self.path_phoenix_bec)
os.system('cat '+self.path_phoenix_bec+'phoenix_bec/scripts/Current_setup.txt')
os.system("cat " + self.path_phoenix_bec + "phoenix_bec/scripts/Current_setup.txt")
@classmethod
def my_log(cls,x):
def my_log(cls, x):
"""
class method allows to write a user defined log file
time is seconds relative to some point max 10 minutes ago
@ -96,21 +109,19 @@ class PhoenixBL():
"""
print(time.time())
now = time.time() - (86400*(time.time()//86400))
now = now - 3600.*(now//3600.)
now = now - 600.*(now//600.)
m=str(now)+' sec '+x
now = time.time() - (86400 * (time.time() // 86400))
now = now - 3600.0 * (now // 3600.0)
now = now - 600.0 * (now // 600.0)
m = str(now) + " sec " + x
logger.success(m)
file=open('MyLogfile.txt','a')
file.write(m+'\n')
file = open("MyLogfile.txt", "a")
file.write(m + "\n")
file.close
class PhGroup():
class PhGroup:
"""
Class to create data groups
compatible with larch groups
@ -134,21 +145,20 @@ class PhGroup():
"""
def __init__(self,description):
def __init__(self, description):
setattr(self,'description',description)
setattr(self, "description", description)
# atribute 'label' for compatibility woith La groups...
setattr(self,'label',description)
#if type(NameTag)==list:
setattr(self, "label", description)
# if type(NameTag)==list:
# for i in NameTag:
# setattr(self,i,None)
# #endfor
#else:
# else:
# setattr(self,NameTag,None)
#endif
# endif
def add(self,NameTag,content):
def add(self, NameTag, content):
"""
Add tags to group...
@ -165,10 +175,9 @@ class PhGroup():
"""
setattr(self,NameTag,content)
setattr(self, NameTag, content)
def keys(self):
"""
Method gets all atributes, which are not methods
and which do not start with __
@ -180,19 +189,18 @@ class PhGroup():
DESCRIPTION.
"""
box=[]
box = []
for i in self.__dir__():
if '__' not in i:
#print(i)
if "__" not in i:
# print(i)
if str(type(self.__getattribute__(i))) != "<class 'method'>":
box.append(i)
#endif
#endfor
# endif
# endfor
return box
def linescan2group(self,this_scan):
def linescan2group(self, this_scan):
"""
method merges results of linescan into group and
@ -204,37 +212,31 @@ class PhGroup():
"""
print('keys')
print("keys")
print(this_scan.scan.data.keys())
for outer_key in this_scan.scan.data.keys():
print('outer_key',outer_key)
print("outer_key", outer_key)
n_outer = len(this_scan.scan.data.keys())
for inner_key in this_scan.scan.data[outer_key].keys():
print('inner_key',inner_key)
print("inner_key", inner_key)
# calculate nunber of points
n_inner = len(this_scan.scan.data[outer_key][inner_key].keys())
value = np.zeros(n_inner)
timestamp = np.zeros(n_inner)
for i in range(n_inner):
try:
value[i] = this_scan.scan.data[outer_key][inner_key][i]['value']
value[i] = this_scan.scan.data[outer_key][inner_key][i]["value"]
except:
value=None
value = None
try:
timestamp[i] = this_scan.scan.data[outer_key][inner_key][i]['timestamp']
timestamp[i] = this_scan.scan.data[outer_key][inner_key][i]["timestamp"]
except:
timestamp[i]=None
#endfor
self.add(inner_key+'_'+ outer_key+'_val',value)
self.add(inner_key+'_'+ outer_key+'_ts',timestamp)
#endfor
#endfor
#endfor
#enddef
timestamp[i] = None
# endfor
self.add(inner_key + "_" + outer_key + "_val", value)
self.add(inner_key + "_" + outer_key + "_ts", timestamp)
# endfor
# endfor
# endfor
# enddef