Reorganize TEMController

src/instamatic/TEMController/__init__.py

@@ -1,4 +1,15 @@
+# ruff: noqa: E402
 from __future__ import annotations
 
+import warnings
+
+from instamatic.utils.deprecated import VisibleDeprecationWarning
+
+warnings.warn(
+    'The `TEMController` module is deprecated since version 2.0.6. Use the `controller`-module instead',
+    VisibleDeprecationWarning,
+    stacklevel=2,
+)
+
 from .microscope import Microscope
 from .TEMController import get_instance, initialize

src/instamatic/TEMController/microscope.py

@@ -1,62 +1,20 @@
 from __future__ import annotations
 
-from instamatic import config
-from instamatic.TEMController.microscope_base import MicroscopeBase
-
-default_tem_interface = config.microscope.interface
+from instamatic.microscope.base import MicroscopeBase
+from instamatic.utils.deprecated import deprecated
 
 __all__ = ['Microscope', 'get_tem']
 
 
+@deprecated(since='2.0.6', alternative='instamatic.microscope.get_microscope_class')
 def get_tem(interface: str) -> 'type[MicroscopeBase]':
-    """Grab tem class with the specific 'interface'."""
-    simulate = config.settings.simulate
-
-    if config.settings.tem_require_admin:
-        from instamatic import admin
-
-        if not admin.is_admin():
-            raise PermissionError('Access to the TEM interface requires admin rights.')
+    from instamatic.microscope import get_microscope_class
 
-    if simulate or interface == 'simulate':
-        from .simu_microscope import SimuMicroscope as cls
-    elif interface == 'jeol':
-        from .jeol_microscope import JeolMicroscope as cls
-    elif interface == 'fei':
-        from .fei_microscope import FEIMicroscope as cls
-    elif interface == 'fei_simu':
-        from .fei_simu_microscope import FEISimuMicroscope as cls
-    else:
-        raise ValueError(f'No such microscope interface: `{interface}`')
-
-    return cls
+    return get_microscope_class(interface=interface)
 
 
+@deprecated(since='2.0.6', alternative='instamatic.microscope.get_microscope')
 def Microscope(name: str = None, use_server: bool = False) -> MicroscopeBase:
-    """Generic class to load microscope interface class.
-
-    name: str
-        Specify which microscope to use, must be one of `jeol`, `fei_simu`, `simulate`
-    use_server: bool
-        Connect to microscope server running on the host/port defined in the config file
-
-    returns: TEM interface class
-    """
-    if name is None:
-        interface = default_tem_interface
-        name = interface
-    elif name != config.settings.microscope:
-        config.load_microscope_config(microscope_name=name)
-        interface = config.microscope.interface
-    else:
-        interface = config.microscope.interface
-
-    if use_server:
-        from .microscope_client import MicroscopeClient
-
-        tem = MicroscopeClient(interface=interface)
-    else:
-        cls = get_tem(interface=interface)
-        tem = cls(name=name)
-
-    return tem
+    from instamatic.microscope import get_microscope
+
+    return get_microscope(name=name, use_server=use_server)

src/instamatic/TEMController/microscope_client.py

@@ -1,183 +1,3 @@
 from __future__ import annotations
 
-import atexit
-import datetime
-import json
-import pickle
-import socket
-import subprocess as sp
-import threading
-import time
-from functools import wraps
-
-from instamatic import config
-from instamatic.exceptions import TEMCommunicationError, exception_list
-from instamatic.server.serializer import dumper, loader
-
-HOST = config.settings.tem_server_host
-PORT = config.settings.tem_server_port
-BUFSIZE = 1024
-
-
-class ServerError(Exception):
-    pass
-
-
-def kill_server(p):
-    # p.kill is not adequate
-    sp.call(['taskkill', '/F', '/T', '/PID', str(p.pid)])
-
-
-def start_server_in_subprocess():
-    cmd = 'instamatic.temserver.exe'
-    p = sp.Popen(cmd, stdout=sp.DEVNULL)
-    print(f'Starting TEM server ({HOST}:{PORT} on pid={p.pid})')
-    atexit.register(kill_server, p)
-
-
-class MicroscopeClient:
-    """Simulates a Microscope object and synchronizes calls over a socket
-    server.
-
-    For documentation, see the actual python interface to the microscope
-    API.
-    """
-
-    def __init__(self, *, interface: str):
-        super().__init__()
-
-        self.interface = interface
-        self.name = interface
-        self._bufsize = BUFSIZE
-
-        try:
-            self.connect()
-        except ConnectionRefusedError:
-            start_server_in_subprocess()
-
-            for t in range(30):
-                try:
-                    self.connect()
-                except ConnectionRefusedError:
-                    time.sleep(1)
-                    if t > 3:
-                        print('Waiting for server')
-                    if t > 30:
-                        raise TEMCommunicationError(
-                            'Cannot establish server connection (timeout)'
-                        )
-                else:
-                    break
-
-        self._init_dict()
-        self.check_goniotool()
-
-        atexit.register(self.s.close)
-
-    def connect(self):
-        self.s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        self.s.connect((HOST, PORT))
-        print(f'Connected to TEM server ({HOST}:{PORT})')
-
-    def __getattr__(self, func_name):
-        try:
-            wrapped = self._dct[func_name]
-        except KeyError as e:
-            raise AttributeError(
-                f'`{self.__class__.__name__}` object has no attribute `{func_name}`'
-            ) from e
-
-        @wraps(wrapped)
-        def wrapper(*args, **kwargs):
-            dct = {'func_name': func_name, 'args': args, 'kwargs': kwargs}
-            return self._eval_dct(dct)
-
-        return wrapper
-
-    def _eval_dct(self, dct):
-        """Takes approximately 0.2-0.3 ms per call if HOST=='localhost'."""
-        self.s.send(dumper(dct))
-
-        response = self.s.recv(self._bufsize)
-
-        if response:
-            status, data = loader(response)
-
-        if status == 200:
-            return data
-
-        elif status == 500:
-            error_code, args = data
-            raise exception_list.get(error_code, TEMCommunicationError)(*args)
-
-        else:
-            raise ConnectionError(f'Unknown status code: {status}')
-
-    def _init_dict(self):
-        from instamatic.TEMController.microscope import get_tem
-
-        tem = get_tem(interface=self.interface)
-
-        self._dct = {
-            key: value for key, value in tem.__dict__.items() if not key.startswith('_')
-        }
-
-    def __dir__(self):
-        return self._dct.keys()
-
-    def check_goniotool(self):
-        """Check whether goniotool is available and update the config as
-        necessary."""
-        if config.settings.use_goniotool:
-            config.settings.use_goniotool = self.is_goniotool_available()
-
-
-class TraceVariable:
-    """Simple class to trace a variable over time.
-
-    Usage:
-        t = TraceVariable(ctrl.stage.get, verbose=True)
-        t.start()
-        t.stage.set(x=0, y=0, wait=False)
-        ...
-        values = t.stop()
-    """
-
-    def __init__(
-        self,
-        func,
-        interval: float = 1.0,
-        name: str = 'variable',
-        verbose: bool = False,
-    ):
-        super().__init__()
-        self.name = name
-        self.func = func
-        self.interval = interval
-        self.verbose = verbose
-
-        self._traced = []
-
-    def start(self):
-        print(f'Trace started: {self.name}')
-        self.update()
-
-    def stop(self):
-        self._timer.cancel()
-
-        print(f'Trace canceled: {self.name}')
-
-        return self._traced
-
-    def update(self):
-        ret = self.func()
-
-        now = datetime.datetime.now().strftime('%H:%M:%S.%f')
-
-        if self.verbose:
-            print(f'{now} | Trace {self.name}: {ret}')
-
-        self._traced.append((now, ret))
-
-        self._timer = threading.Timer(self.interval, self.update)
-        self._timer.start()
+from instamatic.microscope.client import MicroscopeClient