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

Feature/phoenix trigger #3

Merged
wakonig_k merged 3 commits from feature/phoenix_trigger into main 2024-08-28 14:32:43 +02:00
Conversation 6 Commits 3 Files Changed 6 +267 -371
6 changed files with 267 additions and 371 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -7,6 +7,7 @@
**/.vscode
**/.pytest_cache
**/*.egg*
*.gz
# recovery_config files
recovery_config_*

19
phoenix_bec/device_configs/phoenix_devices.yaml
View File

@@ -1,3 +1,20 @@
falcon_nohdf5:
description: Falcon detector x-ray fluoresence II
deviceClass: phoenix_bec.devices.falcon_phoenix_no_hdf5.FalconPhoenix
deviceConfig:
prefix: 'X07MB-SITORO:'
deviceTags:
- phoenix
- falcon
- no hdf5
- phoenix_devices.yaml
onFailure: buffer
enabled: true
readoutPriority: async
softwareTrigger: false
###################################################
#
# phoenix standard devices (motors)
@@ -16,7 +33,7 @@ PH_TTL:
onFailure: buffer
enabled: true
readoutPriority: monitored
softwareTrigger: false
softwareTrigger: true
PH_Dummy:

68
phoenix_bec/devices/falcon_phoenix_no_hdf5.py
View File

@@ -97,6 +97,7 @@ class FalconHDF5Plugins(Device):
array_counter = Cpt(EpicsSignalWithRBV, "ArrayCounter", kind="config")
"""
class FalconSetup(CustomDetectorMixin):
"""
Falcon setup class for cSAXS
@@ -113,7 +114,7 @@ class FalconSetup(CustomDetectorMixin):
"""Initialize Falcon detector"""
self.initialize_default_parameter()
self.initialize_detector()
self.initialize_detector_backend()
# self.initialize_detector_backend()
def initialize_default_parameter(self) -> None:
"""
@@ -139,12 +140,12 @@ class FalconSetup(CustomDetectorMixin):
def initialize_detector(self) -> None:
"""Initialize Falcon detector"""
self.stop_detector()
self.stop_detector_backend()
# self.stop_detector_backend()
self.set_trigger(
mapping_mode=MappingSource.MAPPING, trigger_source=TriggerSource.GATE, ignore_gate=0
mapping_mode=MappingSource.SPECTRUM, trigger_source=TriggerSource.GATE, ignore_gate=0
)
# 1 Realtime
self.parent.preset_mode.put(1)
self.parent.preset_mode.put(0)
# 0 Normal, 1 Inverted
self.parent.input_logic_polarity.put(0)
# 0 Manual 1 Auto
@@ -154,30 +155,34 @@ class FalconSetup(CustomDetectorMixin):
def initialize_detector_backend(self) -> None:
"""Initialize the detector backend for Falcon."""
w=0
#----------------------------------------------------------------------
#self.parent.hdf5.enable.put(1)
w = 0
# ----------------------------------------------------------------------
# self.parent.hdf5.enable.put(1)
# file location of h5 layout for cSAXS
#self.parent.hdf5.xml_file_name.put("layout.xml")
# self.parent.hdf5.xml_file_name.put("layout.xml")
# TODO Check if lazy open is needed and wanted!
#self.parent.hdf5.lazy_open.put(1)
#self.parent.hdf5.temp_suffix.put("")
# self.parent.hdf5.lazy_open.put(1)
# self.parent.hdf5.temp_suffix.put("")
# size of queue for number of spectra allowed in the buffer, if too small at high throughput, data is lost
#self.parent.hdf5.queue_size.put(2000)
# self.parent.hdf5.queue_size.put(2000)
# Segmentation into Spectra within EPICS, 1 is activate, 0 is deactivate
#self.parent.nd_array_mode.put(1)
# self.parent.nd_array_mode.put(1)
def on_stage(self) -> None:
"""Prepare detector and backend for acquisition"""
self.prepare_detector()
self.prepare_data_backend()
self.publish_file_location(done=False, successful=False)
# self.prepare_data_backend()
# self.publish_file_location(done=False, successful=False)
# self.arm_acquisition()
def on_trigger(self) -> None:
"""Actions on pre_scan. This is performed AFTER stage, just before scan_core"""
self.arm_acquisition()
def prepare_detector(self) -> None:
"""Prepare detector for acquisition"""
self.set_trigger(
mapping_mode=MappingSource.MAPPING, trigger_source=TriggerSource.GATE, ignore_gate=0
mapping_mode=MappingSource.SPECTRUM, trigger_source=TriggerSource.GATE, ignore_gate=0
)
self.parent.preset_real.put(self.parent.scaninfo.exp_time)
self.parent.pixels_per_run.put(
@@ -186,7 +191,7 @@ class FalconSetup(CustomDetectorMixin):
def prepare_data_backend(self) -> None:
"""Prepare data backend for acquisition"""
w=9
w = 9
""" --------------------------------------------------------------
self.parent.filepath.set(
self.parent.filewriter.compile_full_filename(f"{self.parent.name}.h5")
@@ -230,30 +235,31 @@ class FalconSetup(CustomDetectorMixin):
def on_complete(self) -> None:
"""Complete detector and backend"""
#------------------------------------------------------------------
#self.finished(timeout=self.parent.TIMEOUT_FOR_SIGNALS)
#self.publish_file_location(done=True, successful=True)
w=9
# ------------------------------------------------------------------
# self.finished(timeout=self.parent.TIMEOUT_FOR_SIGNALS)
# self.publish_file_location(done=True, successful=True)
w = 9
def on_stop(self) -> None:
"""Stop detector and backend"""
self.stop_detector()
#self.stop_detector_backend()
# self.stop_detector_backend()
def stop_detector(self) -> None:
"""Stops detector"""
self.parent.stop_all.put(1)
self.parent.erase_all.put(1)
#-------------------------------------------------------------------
#signal_conditions = [
# -------------------------------------------------------------------
# signal_conditions = [
# (lambda: self.parent.state.read()[self.parent.state.name]["value"], DetectorState.DONE)
#]
# ]
#if not self.wait_for_signals(
# if not self.wait_for_signals(
# signal_conditions=signal_conditions,
# timeout=self.parent.TIMEOUT_FOR_SIGNALS - self.parent.TIMEOUT_FOR_SIGNALS // 2,
# all_signals=False,
#):
# ):
# # Retry stop detector and wait for remaining time
# raise FalconTimeoutError(
# f"Failed to stop detector, timeout with state {signal_conditions[0][0]}"
@@ -261,8 +267,8 @@ class FalconSetup(CustomDetectorMixin):
def stop_detector_backend(self) -> None:
"""Stop the detector backend"""
#self.parent.hdf5.capture.put(0)
w=0
# self.parent.hdf5.capture.put(0)
w = 0
def finished(self, timeout: int = 5) -> None:
"""Check if scan finished succesfully"""
@@ -272,7 +278,7 @@ class FalconSetup(CustomDetectorMixin):
)
signal_conditions = [
(self.parent.dxp.current_pixel.get, total_frames),
# (self.parent.hdf5.array_counter.get, total_frames), ---------------------
# (self.parent.hdf5.array_counter.get, total_frames), ---------------------
]
if not self.wait_for_signals(
signal_conditions=signal_conditions,
@@ -288,10 +294,6 @@ class FalconSetup(CustomDetectorMixin):
self.stop_detector()
self.stop_detector_backend()
def set_trigger(
self, mapping_mode: MappingSource, trigger_source: TriggerSource, ignore_gate: int = 0
) -> None:

319
phoenix_bec/devices/phoenix_trigger.py
View File

@@ -1,25 +1,33 @@
import time
from ophyd import (
ADComponent as ADCpt,
Device,
DeviceStatus,
)
import enum
import numpy as np
from ophyd import Component as Cpt
from ophyd import FormattedComponent as FCpt
from ophyd import Device, EpicsSignal, EpicsSignalRO
from ophyd import DeviceStatus, EpicsSignal, EpicsSignalRO, Kind
from ophyd_devices.interfaces.base_classes.psi_detector_base import PSIDetectorBase, CustomDetectorMixin
from ophyd_devices.interfaces.base_classes.psi_detector_base import (
PSIDetectorBase,
CustomDetectorMixin,
)
from bec_lib import bec_logger, messages
from bec_lib.endpoints import MessageEndpoints
from bec_lib import bec_logger
logger = bec_logger.logger
DETECTOR_TIMEOUT = 5
#class PhoenixTriggerError(Exce start_csmpl=Cpt(EPicsSignal,'START-CSMPL') # cont on / off
class PhoenixTriggerError(Exception):
"""PhoenixTrigger specific error"""
class SAMPLINGDONE(int, enum.Enum):
"""Sampling Done PV"""
RUNNING = 0
DONE = 1
class PhoenixTriggerSetup(CustomDetectorMixin):
"""
@@ -28,237 +36,102 @@ class PhoenixTriggerSetup(CustomDetectorMixin):
"""
#self.acquire = self.parent.smpl.put(1)
#self.continuous_sampling_on = parent.start_cmpl.put(1)
#self.continuous_sampling_off = self.parent.start_cmpl.put(0)
def on_stage(self) -> None:
"""Actions to take place on stage"""
if self.parent.scaninfo.scan_type == "step":
self.parent.start_csmpl.set(0)
self.parent.total_cycles.set(1)
self.parent.smpl.put(1)
time.sleep(0.5)
self.parent.total_cycles.set(np.ceil(self.parent.scaninfo.exp_time * 5))
logger.info(f"Device {self.parent.name} was staged for step scan")
def __init__(self, *args, parent:Device = None, **kwargs):
super().__init__(*args, parent=parent, **kwargs)
self._counter = 0
def on_unstage(self) -> None:
"""Actions to take place on unstage"""
self.on_stop()
def on_trigger(self) -> DeviceStatus:
"""Actions to be performed upon receiving a software trigger"""
# Check first that falcon is set to acquiring
falcon = self.parent.device_manager.devices.get("falcon_nohdf5", None)
if falcon is not None:
if self.wait_for_signals([(falcon.state.get, 1)], timeout=1):
raise PhoenixTriggerError(
f"Falcon not ready to take trigger after 1s timeout in trigger"
)
falcon.state.get() == 1 # Acquiring
if self.parent.scaninfo.scan_type == "step":
time.sleep(0.2)
self.parent.smpl.put(1)
# Minimum of 1 cycle has to be waited. Cycle == 0.2s
time.sleep(0.2)
# Trigger function from ophyd.Device returns a DeviceStatus. This function
# starts a process that creates a DeviceStatus, and waits for the signal_conditions
# self.parent.smpl_done.get to change to the value SAMPLINGDONE.DONE
# Once this takes place, the DeviceStatus.done flag will be set to True.
# When BEC calls trigger() on the devices, this method will be called assuming that
# the devices config softwareTrigger=True is set.
# In ScanBase, the _at_each_point function calls
# self.stubs.wait(wait_type="trigger", group="trigger", wait_time=self.exp_time)
# which ensures that the DeviceStatus object resolves before continuing, i.e. DeviceStatus.done = True
status = self.wait_with_status(
signal_conditions=[(self.parent.smpl_done.get, SAMPLINGDONE.DONE)],
timeout=5 * self.parent.scaninfo.exp_time, # Check if timeout is appropriate
check_stopped=True,
)
return status
def on_acquire(self):
self.parent.smpl.put(1)
print('on_aquire')
def on_cont_sample_on(self):
self.parent.start_csmpl.put(1)
print('on_cont_sample_on')
def on_cont_sample_off(self):
self.parent.start_csmpl.put(0)
print('on_cont_sample_off')
def on_done(self):
done_out = self.parent.smpl_done.get()
if done_out =='1':
done=True
if done_out == '0':
done=False
return done
def on_dwell(self,t):
" calculate cycles from time in sec "
cycles=self.parent.total_cycles.put(0)*5
def on_stage(self):
# is this called on each point in scan or just before scan ???
print('on stage')
self.parent.start_csmpl.put(0)
time.sleep(0.05)
cycles=self.parent.total_cycles.get()
time.sleep(0.05)
self.parent.total_cycles.put(0)
time.sleep(0.05)
def on_stop(self) -> None:
"""Actions to stop the Device"""
# Put the Device in cont mode
self.parent.total_cycles.set(5)
self.parent.start_csmpl.set(1)
self.parent.smpl.put(1)
time.sleep(0.5)
print(cycles)
cycles=self.parent.total_cycles.put(cycles)
logger.success('PhoenixTrigger on stage')
def on_unstage(self):
# is this called on each point in scan or just before scan ???
print('on unstage')
#while self.parent.smpl_done.get()
self.parent.total_cycles.put(5)
time.sleep(0.3)
self.parent.start_csmpl.put(1)
time.sleep(0.3)
self.parent.smpl.put(1)
time.sleep(2)
time.sleep(0.2)
if self.parent.smpl_done.get() == SAMPLINGDONE.RUNNING:
return
self.parent.smpl.put(1)
time.sleep(.5)
logger.success('PhoenixTrigger.on_unstage')
#def on_trigger(self):
# print('on_trigger')
# self.parent.start_smpl.put(1)
# logger.success('PhoenixTrigger on_trigger')
#
# return self.wait_with_status(
# [(self.parent.smpl_done.get, 1)])
# logger.success(' PhoenixTrigger on_trigger complete ')
# if success:
# status.set_finished()
# else:
# status.set_exception(TimeoutError())
# return status
#def on_complete(self):
# print('on_complete')
# timeout =10
# logger.success('XXXX complete %d XXXX' % success)
# success = self.wait_for_signals(
# [
# (self.parent.smpl_done.get, 0)
# ],
# timeout,
# check_stopped=True,
# all_signals=True
# )
# if success:
# status.set_finished()
# else:
# status.set_exception(TimeoutError())
# return status
# hoenixTrigger on_unstage ')
# self.parent.csmpl.put(1)
# self.parent.smpl.put(1)
# logger.success(' PhoenixTrigger on_unstage finished ')
#def on_trigger():
# print('on_trigger')
class PhoenixTrigger(PSIDetectorBase):
"""
Docstring:
Class for PHOENIX TTL hardware trigger
Parent class: PSIDetectorBase
class attributes:
custom_prepare_cls (PhoenixTriggerSetup) : Custom setup for TTL trigger at PHOENIX
inherits from CustomDetectorMixin
in __init__ of PSIDetecor bases
class is initialized
self.custom_prepare = self.custom_prepare_cls(parent=self, **kwargs)
The class PhoenixTrigger is the class to be called via yaml configuration file
the input arguments are defined by PSIDetectorBase,
and need to be given in the yaml configuration file.
To adress chanels such as 'X07MB-OP2:SMPL-DONE':
use prefix 'X07MB-OP2:' in the device definition in the yaml configuration file.
PSIDetectorBase(
prefix='',
*,Q
name,
kind=None,
parent=None,
device_manager=None,
**kwargs,
)
Args:
prefix (str): EPICS PV prefix for component (optional)
name (str): name of the device, as will be reported via read()
kind (str): member of class 'ophydobj.Kind', defaults to Kind.normal
omitted -> readout ignored for read 'ophydobj.read()'
normal -> readout for read
config -> config parameter for 'ophydobj.read_configuration()'
hinted -> which attribute is readout for read
parent (object): instance of the parent device
device_manager (object): bec device manager
**kwargs: keyword arguments
File: /data/test/x07mb-test-bec/bec_deployment/ophyd_devices/ophyd_devices/interfaces/base_classes/psi_detector_base.py
Type: type
Subclasses: EpicsSignal
Class for PHOENIX TTL hardware trigger (X07MB-OP2:)
"""
##################################################################
# Specify which functions are revealed to the user in BEC client
# only a set of predefined functions will be visible in dev.TTL
# The Variable USER_ACCESS contains an ascii list of functions which will be
# visible in dev.TTL as well
# Alternatively one couls also create 2nd instance of PhoenixTrigger,
# which is probably not ideal
USER_ACCESS = ["a_acquire"
,"a_cont_sample_on"
,"a_cont_sample_off"
,"prefix"
,"a_done"
,"a_done_cpt"
,"SMPL"]
#####################################################################
# specify Setup class into variable custom_prepare_cls
# in __init__ of PSIDetectorBase will the initialzed by
# self.custom_prepare = self.custom_prepare_cls(parent=self, **kwargs)
# making the instance of PSIDetectorBase availble in functions
custom_prepare_cls = PhoenixTriggerSetup
#############################################################3
# Now use component to provide channel access
# when PhoenixTrigger is initialized, the parameters of the base class are
# inherided, most notable prefix, which is here X07MB-OP2:
# The input of Component=Cpt is Cpt(deviceClass,suffix)
# if Cpt is used in a class, which has interited Device, here via:
# (Here PhoenixTrigger <-- PSIDetectorBase <- Device
# the Cpt will construct - magically- the Epics channel name
# EpicsPV = prefix+suffix,
# for example
# 'X07MB-OP2:' + 'START-CSMPL' -> 'X07MB-OP2:' + 'START-CSMPL'
#
start_csmpl = Cpt(EpicsSignal, 'START-CSMPL') # cont on / off
intr_count = Cpt(EpicsSignal,'INTR-COUNT') # conter run up
total_cycles = Cpt(EpicsSignal,'TOTAL-CYCLES') # cycles set
smpl = Cpt(EpicsSignal,'SMPL') # start sampling --> aquire
# Signal is of type string
smpl_done = Cpt(EpicsSignal,'SMPL-DONE',string=True) # show trigger is done
start_csmpl = Cpt(
EpicsSignal, "START-CSMPL", kind=Kind.config, put_complete=True
) # cont on / off
intr_count = Cpt(
EpicsSignal, "INTR-COUNT", kind=Kind.config, put_complete=True
) # conter run up
total_cycles = Cpt(
EpicsSignal, "TOTAL-CYCLES", kind=Kind.config, put_complete=True
) # cycles set
smpl = Cpt(
EpicsSignal, "SMPL", kind=Kind.config, put_complete=True
) # start sampling --> aquire
smpl_done = Cpt(
EpicsSignalRO, "SMPL-DONE", kind=Kind.config
) # show trigger is done, consider using string=True
def a_acquire(self):
self.custom_prepare.on_acquire()
if __name__ == "__main__":
trigger = PhoenixTrigger(name="trigger", prefix="X07MB-OP2:")
trigger.wait_for_connection(all_signals=True)
trigger.read()
trigger.read_configuration()
def a_cont_sample_on(self):
self.custom_prepare.on_cont_sample_on()
trigger.stage()
status = trigger.trigger()
while status.done is False:
print(f" Waiting for status, flag is {status.done}")
time.sleep(0.2)
def a_cont_sample_off(self):
self.custom_prepare.on_cont_sample_off()
def a_done(self):
done=self.custom_prepare.on_done()
return done
def a_dwell(self):
self.custom_prepare.on_dwell()
trigger.unstage()

51
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')
re1 = res1.to_numpy()
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')
re2 = res2.to_numpy()
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
@@ -81,4 +82,4 @@ print(res2)
print('Scan2 as numpy ')
print(res2)
"""
"""

180
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)
n_outer = len(this_scan.scan.data.keys())
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)
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