Coverage for python/gsfit/database_readers/st40_astra_mdsplus/setup_bp_probes.py: 0%
30 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 BpProbes
11from .astra_bp_probe_reader import astra_bp_probe_reader
13if TYPE_CHECKING:
14 from . import DatabaseReader
17def setup_bp_probes(
18 self: "DatabaseReader",
19 pulseNo: int,
20 settings: dict[str, typing.Any],
21) -> BpProbes:
22 """
23 This method initialises the Rust `BpProbes` 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 BpProbes Rust class
34 bp_probes = BpProbes()
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 bp_probes
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}.BPPROBE.ALL:B").data().astype(np.float64)
46 names = conn.get(f"\\ASTRA::TOP.{astra_run_name}.BPPROBE.ALL:NAME").data().astype(str)
48 # Read Bp probe geometry from pf_probe.dat
49 bp_probes_data = astra_bp_probe_reader()
51 for bp_probe_name, data in bp_probes_data.items():
52 sensor_name = bp_probe_name.replace("BPPROBE_", "P")
54 if sensor_name in settings["sensor_weights_bp_probe.json"]:
55 fit_settings_comment = settings["sensor_weights_bp_probe.json"][sensor_name]["fit_settings"]["comment"]
56 fit_settings_expected_value = settings["sensor_weights_bp_probe.json"][sensor_name]["fit_settings"]["expected_value"]
57 fit_settings_include = settings["sensor_weights_bp_probe.json"][sensor_name]["fit_settings"]["include"]
58 fit_settings_weight = settings["sensor_weights_bp_probe.json"][sensor_name]["fit_settings"]["weight"]
59 else:
60 fit_settings_comment = ""
61 fit_settings_expected_value = np.nan
62 fit_settings_include = False
63 fit_settings_weight = np.nan
65 # Measured values
66 index = names == bp_probe_name
67 measured = measurements[:, index].squeeze()
69 # Add the sensor to the Rust class
70 bp_probes.add_sensor(
71 name=sensor_name,
72 geometry_angle_pol=data["angle_pol"],
73 geometry_r=data["r"],
74 geometry_z=data["z"],
75 fit_settings_comment=fit_settings_comment,
76 fit_settings_expected_value=fit_settings_expected_value,
77 fit_settings_include=fit_settings_include,
78 fit_settings_weight=fit_settings_weight,
79 time=time,
80 measured=measured,
81 )
83 return bp_probes