|
1 | 1 | import os |
2 | | -import shutil |
3 | 2 | import tempfile |
4 | 3 | from pathlib import Path |
5 | 4 |
|
6 | 5 | from libvcell import vcml_to_finite_volume_input |
7 | 6 |
|
8 | 7 | from pyvcell._internal.solvers.fvsolver import solve as fvsolve |
9 | 8 | from pyvcell.sim_results.result import Result |
10 | | -from pyvcell.vcml import to_vcml_str |
11 | 9 | from pyvcell.vcml.field import Field |
12 | | -from pyvcell.vcml.models import Biomodel |
13 | | - |
14 | | - |
15 | | -class VcmlSpatialSimulation: |
16 | | - bio_model: Biomodel |
17 | | - fields: list[Field] | None |
18 | | - out_dir: Path |
19 | | - |
20 | | - def __init__( |
21 | | - self, |
22 | | - bio_model: Biomodel, |
23 | | - out_dir: Path | str | None = None, |
24 | | - fields: list[Field] | None = None, |
25 | | - ): |
26 | | - self.bio_model = bio_model |
27 | | - self.fields = fields |
28 | | - if out_dir is None: |
29 | | - self.out_dir = Path(tempfile.mkdtemp(prefix="out_dir_")) |
30 | | - else: |
31 | | - self.out_dir = out_dir if isinstance(out_dir, Path) else Path(out_dir) |
32 | | - |
33 | | - def run(self, simulation_name: str) -> Result: |
34 | | - vcml: str = to_vcml_str(bio_model=self.bio_model) |
35 | | - |
36 | | - # check if fields are provided, if yes, write them to the output directory |
37 | | - if self.fields: |
38 | | - for field in self.fields: |
39 | | - fd_path = self.out_dir / field.create_template_filename() |
40 | | - field.write(file_path=fd_path) |
41 | | - |
42 | | - success, error_message = vcml_to_finite_volume_input( |
43 | | - vcml_content=vcml, simulation_name=simulation_name, output_dir_path=self.out_dir |
44 | | - ) |
45 | | - |
46 | | - if not success: |
47 | | - raise ValueError(f"Failed to get solver input files: {error_message}") |
48 | | - |
49 | | - # identify sim_id and job_id from the solver input files |
50 | | - files: list[str] = os.listdir(self.out_dir) |
51 | | - fv_input_file: Path | None = next((self.out_dir / file for file in files if file.endswith(".fvinput")), None) |
52 | | - vcg_input_file: Path | None = next((self.out_dir / file for file in files if file.endswith(".vcg")), None) |
53 | | - if fv_input_file is None or vcg_input_file is None: |
54 | | - raise ValueError(".fvinput file or .vcg file not found") |
55 | | - sim_id = int(fv_input_file.name.split("_")[1]) |
56 | | - job_id = int(fv_input_file.name.split("_")[2]) |
57 | | - |
58 | | - # run the simulation |
59 | | - ret_code = fvsolve(input_file=fv_input_file, vcg_file=vcg_input_file, output_dir=self.out_dir) |
60 | | - if ret_code != 0: |
61 | | - raise ValueError(f"Error in solve: {ret_code}") |
62 | | - |
63 | | - # return the result |
64 | | - return Result(solver_output_dir=self.out_dir, sim_id=sim_id, job_id=job_id) |
65 | | - |
66 | | - def cleanup(self) -> None: |
67 | | - shutil.rmtree(self.out_dir) |
| 10 | +from pyvcell.vcml.models import Biomodel, Simulation |
| 11 | +from pyvcell.vcml.utils import to_vcml_str |
| 12 | +from pyvcell.vcml.workspace import get_workspace_dir |
| 13 | + |
| 14 | + |
| 15 | +def simulate(biomodel: Biomodel, simulation: Simulation | str, fields: list[Field] | None = None) -> Result: |
| 16 | + vcml: str = to_vcml_str(bio_model=biomodel) |
| 17 | + out_dir = Path(tempfile.mkdtemp(prefix="out_dir_", dir=get_workspace_dir())) |
| 18 | + |
| 19 | + # check if fields are provided, if yes, write them to the output directory |
| 20 | + if fields: |
| 21 | + for field in fields: |
| 22 | + fd_path = out_dir / field.create_template_filename() |
| 23 | + field.write(file_path=fd_path) |
| 24 | + |
| 25 | + simulation_name = simulation if isinstance(simulation, str) else simulation.name |
| 26 | + success, error_message = vcml_to_finite_volume_input( |
| 27 | + vcml_content=vcml, simulation_name=simulation_name, output_dir_path=out_dir |
| 28 | + ) |
| 29 | + |
| 30 | + if not success: |
| 31 | + raise ValueError(f"Failed to get solver input files: {error_message}") |
| 32 | + |
| 33 | + # identify sim_id and job_id from the solver input files |
| 34 | + files: list[str] = os.listdir(out_dir) |
| 35 | + fv_input_file: Path | None = next((out_dir / file for file in files if file.endswith(".fvinput")), None) |
| 36 | + vcg_input_file: Path | None = next((out_dir / file for file in files if file.endswith(".vcg")), None) |
| 37 | + if fv_input_file is None or vcg_input_file is None: |
| 38 | + raise ValueError(".fvinput file or .vcg file not found") |
| 39 | + sim_id = int(fv_input_file.name.split("_")[1]) |
| 40 | + job_id = int(fv_input_file.name.split("_")[2]) |
| 41 | + |
| 42 | + # run the simulation |
| 43 | + ret_code = fvsolve(input_file=fv_input_file, vcg_file=vcg_input_file, output_dir=out_dir) |
| 44 | + if ret_code != 0: |
| 45 | + raise ValueError(f"Error in solve: {ret_code}") |
| 46 | + |
| 47 | + # return the result |
| 48 | + return Result(solver_output_dir=out_dir, sim_id=sim_id, job_id=job_id) |
0 commit comments