Merge remote-tracking branch 'origin/dev/main' into origin/dev/feature/custom_tango

Author: SafiullahOmar

src/dt4acc/core/accelerators/accelerator_manager.py

@@ -46,7 +46,7 @@ def initialize(self):
         """
         try:
             from lat2db.model.accelerator import Accelerator
-            acc_model = Accelerator()
+            acc_model = Accelerator(file_name ="bessyii_lattice_json.json", from_json= True)
 
             # Initialize the accelerator with required components
             self.accelerator = AcceleratorImpl(

src/dt4acc/core/accelerators/pyat_accelerator.py

@@ -1,4 +1,5 @@
 import os
+import getpass
 
 from .accelerator_manager import AcceleratorManager
 
@@ -13,7 +14,7 @@ def setup_accelerator():
     Returns:
         AcceleratorManager: Singleton instance of the accelerator manager.
     """
-    prefix = os.getenv("DT4ACC_PREFIX", "Anonym")
+    prefix = os.getenv("DT4ACC_PREFIX", getpass.getuser())
     manager = AcceleratorManager(prefix=prefix)
     manager.initialize()
     return manager

src/dt4acc/core/calculations/pyat_calculator.py

@@ -1,3 +1,4 @@
+import getpass
 import os
 from abc import ABCMeta
 from concurrent.futures import ThreadPoolExecutor
@@ -62,7 +63,7 @@ def calculate(self) -> TwissWithAggregatedKValues:
             RuntimeError: If the calculation fails due to invalid input.
         """
 
-        logger.warning("Starting Twiss calculation (get_optics)")
+        logger.info("Starting Twiss calculation (get_optics)")
 
         twiss_in = {
             'beta': np.array([8.860461, 4.03432]),
@@ -95,7 +96,7 @@ def _extract_pv_values(self) -> Sequence[MainValue]:
             tuple: PV names and their corresponding K values.
         """
         main_values = []
-        prefix = os.environ.get('DT4ACC_PREFIX', 'Anonym')
+        prefix = os.environ.get('DT4ACC_PREFIX', getpass.getuser())
         for element in self.acc:
             if element.__class__.__name__ == "Quadrupole":
                 main_values.append(
@@ -143,7 +144,7 @@ def calculate(self) -> Orbit:
         Raises:
             RuntimeError: If the calculation fails.
         """
-        logger.warning("Starting orbit calculation (find_orbit)")
+        logger.info("Starting orbit calculation (find_orbit)")
 
         try:
             x0, orbit = self.acc.find_orbit(at.All)

src/dt4acc/core/views/shared_view.py

@@ -1,3 +1,4 @@
+import getpass
 import os
 
 from ..utils.logger import get_logger
@@ -16,7 +17,7 @@ def get_view_instance():
     if _view_instance is not None:
         return _view_instance
 
-    prefix = os.environ.get("DT4ACC_PREFIX", "Anonym")
+    prefix = os.environ.get("DT4ACC_PREFIX", getpass.getuser())
     server_type = os.environ.get("server", "epics").lower()
 
     if server_type == "tango":

src/dt4acc/custom_epics/data/querries.py

@@ -1,28 +1,82 @@
-import os
+# import os
+#
+# import pymongo
 
-import pymongo
+# from dt4acc import mongodb_
+#
+# client = pymongo.MongoClient(mongodb_)
+# DB_NAME = os.environ.get("MONGODB_DB", "bessyii")
+# db = client[DB_NAME]
+# collection = db['accelerator.setup']
+# TODO
+# this is a temporary solution to avoid the MongoDB dependency
+# rework this to use the file repository along with mongodb
 
-from dt4acc import mongodb_
+import json
+from pathlib import Path
+from typing import Iterable, List, Dict, Any
 
-client = pymongo.MongoClient(mongodb_)
-DB_NAME = os.environ.get("MONGODB_DB", "bessyii")
-db = client[DB_NAME]
-collection = db['accelerator.setup']
+# -----------------------------------------------------------------
+# locate and load the data file once; keep it cached in _DATA
+# -----------------------------------------------------------------
+_DATA_FILE = (
+        Path(__file__)
+        .resolve()  # .../src/dt4acc/custom_epics/queries_json.py
+        .parent  # .../src/dt4acc/custom_epics/data
+        / "standard"
+        / "accelerator_setup.json"
+)
 
+with _DATA_FILE.open() as fp:
+    _DATA: List[Dict[str, Any]] = json.load(fp)
 
+
+# -----------------------------------------------------------------
+# helper: Mongo-style $in filtering for the tiny use-cases we need
+# -----------------------------------------------------------------
+def _match(doc: Dict[str, Any], field: str, allowed: Iterable[str]) -> bool:
+    return doc.get(field) in allowed
+
+
+# -----------------------------------------------------------------
+# public API – identical signatures to the Mongo version
+# -----------------------------------------------------------------
 def get_magnets():
-    return collection.find({"type": {"$in": ["Quadrupole", "Sextupole", "Steerer"]}})
+    """Return all Quadrupole/Sextupole/Steerer magnets as an iterator."""
+    wanted = {"Quadrupole", "Sextupole", "Steerer"}
+    return (d for d in _DATA if _match(d, "type", wanted))
+
 
+def get_magnets_per_power_converters(pc: str) -> List[Dict[str, Any]]:
+    """Return all magnets driven by the given power-converter name."""
+    return [d for d in _DATA if d.get("pc") == pc]
 
-def get_magnets_per_power_converters(pc):
-    return list(collection.find({"pc": pc}))
 
+def get_unique_power_converters() -> List[str]:
+    """Distinct list of power-converter names for Quad/Sext/Steerer magnets."""
+    wanted = {"Quadrupole", "Sextupole", "Steerer"}
+    return sorted({d["pc"] for d in _DATA if _match(d, "type", wanted)})
 
-def get_unique_power_converters():
-    """Fetch unique power converter names from magnets in the DB."""
-    return collection.distinct("pc", {"type": {"$in": ["Quadrupole", "Sextupole", "Steerer"]}})
 
+def get_unique_power_converters_type_specified(type_list: Iterable[str]) -> List[str]:
+    """Distinct list of power-converter names for the supplied magnet types."""
+    wanted = set(type_list)
+    return sorted({d["pc"] for d in _DATA if _match(d, "type", wanted)})
 
-def get_unique_power_converters_type_specified(type_list):
-    """Fetch unique power converter names from magnets in the DB."""
-    return collection.distinct("pc", {"type": {"$in": type_list}})
+#
+# def get_magnets():
+#     return collection.find({"type": {"$in": ["Quadrupole", "Sextupole", "Steerer"]}})
+#
+#
+# def get_magnets_per_power_converters(pc):
+#     return list(collection.find({"pc": pc}))
+#
+#
+# def get_unique_power_converters():
+#     """Fetch unique power converter names from magnets in the DB."""
+#     return collection.distinct("pc", {"type": {"$in": ["Quadrupole", "Sextupole", "Steerer"]}})
+#
+#
+# def get_unique_power_converters_type_specified(type_list):
+#     """Fetch unique power converter names from magnets in the DB."""
+#     return collection.distinct("pc", {"type": {"$in": type_list}})