Coverage for python/gsfit/database_readers/st40_mdsplus/setup_isoflux_sensors.py: 0%

1import typing 

2from typing import TYPE_CHECKING 

3 

4import numpy as np 

5import numpy.typing as npt 

6from gsfit_rs import Isoflux 

7from shapely.geometry import LineString 

8from st40_database import GetData 

9 

10if TYPE_CHECKING: 

11 from . import DatabaseReader 

12 

13 

14def setup_isoflux_sensors( 

15 self: "DatabaseReader", 

16 pulseNo: int, 

17 settings: dict[str, typing.Any], 

18 times_to_reconstruct: npt.NDArray[np.float64], 

19) -> Isoflux: 

20 """ 

21 This method initialises the Rust `Isoflux` class. 

22 

23 :param pulseNo: Pulse number, used to read from the database 

24 :param settings: Dictionary containing the JSON settings read from the `settings` directory 

25 

26 **This method is specific to ST40's experimental MDSplus database.** 

27 

28 See `python/gsfit/database_readers/interface.py` for more details on how a new database_reader should be implemented. 

29 """ 

30 

31 # Initialise the Isoflux Rust class 

32 isoflux = Isoflux() 

33 

34 isoflux_settings = settings["sensor_weights_isoflux.json"] 

35 

36 if len(isoflux_settings) == 0: 

37 return isoflux 

38 

39 code_name = settings["GSFIT_code_settings.json"]["database_reader"]["st40_mdsplus"]["isoflux_code_name"] 

40 run_name = settings["GSFIT_code_settings.json"]["database_reader"]["st40_mdsplus"]["workflow"][code_name]["run_name"] 

41 

42 # TODO: need to check that this is working! 

43 # for isoflux_name in isoflux_settings.keys(): 

44 # # We need a try/except here because there is sometimes no TS data 

45 # try: 

46 # results = GetData(pulseNo, f"{code_name}#{run_name}") 

47 

48 # time = typing.cast(npt.NDArray[np.float32], results.get("TIME")).astype(np.float64) 

49 # n_time = len(time) 

50 

51 # r = typing.cast(npt.NDArray[np.float32], results.get("R_MID_PROFILES.R")).astype(np.float64) 

52 # te = typing.cast(npt.NDArray[np.float32], results.get("R_MID_PROFILES.TE")).astype(np.float64) 

53 

54 # location_1_r = np.full(n_time, np.nan) 

55 # location_1_z = np.full(n_time, np.nan) 

56 # location_2_r = np.full(n_time, np.nan) 

57 # location_2_z = np.full(n_time, np.nan) 

58 

59 # for i_time in range(0, n_time): 

60 # # Create LineString objects for the two lines 

61 # te_profile_line = LineString(np.column_stack((r[i_time, :], te[i_time, :]))) 

62 

63 # # Read in a LFS coordinate we want to use as the constant 

64 # location_1_r_now = isoflux_settings[isoflux_name]["location_1"]["r"] 

65 # location_1_z_now = 0.0 # Assumed for TS 

66 

67 # # Interpolate the Te profile to find Te at the LFS 

68 # index_sorted = np.argsort(r[i_time, :]) 

69 # te_interpolated = np.interp(location_1_r_now, r[i_time, index_sorted], te[i_time, index_sorted]) 

70 

71 # r_target_line = np.array([location_1_r_now, 200.0]) 

72 # te_target_line = np.array([te_interpolated, te_interpolated]) 

73 # line_to_interpolate = LineString(np.column_stack((r_target_line, te_target_line))) 

74 

75 # # Find the intersection 

76 # intersections = te_profile_line.intersection(line_to_interpolate) 

77 

78 # # Convert `intersections` to list 

79 # geoms = [] 

80 # if intersections.geom_type == "MultiPoint": 

81 # geoms = [geom for geom in intersections.geoms] # TODO: causes mypy error?? 

82 # # if intersections.geom_type == 'Point' # there is only one intersection, so we haven't found both the LFS and HFS 

83 # # if intersections.geom_type == 'MultiLineString': # this only happens when the entire line is 0.0 

84 

85 # # Loop over all intersections 

86 # location_2_r_now = np.nan 

87 # location_2_z_now = np.nan 

88 # if len(geoms) > 0: 

89 # for geom in geoms: 

90 # # Look for HFS intersection 

91 # if geom.x > location_1_r_now: 

92 # location_2_r_now = geom.x 

93 # location_2_z_now = 0.0 

94 

95 # # Store the isoflux coordinates 

96 # if not np.isnan(location_2_r_now): 

97 # location_1_r[i_time] = location_1_r_now 

98 # location_1_z[i_time] = location_1_z_now 

99 # location_2_r[i_time] = location_2_r_now 

100 # location_2_z[i_time] = location_2_z_now 

101 

102 # isoflux.add_sensor( 

103 # name=isoflux_name, 

104 # fit_settings_comment="", 

105 # fit_settings_include=True, 

106 # fit_settings_weight=1.0e3, 

107 # time=time, 

108 # location_1_r=location_1_r, 

109 # location_1_z=location_1_z, 

110 # location_2_r=location_2_r, 

111 # location_2_z=location_2_z, 

112 # times_to_reconstruct=times_to_reconstruct, 

113 # ) 

114 # except Exception as e: 

115 # print(f"Error reading {isoflux_name}; exception={e}") 

116 

117 return isoflux