Migrate uv

pyproject.toml

@@ -12,7 +12,8 @@ requires-python = ">=3.10,<4"
 dependencies = [
     "sdsstools>=1.6.1",
     "sdss-clu>=2.2.7",
-    "astropy>=6.1"
+    "astropy>=6.1",
+    "sdss-drift>=1.2.0",
 ]
 
 [project.optional-dependencies]

src/lvmopstools/__init__.py

@@ -8,11 +8,20 @@
 
 from __future__ import annotations
 
-from sdsstools.metadata import get_package_version
+import os
+import pathlib
+
+from sdsstools.metadata import get_package_version, read_yaml_file
 
 
 __version__ = get_package_version(path=__file__, package_name="lvmopstools")
 
 
+DEFAULT_CONFIG_FILE = pathlib.Path(__file__).parent / "config.yaml"
+CONFIG_FILE = os.environ.get("LVMOPSTOOLS_CONFIG_FILE", str(DEFAULT_CONFIG_FILE))
+
+config = read_yaml_file(CONFIG_FILE)
+
+
 from .retrier import *
 from .socket import *

src/lvmopstools/config.yaml

@@ -0,0 +1,53 @@
+devices:
+  thermistors:
+    host: 10.8.38.180
+    port: 1025
+    mapping:
+      channel7: supply
+      channel8: r1
+      channel9: b1
+      channel6: z1
+      channel10: r3
+      channel11: b3
+      channel12: z3
+      channel13: r2
+      channel14: b2
+      channel15: z2
+
+  ion:
+    # - host: 10.8.38.155
+    #   port: 502
+    #   cameras:
+    #     z1:
+    #       signal_address: 0
+    #       onoff_address: 2020
+    #     b1:
+    #       signal_address: 2
+    #       onoff_address: 2021
+    #     r1:
+    #       signal_address: 4
+    #       onoff_address: 2022
+    - host: 10.8.38.156
+      port: 502
+      cameras:
+        z2:
+          signal_address: 0
+          onoff_address: 2020
+        b2:
+          signal_address: 2
+          onoff_address: 2021
+        r2:
+          signal_address: 4
+          onoff_address: 2022
+    # - host: 10.8.38.157
+    #   port: 502
+    #   cameras:
+    #     z3:
+    #       signal_address: 0
+    #       onoff_address: 2020
+    #     b3:
+    #       signal_address: 2
+    #       onoff_address: 2021
+    #     r3:
+    #       signal_address: 4
+    #       onoff_address: 2022

src/lvmopstools/devices/ion.py

@@ -0,0 +1,105 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# @Author: José Sánchez-Gallego ([email protected])
+# @Date: 2024-09-11
+# @Filename: ion.py
+# @License: BSD 3-clause (http://www.opensource.org/licenses/BSD-3-Clause)
+
+from __future__ import annotations
+
+from drift import Drift
+from drift.convert import data_to_float32
+from sdsstools import GatheringTaskGroup
+
+from lvmopstools import config
+
+
+def convert_pressure(volts: float):
+    """Converts differential voltage to pressure in Torr."""
+
+    # The calibration is a linear fit of the form y = mx + b
+    m = 2.04545
+    b = -6.86373
+
+    log10_pp0 = m * volts + b  # log10(PPa), pressure in Pascal
+
+    torr = 10**log10_pp0 * 0.00750062
+
+    return torr
+
+
+async def _read_one_ion_controller(ion_config: dict):
+    """Reads the signal and on/off status from an ion controller."""
+
+    results: dict[str, dict] = {}
+
+    drift = Drift(ion_config["host"], ion_config.get("port", 502))
+
+    async with drift:
+        for camera, camera_config in ion_config["cameras"].items():
+            signal_address = camera_config["signal_address"]
+            # onoff_address = camera_config["onoff_address"]
+
+            signal = await drift.client.read_input_registers(signal_address, 2)
+            # onoff = await drift.client.read_input_registers(onoff_address, 1)
+
+            diff_volt = data_to_float32(tuple(signal.registers))
+            pressure = convert_pressure(diff_volt)
+
+            # onoff_status = bool(onoff.registers[0])
+            onoff_status = pressure > 1e-8
+
+            results[camera] = {"pressure": pressure, "on": onoff_status}
+
+    return results
+
+
+async def read_ion_pumps():
+    """Reads the signal and on/off status from an ion pump."""
+
+    ion_config: list[dict] = config["devices.ion"]
+
+    async with GatheringTaskGroup() as group:
+        for ion_controller in ion_config:
+            group.create_task(_read_one_ion_controller(ion_controller))
+
+    results = {camera: item[camera] for item in group.results() for camera in item}
+
+    return results
+
+
+async def toggle_ion_pump(camera: str, on: bool):
+    """Turns the ion pump on or off.
+
+    Parameters
+    ----------
+    camera
+        The camera for which to toggle the ion pump.
+    on
+        If `True`, turns the pump on. If `False`, turns the pump off. If `None`,
+        toggles the pump current status.
+
+    """
+
+    ion_config: list[dict] = config["devices.ion"]
+
+    host: str | None = None
+    port: int | None = None
+    on_off_address: int | None = None
+
+    for ic in ion_config:
+        if camera in ic["cameras"]:
+            host = ic["host"]
+            port = ic.get("port", 502)
+            on_off_address = ic["cameras"][camera]["onoff_address"]
+            break
+
+    if host is None or port is None or on_off_address is None:
+        raise ValueError(f"Camera {camera!r} not found in the configuration.")
+
+    drift = Drift(host, port)
+
+    async with drift:
+        value = 2**16 - 1 if on else 0
+        await drift.client.write_register(on_off_address, value)