Protocol for identifying and comparing neuronal ensembles using different algorithms within a graphical user interface.

1. import numpy as np
2. import importlib.metadata
3. from datetime import datetime
4. from concurrent.futures import ProcessPoolExecutor, as_completed
5. from encore.utils.metrics import compute_similarity_matrix
8. def run_single_session(
9. args: tuple[str, str, dict[str, np.ndarray], dict[str, str], dict, bool, list[str]],
10. ) -> tuple[str, dict, dict]:
11. """
12. Auxiliary function to identify the algorithm to run, organize the user's data and
13. organize the result for this run.
14. This is an internal function expected to be used only by
15. :meth:`encore.parallel_runners.sessions.run_parallel_sessions`
16. Every worker uses this function.
18. :param args: Tuple with the arguments for the runner. This is:
19. **algorithm**: str, The 3-chars name of the algorithm, e.g. 'ica', 'svd', 'pca', 'sgc', 'x2p'
20. **database_name**: str, The name of the database used. This is to reference asynchronously the results
21. with the correct database.
22. **database_data**: dict[str, np.ndarray], The data for the analysis.
23. **data_names**: dict[str, str], The names of the API variables.
24. **parameters**: dict, The dictionary with the parameters for the algorithm.
25. **eval_similarity**: bool, whether or not to perform the evaluation of similarities.
26. **similarity_elements**: list[str], List with the names of the variables to compute similarities with.
27. :type args: tuple[str, str, dict[str, np.ndarray], dict[str, str], dict, bool, list[str]]
28. :raises RuntimeError: If the selected algorithm is not known.
29. :return: Tuple with the results of the run. This is:
30. **database_name**: str, the name of the database as it was received
31. **parameters**: dict, the parameters for the algorithm
32. **result**: dict, contains the fields
33. "success": bool, "results": dict, "similarity": dict, "similarity_labels": list[str],
34. :rtype: tuple[str, dict, dict]
35. """
36. algorithm = args[0]
37. database_name = args[1]
38. database_data = args[2]
39. data_names = args[3]
40. parameters = args[4]
41. eval_similarity = args[5]
42. similarity_elements = args[6]
44. print(f"Running [{database_name}]")
46. data = {
47. input_key: database_data[db_key] for input_key, db_key in data_names.items()
48. }
50. # Run algorithm
51. output = {"success": False}
52. try:
53. # Dynamic load of the module and function
54. module_path = "encore.runners.encore"
55. module = importlib.import_module(module_path)
56. function_name = f"run_{algorithm}"
57. func = getattr(module, function_name, None)
59. if func is None or not callable(func):
60. raise RuntimeError(f"The specified algorithm {algorithm} is not defined.")
62. output = func(
63. data,
64. parameters,
65. include_answer=False,
66. )
67. except Exception as exc:
68. print(f"Error for db {database_name}: ", exc)
70. # Failed run
71. if not output["success"]:
72. return (
73. database_name,
74. parameters,
75. {
76. "success": False,
77. "results": None,
78. "similarity": None,
79. "similarity_labels": None,
80. },
81. )
83. # Compute similarity
84. similarities = {}
85. compare_labels = []
86. if eval_similarity:
87. sim_elements = [
88. database_data[elem] for elem in similarity_elements if elem in database_data
89. ]
90. if len(sim_elements) == len(similarity_elements):
91. sim_elements_mat = np.vstack(sim_elements)
92. compare_mat = np.vstack([output["results"]["timecourse"], sim_elements_mat])
93. ensembles_cant = output["results"]["timecourse"].shape[0]
95. compare_labels = [f"ens {i}" for i in range(ensembles_cant)]
96. for element in similarity_elements:
97. # Look for the labels, like "orientation_labels"
98. label_name = f"{element}_labels"
99. element_labels = database_data.get(
100. label_name, list(range(database_data[element].shape[0]))
101. )
102. # Append each new label for each element like "ori 90"
103. compare_labels.extend(f"{label_name[0:3]} {i}" for i in element_labels)
105. sim_methods = ["Cosine", "Euclidean", "Correlation", "Jaccard"]
106. for method in sim_methods:
107. similarities[method] = compute_similarity_matrix(
108. method,
109. compare_mat,
110. )
111. else:
112. if len(sim_elements) == 0:
113. print(f" $$ WARN: No elements for similarity found for {database_name}")
114. else:
115. print(
116. f" $$ WARN: Some elements missing for similarity for {database_name}"
117. )
119. result = {
120. "success": True,
121. "results": output["results"],
122. "similarity": similarities,
123. "similarity_labels": compare_labels,
124. }
126. return database_name, parameters, result
129. def run_parallel_sessions(
130. data: dict[str, dict[str, np.ndarray]],
131. parameters: dict[str, dict],
132. max_workers_cant: int,
133. default_key="default",
134. ):
135. """
136. Run ensembles identification algorithms in parallel for different databases.
137. The parameters for the analysis can be defined for each database or use a default one
138. as a fallback.
140. :param data: Dictionary containing the data for each database.
141. The keys for the first dictionary is a string with the name or ID or the database.
142. Each value at this level contains a dictionary with the data of that database.
143. In this dictionary each key is a specific identifier for a numpy array containing experimental data,
144. e.g. matrix of spikes, fluorescence traces, stimulation, behavior.
145. You can use any name for these variables.
146. :type data: dict[str, dict[str, np.ndarray]]
147. :param parameters: Dictionary where each key is the ID of a database or a fallback name.
148. The dictionary inside should contain the keys:
149. - "analysis", with the 3-letter name of the algorithm to use.
150. - "data_names", This is a dict[str, str] The keys should be one the name of the variables used by the algorithm,
151. e.g "data_neuronal_activity", "data_dFFo" and the value is the name of that variable in the `data` parameter.
152. - "parameters" with another dictionary with the parameters for that specific algorithm, as used by the API.
153. - "evaluate_similarity", a bool variable to evaluate similarity metrics between results/data
154. - "similarity_elements", a list of strings, each string is the key of variables in `data`. When `evaluate_similarity`
155. is True, the batch processing will also calculate the similarity between the ensembles timecourse and this other variables, so
156. these variables should be of the same length as the recording. This is useful to evaluate the performance of each
157. algorithm in every database.
158. :type parameters: dict[str, dict]
159. :param max_workers_cant: Number of workers to use for parallel processing.
160. :type max_workers_cant: int
161. :param default_key: Specific key of the fallback parameter key. These set of parameters
162. will be used for any database that is not explicitly in the keys of the `parameters` argument. Defaults to "default".
163. :type default_key: str, optional
165. :return: Dictionary with the results of the batch analyses. Contains the following keys.
166. **info**: dict[str, str], the date and version information of the analyzer
167. **parameters**: dict, The parameters used for the entire batch analysis.
168. This is a copy of the parameters passed by the user to the function.
169. **parameters_used**: dict, each key is the name of each database, contains the batch parameters used
170. by this database. If it used the default parameters then it's a copy of its contents.
171. **results**: dict, each key is the name of each database. Contains the minimal results of the ensembles algorithm
172. The matrix of similarity and the labels of each similarity matrix and a flag of success for the algorithm.
173. :rtype: dict[str, dict]
175. :raises RuntimeError: If no fallback parameter is assigned and a database was not in the `parameters` keys.
176. """
178. print(
179. f"\n{'=' * 70}\n" f" ENCORE: Running parallel session analysis\n" f"{'=' * 70}"
180. )
182. # Define the results info
183. now = datetime.now()
184. formatted_time = now.strftime("%d%m%y_%H%M%S")
185. encore_version = str(importlib.metadata.version("encore-toolkit"))
186. run_information = {
187. "analyzer": "ENCORE Parallel Sessions API",
188. "date": formatted_time,
189. "ENCORE_version": encore_version,
190. }
192. # Define results
193. results = {
194. "info": run_information,
195. "parameters": parameters,
196. "parameters_used": {},
197. "results": {},
198. }
200. # Process each database as a separate job
201. jobs = []
202. for database_name, database_data in data.items():
203. parameters_set = {}
204. if database_name in parameters:
205. parameters_set = parameters[database_name]
206. elif default_key in parameters:
207. parameters_set = parameters[default_key]
208. else:
209. raise RuntimeError(
210. f"No default parameters under key '{default_key}' set and it was needed."
211. )
213. results["parameters_used"][database_name] = parameters_set
214. jobs.append(
215. (
216. parameters_set.get("analysis"),
217. database_name,
218. database_data,
219. parameters_set.get("data_names"),
220. parameters_set.get("parameters"),
221. parameters_set.get("evaluate_similarity"),
222. parameters_set.get("similarity_elements"),
223. )
224. )
226. print(
227. f"\n -> After matching dbs and params {len(jobs)} databases will be added.\n"
228. )
230. # Parallel execution
231. with ProcessPoolExecutor(max_workers=max_workers_cant) as executor:
232. futures = [executor.submit(run_single_session, job) for job in jobs]
234. for completed, future in enumerate(as_completed(futures), start=1):
235. try:
236. database_name, parameters_used, result = future.result()
237. results["results"][database_name] = result
239. print(f"[{completed}/{len(futures)}] " f"Finished {database_name}.")
241. except Exception as exc:
242. print(f"[{completed}/{len(futures)}] " f"ERROR: {exc}")
244. print(" - Done with databases.")
245. return results

sessions.py at commit 627947f, under GPL-3.0 · at the source