Coverage for src/cell_abm_pipeline/flows/plot_basic_metrics.py: 0%

1""" 

2Workflow for plotting basic metrics. 

3 

4Working location structure: 

5 

6.. code-block:: bash 

7 

8 (name) 

9 ├── groups 

10 │ └── groups.BASIC_METRICS 

11 │ ├── (name).metrics_bins.(key).(seed).(tick).(metric).csv 

12 │ ├── (name).metrics_distributions.(metric).json 

13 │ ├── (name).metrics_individuals.(key).(seed).(metric).json 

14 │ ├── (name).metrics_spatial.(key).(seed).(tick).(metric).csv 

15 │ ├── (name).metrics_temporal.(key).(metric).json 

16 │ └── (name).population_counts.(tick).csv 

17 └── plots 

18 └── plots.BASIC_METRICS 

19 ├── (name).metrics_bins.(key).(seed).(tick).(metric).png 

20 ├── (name).metrics_distributions.(metric).png 

21 ├── (name).metrics_individuals.(key).(seed).(metric).png 

22 ├── (name).metrics_spatial.(key).(seed).(tick).(metric).png 

23 ├── (name).metrics_temporal.(key).(metric).png 

24 └── (name).population_counts.(tick).png 

25 

26Plots use grouped data from **groups.BASIC_METRICS**. Plots are saved to 

27**plots.BASIC_METRICS**. 

28""" 

29 

30from dataclasses import dataclass, field 

31from typing import Optional 

32 

33from io_collection.keys import make_key 

34from io_collection.load import load_dataframe, load_json 

35from io_collection.save import save_figure 

36from prefect import flow 

37 

38from cell_abm_pipeline.flows.group_basic_metrics import ( 

39 BIN_METRICS, 

40 CELL_PHASES, 

41 DISTRIBUTION_METRICS, 

42 INDIVIDUAL_METRICS, 

43 SPATIAL_METRICS, 

44 TEMPORAL_METRICS, 

45) 

46from cell_abm_pipeline.tasks import ( 

47 make_bar_figure, 

48 make_density_figure, 

49 make_histogram_figure, 

50 make_line_figure, 

51 make_range_figure, 

52 make_scatter_figure, 

53) 

54 

55PLOTS: list[str] = [ 

56 "metrics_bins", 

57 "metrics_distributions", 

58 "metrics_individuals", 

59 "metrics_spatial", 

60 "metrics_temporal", 

61 "population_counts", 

62] 

63 

64PHASE_COLORS: dict[str, str] = { 

65 "PROLIFERATIVE_G1": "#7F3C8D", 

66 "PROLIFERATIVE_S": "#11A579", 

67 "PROLIFERATIVE_G2": "#3969AC", 

68 "PROLIFERATIVE_M": "#F2B701", 

69 "APOPTOTIC_EARLY": "#E73F74", 

70 "APOPTOTIC_LATE": "#80BA5A", 

71} 

72 

73POPULATION_COLORS: dict[int, str] = { 

74 1: "#7F3C8D", 

75} 

76 

77 

78@dataclass 

79class ParametersConfigMetricsBins: 

80 """Parameter configuration for plot basic metrics subflow - metrics bins.""" 

81 

82 metrics: list[str] = field(default_factory=lambda: BIN_METRICS) 

83 """List of bin metrics.""" 

84 

85 seed: int = 0 

86 """Simulation seed to use for plotting bin metrics.""" 

87 

88 ticks: list[int] = field(default_factory=lambda: [0]) 

89 """Simulation ticks to use for plotting bin metrics.""" 

90 

91 scale: float = 1 

92 """Metric bin scaling.""" 

93 

94 

95@dataclass 

96class ParametersConfigMetricsDistributions: 

97 """Parameter configuration for plot basic metrics subflow - metrics distributions.""" 

98 

99 metrics: list[str] = field(default_factory=lambda: DISTRIBUTION_METRICS) 

100 """List of distribution metrics.""" 

101 

102 phases: list[str] = field(default_factory=lambda: CELL_PHASES) 

103 """List of cell cycle phases.""" 

104 

105 regions: list[str] = field(default_factory=lambda: ["DEFAULT"]) 

106 """List of subcellular regions.""" 

107 

108 

109@dataclass 

110class ParametersConfigMetricsIndividuals: 

111 """Parameter configuration for plot basic metrics subflow - metrics individuals.""" 

112 

113 metrics: list[str] = field(default_factory=lambda: INDIVIDUAL_METRICS) 

114 """List of individual metrics.""" 

115 

116 seed: int = 0 

117 """Simulation seed to use for plotting individual metrics.""" 

118 

119 regions: list[str] = field(default_factory=lambda: ["DEFAULT"]) 

120 """List of subcellular regions.""" 

121 

122 phase_colors: dict[str, str] = field(default_factory=lambda: PHASE_COLORS) 

123 """Colors for each cell cycle phase.""" 

124 

125 

126@dataclass 

127class ParametersConfigMetricsSpatial: 

128 """Parameter configuration for plot basic metrics subflow - metrics spatial.""" 

129 

130 metrics: list[str] = field(default_factory=lambda: SPATIAL_METRICS) 

131 """List of spatial metrics.""" 

132 

133 seeds: list[int] = field(default_factory=lambda: [0]) 

134 """Simulation seeds to use for plotting spatial metrics.""" 

135 

136 regions: list[str] = field(default_factory=lambda: ["DEFAULT"]) 

137 """List of subcellular regions.""" 

138 

139 ticks: list[int] = field(default_factory=lambda: [0]) 

140 """Simulation ticks to use for plotting spatial metrics.""" 

141 

142 phase_colors: dict[str, str] = field(default_factory=lambda: PHASE_COLORS) 

143 """Colors for each cell cycle phase.""" 

144 

145 population_colors: dict[int, str] = field(default_factory=lambda: POPULATION_COLORS) 

146 """Colors for each cell population.""" 

147 

148 

149@dataclass 

150class ParametersConfigMetricsTemporal: 

151 """Parameter configuration for plot basic metrics subflow - metrics temporal.""" 

152 

153 metrics: list[str] = field(default_factory=lambda: TEMPORAL_METRICS) 

154 """List of temporal metrics.""" 

155 

156 regions: list[str] = field(default_factory=lambda: ["DEFAULT"]) 

157 """List of subcellular regions.""" 

158 

159 populations: list[int] = field(default_factory=lambda: [1]) 

160 """List of cell populations.""" 

161 

162 phases: list[str] = field(default_factory=lambda: CELL_PHASES) 

163 """List of cell cycle phases.""" 

164 

165 

166@dataclass 

167class ParametersConfigPopulationCounts: 

168 """Parameter configuration for plot basic metrics subflow - population counts.""" 

169 

170 tick: int = 0 

171 """Simulation tick to use for plotting population counts.""" 

172 

173 

174@dataclass 

175class ParametersConfig: 

176 """Parameter configuration for plot basic metrics flow.""" 

177 

178 plots: list[str] = field(default_factory=lambda: PLOTS) 

179 """List of basic metric plots.""" 

180 

181 metrics_bins: ParametersConfigMetricsBins = ParametersConfigMetricsBins() 

182 """Parameters for plot metrics bins subflow.""" 

183 

184 metrics_distributions: ParametersConfigMetricsDistributions = ( 

185 ParametersConfigMetricsDistributions() 

186 ) 

187 """Parameters for plot metrics distributions subflow.""" 

188 

189 metrics_individuals: ParametersConfigMetricsIndividuals = ParametersConfigMetricsIndividuals() 

190 """Parameters for plot metrics individuals subflow.""" 

191 

192 metrics_spatial: ParametersConfigMetricsSpatial = ParametersConfigMetricsSpatial() 

193 """Parameters for plot metrics spatial subflow.""" 

194 

195 metrics_temporal: ParametersConfigMetricsTemporal = ParametersConfigMetricsTemporal() 

196 """Parameters for plot metrics temporal subflow.""" 

197 

198 population_counts: ParametersConfigPopulationCounts = ParametersConfigPopulationCounts() 

199 """Parameters for plot population counts subflow.""" 

200 

201 

202@dataclass 

203class ContextConfig: 

204 """Context configuration for plot basic metrics flow.""" 

205 

206 working_location: str 

207 """Location for input and output files (local path or S3 bucket).""" 

208 

209 

210@dataclass 

211class SeriesConfig: 

212 """Series configuration for plot basic metrics flow.""" 

213 

214 name: str 

215 """Name of the simulation series.""" 

216 

217 conditions: list[dict] 

218 """List of series condition dictionaries (must include unique condition "key").""" 

219 

220 

221@flow(name="plot-basic-metrics") 

222def run_flow(context: ContextConfig, series: SeriesConfig, parameters: ParametersConfig) -> None: 

223 """ 

224 Main plot basic metrics flow. 

225 

226 Calls the following subflows, if the plot is specified: 

227 

228 - :py:func:`run_flow_plot_metrics_bins` 

229 - :py:func:`run_flow_plot_metrics_distributions` 

230 - :py:func:`run_flow_plot_metrics_individuals` 

231 - :py:func:`run_flow_plot_metrics_spatial` 

232 - :py:func:`run_flow_plot_metrics_temporal` 

233 - :py:func:`run_flow_plot_population_stats` 

234 """ 

235 

236 if "metrics_bins" in parameters.plots: 

237 run_flow_plot_metrics_bins(context, series, parameters.metrics_bins) 

238 

239 if "metrics_distributions" in parameters.plots: 

240 run_flow_plot_metrics_distributions(context, series, parameters.metrics_distributions) 

241 

242 if "metrics_individuals" in parameters.plots: 

243 run_flow_plot_metrics_individuals(context, series, parameters.metrics_individuals) 

244 

245 if "metrics_spatial" in parameters.plots: 

246 run_flow_plot_metrics_spatial(context, series, parameters.metrics_spatial) 

247 

248 if "metrics_temporal" in parameters.plots: 

249 run_flow_plot_metrics_temporal(context, series, parameters.metrics_temporal) 

250 

251 if "population_counts" in parameters.plots: 

252 run_flow_plot_population_counts(context, series, parameters.population_counts) 

253 

254 

255@flow(name="plot-basic-metrics_plot-metrics-bins") 

256def run_flow_plot_metrics_bins( 

257 context: ContextConfig, series: SeriesConfig, parameters: ParametersConfigMetricsBins 

258) -> None: 

259 """Plot basic metrics subflow for binned metrics.""" 

260 

261 group_key = make_key(series.name, "groups", "groups.BASIC_METRICS") 

262 plot_key = make_key(series.name, "plots", "plots.BASIC_METRICS") 

263 keys = [condition["key"] for condition in series.conditions] 

264 

265 for key in keys: 

266 for tick in parameters.ticks: 

267 for metric in parameters.metrics: 

268 metric_key = f"{key}.{parameters.seed:04d}.{tick:06d}.{metric.upper()}" 

269 

270 group = load_dataframe( 

271 context.working_location, 

272 make_key(group_key, f"{series.name}.metrics_bins.{metric_key}.csv"), 

273 ) 

274 

275 save_figure( 

276 context.working_location, 

277 make_key(plot_key, f"{series.name}.metrics_bins.{metric_key}.png"), 

278 make_density_figure(group, parameters.scale), 

279 ) 

280 

281 

282@flow(name="plot-basic-metrics_plot-metrics-distributions") 

283def run_flow_plot_metrics_distributions( 

284 context: ContextConfig, series: SeriesConfig, parameters: ParametersConfigMetricsDistributions 

285) -> None: 

286 """Plot basic metrics subflow for metrics distributions.""" 

287 

288 group_key = make_key(series.name, "groups", "groups.BASIC_METRICS") 

289 plot_key = make_key(series.name, "plots", "plots.BASIC_METRICS") 

290 keys = [condition["key"] for condition in series.conditions] 

291 

292 metrics: list[str] = [] 

293 for metric in parameters.metrics: 

294 if metric in ["volume", "height"]: 

295 metrics = metrics + [f"{metric}.{region}" for region in parameters.regions] 

296 elif metric == "phase": 

297 metrics = metrics + [f"{metric}.{phase}" for phase in parameters.phases] 

298 else: 

299 continue 

300 

301 for metric in metrics: 

302 metric_key = metric.upper() 

303 

304 group = load_json( 

305 context.working_location, 

306 make_key(group_key, f"{series.name}.metrics_distributions.{metric_key}.json"), 

307 ) 

308 

309 assert isinstance(group, dict) 

310 

311 save_figure( 

312 context.working_location, 

313 make_key(plot_key, f"{series.name}.metrics_distributions.{metric_key}.png"), 

314 make_histogram_figure(keys, group), 

315 ) 

316 

317 

318@flow(name="plot-basic-metrics_plot-metrics-individuals") 

319def run_flow_plot_metrics_individuals( 

320 context: ContextConfig, series: SeriesConfig, parameters: ParametersConfigMetricsIndividuals 

321) -> None: 

322 """Plot basic metrics subflow for individual metrics.""" 

323 

324 group_key = make_key(series.name, "groups", "groups.BASIC_METRICS") 

325 plot_key = make_key(series.name, "plots", "plots.BASIC_METRICS") 

326 keys = [condition["key"] for condition in series.conditions] 

327 

328 metrics: list[str] = [ 

329 f"{metric}.{region}" for metric in parameters.metrics for region in parameters.regions 

330 ] 

331 

332 for key in keys: 

333 for metric in metrics: 

334 metric_key = f"{key}.{parameters.seed:04d}.{metric.upper()}" 

335 

336 group = load_json( 

337 context.working_location, 

338 make_key(group_key, f"{series.name}.metrics_individuals.{metric_key}.json"), 

339 ) 

340 

341 group_flat = [ 

342 { 

343 "x": line["time"], 

344 "y": line["value"], 

345 "color": parameters.phase_colors[line["phase"]], 

346 } 

347 for item in group 

348 for line in item 

349 ] 

350 

351 save_figure( 

352 context.working_location, 

353 make_key(plot_key, f"{series.name}.metrics_individuals.{metric_key}.png"), 

354 make_line_figure(group_flat), 

355 ) 

356 

357 

358@flow(name="plot-basic-metrics_plot-metrics-spatial") 

359def run_flow_plot_metrics_spatial( 

360 context: ContextConfig, series: SeriesConfig, parameters: ParametersConfigMetricsSpatial 

361) -> None: 

362 """Plot basic metrics subflow for spatial metrics.""" 

363 

364 group_key = make_key(series.name, "groups", "groups.BASIC_METRICS") 

365 plot_key = make_key(series.name, "plots", "plots.BASIC_METRICS") 

366 keys = [condition["key"] for condition in series.conditions] 

367 

368 metrics: list[str] = [] 

369 for metric in parameters.metrics: 

370 if metric in ["volume", "height"]: 

371 metrics = metrics + [f"{metric}.{region}" for region in parameters.regions] 

372 else: 

373 metrics.append(metric) 

374 

375 for key in keys: 

376 for seed in parameters.seeds: 

377 for tick in parameters.ticks: 

378 for metric in metrics: 

379 metric_key = f"{key}.{seed:04d}.{tick:06d}.{metric.upper()}" 

380 

381 colormap: Optional[dict] = None 

382 

383 if metric == "phase": 

384 colormap = parameters.phase_colors 

385 elif metric == "population": 

386 colormap = parameters.population_colors 

387 

388 group = load_dataframe( 

389 context.working_location, 

390 make_key(group_key, f"{series.name}.metrics_spatial.{metric_key}.csv"), 

391 ) 

392 

393 save_figure( 

394 context.working_location, 

395 make_key(plot_key, f"{series.name}.metrics_spatial.{metric_key}.png"), 

396 make_scatter_figure(group, colormap), 

397 ) 

398 

399 

400@flow(name="plot-basic-metrics_plot-metrics-temporal") 

401def run_flow_plot_metrics_temporal( 

402 context: ContextConfig, series: SeriesConfig, parameters: ParametersConfigMetricsTemporal 

403) -> None: 

404 """Plot basic metrics subflow for temporal metrics.""" 

405 

406 group_key = make_key(series.name, "groups", "groups.BASIC_METRICS") 

407 plot_key = make_key(series.name, "plots", "plots.BASIC_METRICS") 

408 keys = [condition["key"] for condition in series.conditions] 

409 

410 metrics: list[str] = [] 

411 for metric in parameters.metrics: 

412 if metric in ["volume", "height"]: 

413 metrics = metrics + [f"{metric}.{region}" for region in parameters.regions] 

414 elif metric == "population": 

415 metrics = metrics + [f"{metric}.{population}" for population in parameters.populations] 

416 elif metric == "phase": 

417 metrics = metrics + [f"{metric}.{phase}" for phase in parameters.phases] 

418 else: 

419 metrics.append(metric) 

420 

421 for key in keys: 

422 for metric in metrics: 

423 metric_key = f"{key}.{metric.upper()}" 

424 

425 group = load_json( 

426 context.working_location, 

427 make_key(group_key, f"{series.name}.metrics_temporal.{metric_key}.json"), 

428 ) 

429 

430 assert isinstance(group, dict) 

431 

432 save_figure( 

433 context.working_location, 

434 make_key(plot_key, f"{series.name}.metrics_temporal.{metric_key}.png"), 

435 make_range_figure(group), 

436 ) 

437 

438 

439@flow(name="plot-basic-metrics_plot-population-counts") 

440def run_flow_plot_population_counts( 

441 context: ContextConfig, series: SeriesConfig, parameters: ParametersConfigPopulationCounts 

442) -> None: 

443 """Plot basic metrics subflow for population counts.""" 

444 

445 group_key = make_key(series.name, "groups", "groups.BASIC_METRICS") 

446 plot_key = make_key(series.name, "plots", "plots.BASIC_METRICS") 

447 keys = [condition["key"] for condition in series.conditions] 

448 

449 group = load_dataframe( 

450 context.working_location, 

451 make_key(group_key, f"{series.name}.population_counts.{parameters.tick:06d}.csv"), 

452 ) 

453 

454 key_group = { 

455 key: { 

456 "COUNT": { 

457 "mean": group[group["key"] == key]["count"].mean(), 

458 "std": group[group["key"] == key]["count"].std(ddof=1), 

459 } 

460 } 

461 for key in keys 

462 } 

463 

464 save_figure( 

465 context.working_location, 

466 make_key(plot_key, f"{series.name}.population_counts.{parameters.tick:06d}.png"), 

467 make_bar_figure(keys, key_group), 

468 )