Commit 31c4e12e authored by Jerome Carnis's avatar Jerome Carnis 🦊
Browse files

deviceEigerP11.py: add some types

parent b51774c4
Pipeline #28453 passed with stages
in 8 minutes and 42 seconds
......@@ -4,7 +4,7 @@ import os
import time
from time import sleep
from threading import Thread
from typing import Any, Dict, List, Optional, Tuple
from typing import Any, Dict, List, Optional, Tuple, Union
import numpy as np
......@@ -26,7 +26,9 @@ from kamzik3.constants import (
from kamzik3.devices.device import Device
from kamzik3.devices.deviceTango import DeviceTango
from kamzik3.devices.general.deviceScanner import DeviceScanner
from kamzik3.macro.macro import Macro
from kamzik3.macro.scan import Scan
from kamzik3.macro.step import Step, StepDeviceMethod
from kamzik3.snippets.snippetsDecorators import expose_method
......@@ -326,7 +328,9 @@ class EigerScanner(DeviceScanner):
DeviceScanner.__init__(self, device_id, config)
self.connect()
def _recount_steps(self, parent_macro, to_link=None) -> int:
def _recount_steps(
self, parent_macro: Macro, to_link: Optional[Union[Scan, Step]] = None
) -> int:
total_frames = 1
for link in parent_macro.chain:
if link == to_link:
......@@ -347,16 +351,31 @@ class EigerScanner(DeviceScanner):
self.detector.stop()
@expose_method()
def get_scanner_macro(self, scanner_input, scanner_attributes, parent_macro=None):
def get_scanner_macro(
self,
scanner_input: StepDeviceMethod,
scanner_attributes: Dict[str, Any],
parent_macro: Macro,
):
if kamzik3.session is None:
raise DeviceError(
"kamzik session is None, can't get the device MacroServer"
)
macro_server = kamzik3.session.get_device("MacroServer")
current_scan_count = macro_server.get_macro_count()
if current_scan_count != self.last_scan_count:
self._init_new_scan(scanner_input, scanner_attributes, parent_macro)
self.last_scan_count = current_scan_count
scanner_input.on_macro_done = self.on_macro_done
scanner_input.on_macro_done = self.on_macro_done # type:ignore
# see https://github.com/python/mypy/issues/2427
return scanner_input
def _init_new_scan(self, scanner_input, scanner_attributes, parent_macro) -> None:
def _init_new_scan(
self,
scanner_input: StepDeviceMethod,
scanner_attributes: Dict[str, Any],
parent_macro: Macro,
) -> None:
if kamzik3.session is None:
raise DeviceError("kamzik session is None, can't get the scan number")
self.logger.info(
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment