Coverage for python/gsfit/database_readers/st40_astra_mdsplus/setup_rogowski_coils.py: 0%
38 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-07 13:12 +0000
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-07 13:12 +0000
1# mypy: ignore-errors
2# TODO: need to fix mypy errors
4import typing
5from typing import TYPE_CHECKING
7import mdsthin
8import numpy as np
9from gsfit_rs import RogowskiCoils
11from .astra_rogowski_coils_reader import astra_rogowski_coils_reader
13if TYPE_CHECKING:
14 from . import DatabaseReader
17def setup_rogowski_coils(
18 self: "DatabaseReader",
19 pulseNo: int,
20 settings: dict[str, typing.Any],
21) -> RogowskiCoils:
22 """
23 This method initialises the Rust `RogowskiCoils` class.
25 :param pulseNo: Pulse number, used to read from the database
26 :param settings: Dictionary containing the JSON settings read from the `settings` directory
28 **This method is specific to ST40's ASTRA stored on MDSplus.**
30 See `python/gsfit/database_readers/interface.py` for more details on how a new database_reader should be implemented.
31 """
33 # Initialise the RogowskiCoils Rust class
34 rogowski_coils = RogowskiCoils()
36 # Extract the astra_run_name from settings
37 astra_run_name = settings["GSFIT_code_settings.json"]["database_reader"]["st40_astra_mdsplus"]["workflow"]["astra"]["run_name"]
39 # Connect to MDSplus
40 conn = mdsthin.Connection("smaug")
41 conn.openTree("ASTRA", pulseNo)
43 # ASTRA rogowki_coils from MDSplus
44 time = conn.get(f"\\ASTRA::TOP.{astra_run_name}:TIME").data().astype(np.float64)
45 measurements = conn.get(f"\\ASTRA::TOP.{astra_run_name}.ROG.ALL:I").data().astype(np.float64)
46 names = conn.get(f"\\ASTRA::TOP.{astra_run_name}.ROG.ALL:NAME").data().astype(str)
48 # Read Rogowski coil geometry from rogpath.dat
49 rogowski_coils_data = astra_rogowski_coils_reader()
51 for rog_name, data in rogowski_coils_data.items():
52 path_r = data["r"]
53 path_z = data["z"]
55 # Don't store the "fake" Rogowski coils (e.g. the MC supports)
56 if len(path_r) > 4:
57 sensor_name = rog_name.replace("ROG_", "")
59 if sensor_name in settings["sensor_weights_rogowski_coils.json"]:
60 fit_settings_comment = settings["sensor_weights_rogowski_coils.json"][sensor_name]["fit_settings"]["comment"]
61 fit_settings_expected_value = settings["sensor_weights_rogowski_coils.json"][sensor_name]["fit_settings"]["expected_value"]
62 fit_settings_include = settings["sensor_weights_rogowski_coils.json"][sensor_name]["fit_settings"]["include"]
63 fit_settings_weight = settings["sensor_weights_rogowski_coils.json"][sensor_name]["fit_settings"]["weight"]
64 else:
65 fit_settings_comment = ""
66 fit_settings_expected_value = np.nan
67 fit_settings_include = False
68 fit_settings_weight = np.nan
70 # Measured values
71 index = names == rog_name
72 measured = measurements[:, index].squeeze()
74 # ASTRA does not include gaps in Rogowski coils
75 gaps_r = np.array([])
76 gaps_z = np.array([])
77 gaps_d_r = np.array([])
78 gaps_d_z = np.array([])
79 gaps_name = []
81 # Add Rogowski coil to the Rust class
82 rogowski_coils.add_sensor(
83 name=sensor_name,
84 r=path_r,
85 z=path_z,
86 fit_settings_comment=fit_settings_comment,
87 fit_settings_expected_value=fit_settings_expected_value,
88 fit_settings_include=fit_settings_include,
89 fit_settings_weight=fit_settings_weight,
90 time=time,
91 measured=measured,
92 gaps_r=gaps_r,
93 gaps_z=gaps_z,
94 gaps_d_r=gaps_d_r,
95 gaps_d_z=gaps_d_z,
96 gaps_name=gaps_name,
97 )
99 return rogowski_coils