Coverage for nexusLIMS/instruments.py: 100%

1# pylint: disable=duplicate-code 

2""" 

3Methods and representations for instruments in a NexusLIMS system. 

4 

5Attributes 

6---------- 

7instrument_db : dict 

8 A dictionary of :py:class:`~nexusLIMS.db.models.Instrument` objects. 

9 

10 Each object in this dictionary represents an instrument detected in the 

11 NexusLIMS remote database. 

12""" 

13 

14import logging 

15from pathlib import Path 

16 

17from pydantic import ValidationError 

18from sqlmodel import Session as DBSession 

19from sqlmodel import create_engine, select 

20 

21from nexusLIMS.config import settings 

22from nexusLIMS.db.engine import get_engine 

23from nexusLIMS.db.models import Instrument 

24from nexusLIMS.utils.paths import is_subpath 

25 

26logging.basicConfig() 

27_logger = logging.getLogger(__name__) 

28 

29# Lazy-loaded instrument cache. Populated on first access via 

30# _ensure_instrument_db_loaded(). Reset between tests by 

31# SingletonResetter.reset_instrument_cache(). 

32instrument_db: dict = {} 

33_instrument_db_initialized = False 

34 

35 

36def _ensure_instrument_db_loaded(): 

37 """Populate ``instrument_db`` from the database on first call.""" 

38 global _instrument_db_initialized # noqa: PLW0603 

39 if not _instrument_db_initialized: 

40 _instrument_db_initialized = True 

41 instrument_db.update(_get_instrument_db()) 

42 

43 

44def _get_instrument_db(db_path: Path | str | None = None): 

45 """ 

46 Get dictionary of instruments from the NexusLIMS database. 

47 

48 Parameters 

49 ---------- 

50 db_path : Path | str | None, optional 

51 Path to the database file. If None, uses the path from settings. 

52 This parameter is primarily for testing purposes. 

53 

54 Returns 

55 ------- 

56 instrument_db : dict 

57 A dictionary of `Instrument` instances that describe all the 

58 instruments that were found in the ``instruments`` table of the 

59 NexusLIMS database 

60 """ 

61 try: 

62 # Use provided path or fall back to settings 

63 _db_path = db_path if db_path is not None else settings.NX_DB_PATH 

64 

65 # Create temporary engine if non-default path (for testing) 

66 if db_path is not None: 

67 temp_engine = create_engine(f"sqlite:///{_db_path}") 

68 else: 

69 temp_engine = get_engine() 

70 

71 with DBSession(temp_engine) as session: 

72 instruments_list = session.exec(select(Instrument)).all() 

73 return {inst.instrument_pid: inst for inst in instruments_list} 

74 except ValidationError as e: 

75 _logger.debug( 

76 "NexusLIMS config not available (standalone mode); " 

77 "instrument database not loaded. Details: %s", 

78 e, 

79 ) 

80 return {} 

81 except Exception as e: 

82 _logger.warning( 

83 "Could not connect to database or retrieve instruments. " 

84 "Returning empty instrument dictionary.\n\n Details:\n %s", 

85 e, 

86 ) 

87 return {} 

88 

89 

90def get_instr_from_filepath(path: Path) -> Instrument | None: 

91 """ 

92 Get an instrument object by a given path Using the NexusLIMS database. 

93 

94 Parameters 

95 ---------- 

96 path 

97 A path (relative or absolute) to a file saved in the central 

98 filestore that will be used to search for a matching instrument 

99 

100 Returns 

101 ------- 

102 instrument : Instrument or None 

103 An `Instrument` instance matching the path, or None if no match was 

104 found 

105 

106 Examples 

107 -------- 

108 >>> inst = get_instr_from_filepath('/path/to/file.dm3') 

109 >>> str(inst) 

110 'FEI-Titan-TEM-012345 in Bldg 1/Room A' 

111 """ 

112 _ensure_instrument_db_loaded() 

113 for _, v in instrument_db.items(): 

114 if is_subpath( 

115 path, 

116 Path(settings.NX_INSTRUMENT_DATA_PATH) / v.filestore_path, 

117 ): 

118 return v 

119 

120 return None 

121 

122 

123def get_instr_from_calendar_name(cal_name): 

124 """ 

125 Get an instrument object from the NexusLIMS database by its calendar name. 

126 

127 Parameters 

128 ---------- 

129 cal_name : str 

130 A calendar name (e.g. "FEITitanTEMEvents") that will be used to search 

131 for a matching instrument in the ``api_url`` values 

132 

133 Returns 

134 ------- 

135 instrument : Instrument or None 

136 An `Instrument` instance matching the path, or None if no match was 

137 found 

138 

139 Examples 

140 -------- 

141 >>> inst = get_instr_from_calendar_name('FEITitanTEMEvents') 

142 >>> str(inst) 

143 'FEI-Titan-TEM-012345 in Bldg 1/Room A' 

144 """ 

145 _ensure_instrument_db_loaded() 

146 for _, v in instrument_db.items(): 

147 if cal_name in v.api_url: 

148 return v 

149 

150 return None 

151 

152 

153def get_instr_from_api_url(api_url: str) -> Instrument | None: 

154 """ 

155 Get an instrument object from the NexusLIMS database by its ``api_url``. 

156 

157 Parameters 

158 ---------- 

159 api_url 

160 An api_url (e.g. "FEITitanTEMEvents") that will be used to search 

161 for a matching instrument in the ``api_url`` values 

162 

163 Returns 

164 ------- 

165 Instrument 

166 An ``Instrument`` instance matching the ``api_url``, or ``None`` if no 

167 match was found 

168 

169 Examples 

170 -------- 

171 >>> inst = get_instr_from_api_url('https://nemo.example.com/api/tools/?id=1') 

172 >>> str(inst) 

173 'FEI-Titan-STEM-012345 in Bldg 1/Room A' 

174 """ 

175 _ensure_instrument_db_loaded() 

176 for _, v in instrument_db.items(): 

177 if api_url == v.api_url: 

178 return v 

179 

180 return None