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

52 statements  

« prev     ^ index     » next       coverage.py v7.1.0, created at 2024-06-05 19:14 +0000

1""" 

2Workflow for converting PhysiCell simulations to other formats. 

3 

4Working location structure: 

5 

6.. code-block:: bash 

7 

8 (name) 

9 ├── converted 

10 │ └── converted.SIMULARIUM 

11 │ └── (name)_(key)_(seed).simularium 

12 └── data 

13 └── (name)_(key)_(seed).tar.xz 

14 

15Formatted data are saved to **converted**. 

16""" 

17 

18from dataclasses import dataclass, field 

19 

20from io_collection.keys import make_key 

21from io_collection.load import load_tar 

22from io_collection.save import save_text 

23from prefect import flow 

24 

25from cell_abm_pipeline.tasks.physicell import convert_physicell_to_simularium 

26 

27FORMATS: list[str] = [ 

28 "simularium", 

29] 

30 

31SUBSTRATE_COLOR = "#d0c5c8" 

32 

33CELL_COLORS = [ 

34 "#fee34d", 

35 "#f7b232", 

36 "#bf5736", 

37 "#94a7fc", 

38 "#ce8ec9", 

39 "#58606c", 

40 "#0ba345", 

41 "#9267cb", 

42 "#81dbe6", 

43 "#bd7800", 

44 "#bbbb99", 

45 "#5b79f0", 

46 "#89a500", 

47 "#da8692", 

48 "#418463", 

49 "#9f516c", 

50 "#00aabf", 

51] 

52 

53 

54@dataclass 

55class ParametersConfig: 

56 """Parameter configuration for convert PhysiCell format flow.""" 

57 

58 box_size: tuple[float, float, float] = (1.0, 1.0, 1.0) 

59 """Size of bounding box.""" 

60 

61 timestep: float = 1.0 

62 """Simulation timestep.""" 

63 

64 formats: list[str] = field(default_factory=lambda: FORMATS) 

65 """List of convert formats.""" 

66 

67 substrate_color: str = SUBSTRATE_COLOR 

68 """Color for substrate.""" 

69 

70 cell_colors: list[str] = field(default_factory=lambda: CELL_COLORS) 

71 """Colors for individual cells.""" 

72 

73 frame_spec: tuple[int, int, int] = (0, 1, 1) 

74 """Specification for simulation ticks to use for converting to simularium.""" 

75 

76 

77@dataclass 

78class ContextConfig: 

79 """Context configuration for convert PhysiCell format flow.""" 

80 

81 working_location: str 

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

83 

84 

85@dataclass 

86class SeriesConfig: 

87 """Series configuration for convert PhysiCell format flow.""" 

88 

89 name: str 

90 """Name of the simulation series.""" 

91 

92 seeds: list[int] 

93 """List of series random seeds.""" 

94 

95 conditions: list[dict] 

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

97 

98 

99@flow(name="convert-physicell-format") 

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

101 """Main convert PhysiCell format flow.""" 

102 

103 if "simularium" in parameters.formats: 

104 run_flow_convert_to_simularium(context, series, parameters) 

105 

106 

107@flow(name="convert-physicell-format_convert-to-simularium") 

108def run_flow_convert_to_simularium( 

109 context: ContextConfig, series: SeriesConfig, parameters: ParametersConfig 

110) -> None: 

111 """Convert PhysiCell format subflow for Simularium.""" 

112 

113 data_key = make_key(series.name, "data") 

114 converted_key = make_key(series.name, "converted", "converted.SIMULARIUM") 

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

116 

117 for key in keys: 

118 for seed in series.seeds: 

119 series_key = f"{series.name}_{key}_{seed:04d}" 

120 tar_key = make_key(data_key, f"{series_key}.tar.xz") 

121 tar_file = load_tar(context.working_location, tar_key) 

122 

123 simularium = convert_physicell_to_simularium( 

124 tar_file, 

125 parameters.box_size, 

126 parameters.timestep, 

127 parameters.frame_spec, 

128 parameters.substrate_color, 

129 parameters.cell_colors, 

130 ) 

131 simularium_key = make_key(converted_key, f"{series_key}.simularium") 

132 save_text(context.working_location, simularium_key, simularium)