Merge pull request #127 from wright-group/io_Topas4

Io topas4

CHANGELOG.md

@@ -5,6 +5,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/).
 
 ## [Unreleased]
 
+### Added
+- Reading Topas4 formatted curves (from a directory of JSON files)
+- Setables now have the ability to have defaults, which are added to Notes if not otherwise specified.
+
 ### Fixed
 - Tune Test does not overwrite discrete tunes and runs instead of failing if they exist
 
@@ -16,6 +20,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/).
 
 ## [0.4.1]
 
+### Added
+- from_topas4  and associated test script, for import of Topas4 file format 
+
 ### Changed
 - setables are now fully optional
 

attune/__init__.py

@@ -16,3 +16,4 @@
 from ._store import *
 from ._tune import *
 from ._tune_test import *
+from .io import *

attune/_instrument.py

@@ -113,6 +113,9 @@ def __call__(self, ind_value, arrangement_name=None) -> Note:
                     continue
                 setable_positions[tune_name] = tune(v)
                 setables[tune_name] = Setable(tune_name)
+        for setable in self._setables:
+            if setable not in setable_positions and self._setables[setable].default is not None:
+                setable_positions[setable] = self._setables[setable].default
         # finish
         note = Note(
             setables=self._setables,

attune/_note.py

@@ -31,7 +31,7 @@ def __getitem__(self, k):
         return self.setable_positions[k]
 
     def __repr__(self):
-        return f"Note({self.setables}, {self.setable_positions}, {self.arrangement_name})"
+        return f"Note({self.setables}, {self.setable_positions}, {repr(self.arrangement_name)})"
 
     def items(self):
         """Items in the Note."""

attune/_setable.py

@@ -1,8 +1,10 @@
 __all__ = ["Setable"]
 
+from typing import Union, Optional
+
 
 class Setable(object):
-    def __init__(self, name: str, **kwargs):
+    def __init__(self, name: str, default: Optional[Union[str, float]] = None, **kwargs):
         """Setable object representation.
 
         Parameters
@@ -11,15 +13,17 @@ def __init__(self, name: str, **kwargs):
             The key for this setable
         """
         self.name = name
+        self.default = default
 
     def __repr__(self):
-        return f"Setable({repr(self.name)})"
+        return f"Setable({repr(self.name)}, {repr(self.default)})"
 
     def __eq__(self, other):
-        return self.name == other.name
+        return self.name == other.name and self.default == other.default
 
     def as_dict(self):
         """Representation as a JSON encodable dictionary."""
         out = {}
         out["name"] = self.name
+        out["default"] = self.default
         return out

attune/io/__init__.py

@@ -0,0 +1 @@
+from .topas4 import *

attune/io/topas4.py

@@ -0,0 +1,128 @@
+__all__ = ["from_topas4"]
+
+import json
+import os
+import re
+
+from .. import Arrangement
+from .. import DiscreteTune
+from .. import Instrument
+from .. import Setable
+from .. import Tune
+
+
+def from_topas4(topas4_folder):
+    """Convert a LightConversion marked up Topas4 calibration JSON file into an Instrument object.
+    Topas4 Folder must contain at least the following 2 files:  OpticalDevices.json and Motors.json."""
+    with open(os.path.join(topas4_folder, "OpticalDevices.json"), "r") as f:
+        opt_dev = json.load(f)
+
+    with open(os.path.join(topas4_folder, "Motors.json"), "r") as g:
+        motors = json.load(g)["Motors"]
+
+    with open(os.path.join(topas4_folder, "SeparationDevices.json"), "r") as g:
+        sep_dev = json.load(g)
+
+    opt_dev_active_guid = opt_dev.get("ActiveConfigurationGUID")
+    opt_dev_conf = opt_dev["Configurations"]
+
+    ind = 0
+    for a in opt_dev_conf:
+        if a["GUID"] == opt_dev_active_guid:
+            opt_dev_active = a["OpticalDevices"]
+            break
+
+    sep_dev_active_guid = sep_dev.get("ActiveConfigurationGUID")
+    sep_dev_conf = sep_dev["Configurations"]
+
+    ind = 0
+    for a in sep_dev_conf:
+        if a["GUID"] == sep_dev_active_guid:
+            sep_dev_active = a["SeparationDevices"]
+            break
+
+    arrangements = {}
+    setables = {}
+
+    for interaction in opt_dev_active:
+
+        motorlist = {}
+        for motor in motors:
+            index = motor["Index"]
+            motorlist[index] = motor["Title"]
+
+        neutral_positions = interaction["NeutralPositions"]
+        for pos in neutral_positions:
+            index = pos["MotorIndex"]
+            mot_name = motorlist[index]
+            if pos["UseNamedPositionToResolveValue"]:
+                position = pos["NamedPosition"]
+            else:
+                position = pos["Position"]
+            setables[mot_name] = Setable(mot_name, position)
+
+        interaction = interaction["Interactions"]
+
+        for independent in interaction:
+            arrange_name_full = independent.get("Type")
+
+            if ">" in arrange_name_full:
+                arrange_name, parent = arrange_name_full.split(">", maxsplit=1)
+            else:
+                arrange_name = arrange_name_full
+                parent = arrange_name_full.split("-", maxsplit=1)[-1]
+                if arrange_name == parent:
+                    parent = None
+
+            tunes = {}
+
+            if parent is not None:
+                inputpoints = independent.get("InputPoints")
+                deparr = []
+                indarr = []
+                for inputpoint in inputpoints:
+                    indarr.append(inputpoint.get("Output"))
+                    deparr.append(inputpoint.get("Input"))
+
+                tune = Tune(indarr, deparr)
+                tunes[parent] = tune
+
+            ind_motors = independent.get("MotorPositionCurves")
+            for points in ind_motors:
+                motorindex = points["MotorIndex"]
+                k = motorlist[motorindex]
+                deparr = []
+                indarr = []
+                for point in points["Points"]:
+                    indarr.append(point.get("Input"))
+                    deparr.append(point.get("Output"))
+
+                tune = Tune(indarr, deparr)
+                tunes[k] = tune
+
+            arrangements[arrange_name] = Arrangement(arrange_name, tunes).as_dict()
+
+    discrete_tunes = {x: {} for x in arrangements}
+    for interaction in sep_dev_active:
+        for mot_position in interaction["MotorPositionCurves"]:
+            mot_index = mot_position["Index"]
+            mot_name = motorlist[mot_index]
+            for point in mot_position["Points"]:
+                interaction = point["InteractionType"]
+                interaction = interaction.replace("[", "")
+                interaction = interaction.replace("]", "")
+                interaction = interaction.replace("*", ".*")
+                for arr, mots in discrete_tunes.items():
+                    if re.match(interaction, arr):
+                        x = mots.get(mot_name, {})
+                        x[point["NamedPosition"]] = (point["Input"]["From"], point["Input"]["To"])
+                        mots[mot_name] = x
+
+    for arr, mots in discrete_tunes.items():
+        for mot_name, pos in mots.items():
+            tune = DiscreteTune(pos)
+            arrangements[arr]["tunes"][mot_name] = tune
+
+    instr = Instrument(arrangements, setables)
+
+    return instr

tests/io/test_Topas4.py

@@ -0,0 +1,73 @@
+import math
+import attune
+import pytest
+import json
+from pathlib import Path
+from attune import Tune, DiscreteTune
+import numpy as np
+
+test_dir = Path(__file__).parent / "twin_test_data"
+
+
+def test_from_topas4():
+    instr = attune.from_topas4(test_dir)
+
+    TOTAL_ARRANGEMENTS = 3
+    TOTAL_MINS_MAXES = 7
+    MIN_DEPS_REF = [
+        2.230625,
+        -2.2662695266462407,
+        2.44375,
+        0.29442646838236897,
+        2500.0,
+        1640.7,
+        145.44583333333333,
+    ]
+    MIN_INDEPS_REF = [1300.0, 1300.0, 1300.0, 1300.0, 1757.493188010899, 4000.0, 4000.0]
+    MAX_DEPS_REF = [
+        4.9759375,
+        8.030085499389287,
+        2.65625,
+        -26.225919814858464,
+        1300.0,
+        1952.081,
+        124.17083333333333,
+    ]
+    MAX_INDEPS_REF = [2500.0, 2500.0, 2500.0, 2500.0, 5005.9701492537315, 18000.0, 18000.0]
+
+    arrangements = {"SIG", "IDL", "DFG-SIG"}
+
+    assert isinstance(instr, attune.Instrument)
+
+    min_deps = []
+    min_indeps = []
+    max_deps = []
+    max_indeps = []
+
+    assert set(instr.arrangements.keys()) == arrangements
+
+    for key in instr.arrangements:
+        for tune in instr.arrangements[key].tunes.values():
+            if isinstance(tune, Tune):
+                min_deps.append(tune.dependent[0])
+                min_indeps.append(tune.independent[0])
+                max_deps.append(tune.dependent[-1])
+                max_indeps.append(tune.independent[-1])
+
+    assert len(min_deps) == TOTAL_MINS_MAXES
+    assert len(min_indeps) == TOTAL_MINS_MAXES
+    assert len(max_deps) == TOTAL_MINS_MAXES
+    assert len(max_indeps) == TOTAL_MINS_MAXES
+
+    for i in range(len(min_deps)):
+        assert np.isclose(min_deps[i], MIN_DEPS_REF[i])
+        assert np.isclose(min_indeps[i], MIN_INDEPS_REF[i])
+        assert np.isclose(max_deps[i], MAX_DEPS_REF[i])
+        assert np.isclose(max_indeps[i], MAX_INDEPS_REF[i])
+
+    assert isinstance(instr["DFG-SIG"]["RP Stage"], DiscreteTune)
+    assert instr(10000)["RP Stage"] == "IN"
+
+
+if __name__ == "__main__":
+    test_from_topas4()

tests/io/twin_test_data/AccessControl.json

@@ -0,0 +1,6 @@
+{
+  "AllowedIPs": [
+    "127.0.0.1"
+  ],
+  "OverrideAllowAll": false
+}

tests/io/twin_test_data/DefaultUserPreferences.json

@@ -0,0 +1,4 @@
+{
+  "Motors": [],
+  "MotorSliderStepSizeFallback": 1.0
+}

tests/io/twin_test_data/General.json

@@ -0,0 +1,7 @@
+{
+  "SerialNumber": "PA0465",
+  "Nickname": "",
+  "RestServicesPort": 8008,
+  "IsDemo": false,
+  "DeviceModel": "Topas-Twins"
+}

tests/io/twin_test_data/LaserControl.json

@@ -0,0 +1,4 @@
+{
+  "Type": "CarbideTitanV1_Comp",
+  "Address": "192.168.11.251"
+}

tests/io/twin_test_data/Loggers.json

@@ -0,0 +1,16 @@
+{
+  "RootDirectory": "",
+  "Log1DEveryMs": 5000,
+  "IsEnabled": false,
+  "ADCAndTemperature": {
+    "Log1DEveryMs": 0,
+    "IsEnabled": null
+  },
+  "MotorPositions": {
+    "LogSteps": true,
+    "LogUnits": true,
+    "LogSuperUnits": true,
+    "Log1DEveryMs": 0,
+    "IsEnabled": null
+  }
+}