Coverage for python/gsfit/database_writers/tokamak_energy_mdsplus_new/create_mdsplus_links.py: 0%

1import mdsthin 

2import numpy as np 

3 

4 

5def create_mdsplus_links(pulseNo_write: int, run_name: str, tree_name: str = "gsfit") -> None: 

6 """ 

7 Add the "ALL" nodes for the "CONSTRAINTS" 

8 

9 :param pulseNo_write: Pulse number to write to 

10 :param run_name: Run name, e.g. "TEST07" 

11 :param tree_name: MDSplus tree to open, e.g. "gsfit" or "GSFIT_TS" [dimensionless] 

12 """ 

13 

14 conn = mdsthin.Connection("smaug") 

15 conn.openTree(tree_name, pulseNo_write) 

16 

17 tree_name_upper = tree_name.upper() 

18 

19 # Add signals 

20 sensor_types = ["BP_PROBE", "FLUX_LOOP", "ROGOWSKI", "PF_CURRENT", "PRESSURE"] 

21 signal_names = ["MEASURED", "RECONSTRUCT"] 

22 for sensor_type in sensor_types: 

23 tdi_command = f"getnci('\\\\{tree_name_upper}::TOP.{run_name}.CONSTRAINTS.{sensor_type}.*', 'node_name')" 

24 sensor_names_on_mds_np = conn.get(tdi_command).data() 

25 sensor_names_on_mds = np.char.strip(sensor_names_on_mds_np).astype(str, copy=False).tolist() 

26 sensor_names_on_mds.remove("ALL") 

27 

28 for signal_name in signal_names: 

29 tdi_command = "Build_Signal(TRANSPOSE([" 

30 for sensor_name in sensor_names_on_mds: 

31 tdi_command += f'DATA(GETNCI("\\\\{tree_name_upper}::TOP.{run_name}.CONSTRAINTS.{sensor_type}.{sensor_name}:{signal_name}", "record")), ' 

32 tdi_command = ( 

33 tdi_command[:-2] # remove the last comma and space 

34 + "]), " # closing TRANSPOSE 

35 + "*, " # no data in "raw" 

36 + f'DATA(GETNCI("\\\\{tree_name_upper}::TOP.{run_name}.CONSTRAINTS.{sensor_type}.ALL:NAMES", "record")), DATA(GETNCI("\\\\{tree_name_upper}::TOP.{run_name}:TIME", "record"))' 

37 + ")" 

38 ) 

39 conn.put(f"\\{tree_name_upper}::TOP.{run_name}.CONSTRAINTS.{sensor_type}.ALL:{signal_name}", tdi_command) 

40 

41 conn.put(f"\\{tree_name_upper}::TOP.{run_name}.CONSTRAINTS.{sensor_type}.ALL:NAMES", "$1", np.array(sensor_names_on_mds)) 

42 

43 sensor_types = ["BP_PROBE", "FLUX_LOOP", "ROGOWSKI", "PF_CURRENT", "PRESSURE"] 

44 signal_names = ["INCLUDE", "WEIGHT"] 

45 for sensor_type in sensor_types: 

46 tdi_command = f"getnci('\\\\{tree_name_upper}::TOP.{run_name}.CONSTRAINTS.{sensor_type}.*', 'node_name')" 

47 sensor_names_on_mds_np = conn.get(tdi_command).data() 

48 sensor_names_on_mds = np.char.strip(sensor_names_on_mds_np).astype(str, copy=False).tolist() 

49 sensor_names_on_mds.remove("ALL") 

50 

51 for signal_name in signal_names: 

52 tdi_command = "Build_Signal([" 

53 for sensor_name in sensor_names_on_mds: 

54 tdi_command += f'DATA(GETNCI("\\\\{tree_name_upper}::TOP.{run_name}.CONSTRAINTS.{sensor_type}.{sensor_name}:{signal_name}", "record")), ' 

55 tdi_command = ( 

56 tdi_command[:-2] # remove the last comma and space 

57 + "], " # closing TRANSPOSE 

58 + "*, " # no data in "raw" 

59 + f'DATA(GETNCI("\\\\{tree_name_upper}::TOP.{run_name}.CONSTRAINTS.{sensor_type}.ALL:NAMES", "record")), DATA(GETNCI("\\\\{tree_name_upper}::TOP.{run_name}:TIME", "record"))' 

60 + ")" 

61 ) 

62 conn.put(f"\\{tree_name_upper}::TOP.{run_name}.CONSTRAINTS.{sensor_type}.ALL:{signal_name}", tdi_command) 

63 

64 # Add NAMES 

65 sensor_types = ["BP_PROBE", "FLUX_LOOP", "ROGOWSKI", "PF_CURRENT", "PRESSURE"] 

66 for sensor_type in sensor_types: 

67 tdi_command = f"getnci('\\\\{tree_name_upper}::TOP.{run_name}.CONSTRAINTS.{sensor_type}.*', 'node_name')" 

68 sensor_names_on_mds_np = conn.get(tdi_command).data() 

69 sensor_names_on_mds = np.char.strip(sensor_names_on_mds_np).astype(str, copy=False).tolist() 

70 sensor_names_on_mds.remove("ALL") 

71 

72 conn.put(f"\\{tree_name_upper}::TOP.{run_name}.CONSTRAINTS.{sensor_type}.ALL:NAMES", "$1", np.array(sensor_names_on_mds)) 

73 

74 # Close connection 

75 conn.closeAllTrees()