From 3423a978b65c21216c26fea6c2dfa2d2acd39cd7 Mon Sep 17 00:00:00 2001 From: appel_c Date: Wed, 28 Aug 2024 16:13:25 +0200 Subject: [PATCH] fix: remove copy/paste bug --- phoenix_bec/scans/phoenix_scans.py | 117 ++++++++++++++--------------- 1 file changed, 56 insertions(+), 61 deletions(-) diff --git a/phoenix_bec/scans/phoenix_scans.py b/phoenix_bec/scans/phoenix_scans.py index dbbad2e..578772c 100644 --- a/phoenix_bec/scans/phoenix_scans.py +++ b/phoenix_bec/scans/phoenix_scans.py @@ -23,90 +23,87 @@ but they are executed in a specific order: """ # imports in ScanBase -#from __future__ import annotations +# from __future__ import annotations -#import ast -#import enum -#import threading -#import time -#import uuid -#from abc import ABC, abstractmethod -#from typing import Any, Literal +# import ast +# import enum +# import threading +# import time +# import uuid +# from abc import ABC, abstractmethod +# from typing import Any, Literal -#import numpy as np +# import numpy as np -#from bec_lib.device import DeviceBase -#from bec_lib.devicemanager import DeviceManagerBase -#from bec_lib.endpoints import MessageEndpoints -#from bec_lib.logger import bec_logger +# from bec_lib.device import DeviceBase +# from bec_lib.devicemanager import DeviceManagerBase +# from bec_lib.endpoints import MessageEndpoints +# from bec_lib.logger import bec_logger -#from .errors import LimitError, ScanAbortion -#from .path_optimization import PathOptimizerMixin -#from .scan_stubs import ScanStubs +# from .errors import LimitError, ScanAbortion +# from .path_optimization import PathOptimizerMixin +# from .scan_stubs import ScanStubs # end imports in ScanBase # import time # import numpy as np +import time + +import numpy as np + # from bec_lib.endpoints import MessageEndpoints from bec_lib.logger import bec_logger +from bec_server.scan_server.scans import ScanArgType, ScanBase + +from phoenix_bec.scripts.phoenix import PhoenixBL + # from bec_lib import messages # from bec_server.scan_server.errors import ScanAbortion # from bec_server.scan_server.scans import FlyScanBase, RequestBase, ScanArgType, ScanBase # logger = bec_logger.logger -from bec_server.scan_server.scans import ScanBase, ScanArgType -import numpy as np -import time -from bec_lib.logger import bec_logger -from phoenix_bec.scripts.phoenix import PhoenixBL - logger = bec_logger.logger -class LogTime(): +class LogTime: def __init__(self): - logger.success('init LogTime') - self.t0=time.time() + logger.success("init LogTime") + self.t0 = time.time() - def p_s(self,x): - now=time.time() - #delta=now-self.t0 - m=str(now)+' sec '+x - logger.success(m)custom_prepare_cls(parent=self, **kwargs) - # making the instance of PSID - #self.t0=now - file=open('MyLogfile.txt','a') - file.write(m+'\n') + def p_s(self, x): + now = time.time() + # delta=now-self.t0 + m = str(now) + " sec " + x + logger.success(m) + # making the instance of PSID + # self.t0=now + file = open("MyLogfile.txt", "a") + file.write(m + "\n") file.close - - - class PhoenixScanBaseTTL(ScanBase): """ Base scan cl p_s('init scrips.phoenix.scans.PhoenixLineScan') """ - - def scan_core(self): """perform the scan core procedure""" - self.p_s('PhoenixScanBaseTT.scan_core') + self.p_s("PhoenixScanBaseTT.scan_core") for ind, pos in self._get_position(): for self.burst_index in range(self.burst_at_each_point): - self.p_s('PhoenixScanBaseTT.scan_core in loop ') + self.p_s("PhoenixScanBaseTT.scan_core in loop ") yield from self._at_each_point(ind, pos) self.burst_index = 0 def _at_each_point(self, ind=None, pos=None): - self.p_s('PhoenixScanBaseTT._at_each_point') + self.p_s("PhoenixScanBaseTT._at_each_point") yield from self._move_scan_motors_and_wait(pos) if ind > 0: yield from self.stubs.wait( @@ -123,12 +120,11 @@ class PhoenixScanBaseTTL(ScanBase): ) self.point_id += 1 - self.p_s('done') + self.p_s("done") + class PhoenixLineScan(PhoenixScanBaseTTL): - - scan_name = "phoenix_line_scan" required_kwargs = ["steps", "relative"] arg_input = { @@ -149,25 +145,25 @@ class PhoenixLineScan(PhoenixScanBaseTTL): steps: int = None, relative: bool = False, burst_at_each_point: int = 1, - setup_device:str= None, + setup_device: str = None, **kwargs, ): """ - A phoenix line scan for one or more motors. + A phoenix line scan for one or more motors. - Args: - *args (Device, float, float): pairs of device / start position / end position - exp_time (float): exposure time in s. Default: 0 - steps (int): number of steps. Default: 10 - relative (bool): if True, the start and end positions are relative to the current position. Default: False - burst_a Specifies the level of type checking analysis to perform. -ans.line_scan(dev.motor1, -5, 5, dev.motor2, -5, 5, steps=10, exp_time=0.1, relative=True) + Args: + *args (Device, float, float): pairs of device / start position / end position + exp_time (float): exposure time in s. Default: 0 + steps (int): number of steps. Default: 10 + relative (bool): if True, the start and end positions are relative to the current position. Default: False + burst_a Specifies the level of type checking analysis to perform. + ans.line_scan(dev.motor1, -5, 5, dev.motor2, -5, 5, steps=10, exp_time=0.1, relative=True) """ - #from phoenix_bec.scripts.phoenix import PhoenixBL - self.p_s=PhoenixBL.my_log + # from phoenix_bec.scripts.phoenix import PhoenixBL + self.p_s = PhoenixBL.my_log - self.p_s('init scripts.phoenix.scans.PhoenixLineScan') + self.p_s("init scripts.phoenix.scans.PhoenixLineScan") super().__init__( exp_time=exp_time, relative=relative, burst_at_each_point=burst_at_each_point, **kwargs @@ -176,14 +172,13 @@ ans.line_scan(dev.motor1, -5, 5, dev.motor2, -5, 5, steps=10, exp_time=0.1, rela self.setup_device = setup_device time.sleep(1) - self.p_s('done') + self.p_s("done") def _calculate_positions(self) -> None: - self.p_s('PhoenixLineScan._calculate_positions') + self.p_s("PhoenixLineScan._calculate_positions") axis = [] for _, val in self.caller_args.items(): ax_pos = np.linspace(val[0], val[1], self.steps, dtype=float) axis.append(ax_pos) self.positions = np.array(list(zip(*axis)), dtype=float) - self.p_s('done') - + self.p_s("done")