Skip to content

simulation.py

Simulation 

Simulation(
    name: str | None,
    path: list[str],
    info: dict,
    do_snapshot_grid: bool = False,
    do_verbose: bool = True,
)

Class to manage a single DP Langevin field integration.

Parameters:

  • name

    (str | None) –

    of sim constructed from parameters etc

  • path

    (list[str]) –

    path to file

  • info

    (dict) –

    dictionary containing sim coefficients, model parameters, etc

  • do_snapshot_grid

    (bool, default: False ) –

    flag whether to copy out final time-slice density grid into numpy array

  • do_verbose

    (bool, default: True ) –

    flag whether to use tqdm progress bar, report from dplvn.SimDP

Methods:

  • exec –

    Carry out all simulation steps, including initialization & running.

  • initialize –

    Create and initialize a dpvln.SimSP class instance.

  • plot –

    Generate all the required graphs and images.

  • run –

    Execute a dpvln.SimSP simulation.

  • run_wrapper –

    Wrapper around dpvln.SimSP run to provide timing.

  • save –

    Export outfo JSON, graphs, and data files.

Source code in .venv/lib/python3.14/site-packages/lvn/dp/simulation.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def __init__(
        self, 
        name: str | None, 
        path: list[str], 
        info: dict, 
        do_snapshot_grid: bool=False,
        do_verbose: bool=True,
    ) -> None:
    """
    Constructor.

    Args:
        name: of sim constructed from parameters etc
        path: path to file
        info: dictionary containing sim coefficients, model parameters, etc
        do_snapshot_grid: flag whether to copy out final time-slice
            density grid into numpy array
        do_verbose: flag whether to use tqdm progress bar, report 
            from `dplvn.SimDP`
    """
    self.analysis: dict = info["Analysis"]
    self.parameters: dict = info["Parameters"]
    self.misc: dict = info["Misc"]

    # Henkel et al, 2008
    self.analysis.update({
        "dp_β": 0.5834,
        "dp_ν_pp": 0.7333,
        "dp_ν_ll": 1.2950,
        "dp_δ": 0.4505,
        "dp_z": 1.7660,
    })
    self.misc["path"] = path + [set_name(
        self.parameters, self.analysis, do_dir=True,
    )]
    if name is None:
        self.misc["name"] = set_name(
            self.parameters, self.analysis, do_dir=False,
        )
    # elif name!=set_name(self.parameters, self.analysis,):
    #     raise NameError(f"Problem with {name}")
    else:
        self.misc["name"] = name
        self.misc["path"] = path
    # else:
    #     raise NameError(f"Problem with {name}")
    self.misc["dplvn_version"] = dplvn.__version__
    self.misc["date_time"] \
        = datetime.now().replace(microsecond=0).isoformat(sep=" ")

    self.do_snapshot_grid: bool = do_snapshot_grid
    self.do_verbose: bool = do_verbose
    self.t_epochs: NDArray = np.empty([])
    self.mean_densities: NDArray= np.empty([])
    self.density_dict: dict[float, NDArray] = {}
    self.density_image_dict: dict[int, Any] = {}

exec 

exec() -> Sequence[tuple]

Carry out all simulation steps, including initialization & running.

Returns:

Source code in .venv/lib/python3.14/site-packages/lvn/dp/simulation.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def exec(self) -> Sequence[tuple]:
    """
    Carry out all simulation steps, including initialization & running.

    Returns:
        serialized versions of sim epoch times, mean grid densities, and 
        computation run time.
    """
    self.initialize()
    computation_time_report: str = self.run_wrapper()
    if self.do_verbose:
        print(computation_time_report)
    return (
        tuple(self.t_epochs.tolist()), 
        tuple(self.mean_densities.tolist()),
        self.misc["computation_time"],
    )

initialize 

initialize() -> None

Create and initialize a dpvln.SimSP class instance.

Source code in .venv/lib/python3.14/site-packages/lvn/dp/simulation.py 
92
93
94
95
96
97
98
99
def initialize(self) -> None:
    """
    Create and initialize a `dpvln.SimSP` class instance.
    """
    self.sim = dplvn.SimDP(**self.parameters, do_verbose=self.do_verbose,)
    if not self.sim.initialize(self.misc["n_round_Δt_summation"]):
        raise Exception("Failed to initialize sim")
    self.analysis["n_epochs"] = self.sim.get_n_epochs()

plot 

plot() -> None

Generate all the required graphs and images.

Source code in .venv/lib/python3.14/site-packages/lvn/dp/simulation.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
def plot(self) -> None:
    """
    Generate all the required graphs and images.
    """
    self.graphs: VizDP = VizDP()
    self.images: VizDP = VizDP()
    self.graphs.plot_mean_density_evolution(
        "ρ_t_loglog",
        self.parameters, self.analysis, self.misc,
        self.t_epochs, self.mean_densities, 
        do_rescale=False, y_sf=0.75,
    )
    self.graphs.plot_mean_density_evolution(
        "ρ_t_rescaled",
        self.parameters, self.analysis, self.misc,
        self.t_epochs, self.mean_densities, 
        do_rescale=True,
    )

run 

run() -> None

Execute a dpvln.SimSP simulation.

Source code in .venv/lib/python3.14/site-packages/lvn/dp/simulation.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def run(self) -> None:
    """
    Execute a `dpvln.SimSP` simulation.
    """
    n_segments: int = self.misc["n_segments"]
    n_epochs: int = self.analysis["n_epochs"]
    n_segment_epochs: int = (n_epochs-1) // n_segments
    if (n_segment_epochs*n_segments+1)!=n_epochs:
        raise Exception(
            f"Failed to segment sim with {n_epochs} epochs "
            + f"into {n_segments} segment(s)"
        )
    progress_bar: Callable = (
        progress if self.do_verbose else progress_disabled
    )
    def step(i_segment_: int,):
        if i_segment_>0 and not self.sim.run(n_segment_epochs):
            raise Exception("Failed to run sim")
        if not self.sim.postprocess():
            raise Exception("Failed to process sim results")
        t_epoch_ = self.sim.get_t_current_epoch()
        if self.do_snapshot_grid:
            self.density_dict[t_epoch_] = self.sim.get_density()
    # This ridiculous verbiage is needed because tqdm, even when
    #   disabled, generates some "leaked semaphore objects" errors
    #   when invoked in a `multiprocessing` process
    i_segment_: int
    if self.do_verbose:
        for i_segment_ in progress_bar(range(0, n_segments+1, 1)):
            step(i_segment_)
    else:
        for i_segment_ in range(0, n_segments+1, 1):
            step(i_segment_)
    self.t_epochs = self.sim.get_t_epochs()
    self.mean_densities = self.sim.get_mean_densities()

run_wrapper 

run_wrapper() -> str

Wrapper around dpvln.SimSP run to provide timing.

Returns:

  • str –

    printable string describing computation (sim run) time

Source code in .venv/lib/python3.14/site-packages/lvn/dp/simulation.py 
137
138
139
140
141
142
143
144
145
146
147
148
def run_wrapper(self) -> str:
    """
    Wrapper around `dpvln.SimSP` run to provide timing.

    Returns:
        printable string describing computation (sim run) time
    """
    tick: float = perf_counter()
    self.run()
    tock: float = perf_counter()
    self.misc["computation_time"] = f"{timedelta(seconds=round(tock-tick))}"
    return (f"Computation time = {self.misc["computation_time"]}")

save 

save(module: Any, do_dummy: bool = False, do_verbose: bool = False) -> None

Export outfo JSON, graphs, and data files.

Parameters:

  • module

    (Any) –

    dplvn or other class module

  • do_dummy

    (bool, default: False ) –

    just print (possibly create) the output folders

  • do_verbose

    (bool, default: False ) –

    report how the exporting is going

Source code in .venv/lib/python3.14/site-packages/lvn/dp/simulation.py 
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
def save(
        self, 
        module: Any,
        do_dummy: bool=False, 
        do_verbose: bool=False,
    ) -> None:
    """
    Export outfo JSON, graphs, and data files.

    Args:
        module: dplvn or other class module
        do_dummy: just print (possibly create) the output folders
        do_verbose: report how the exporting is going
    """
    if self.do_verbose | do_verbose:
        print(f"Outfo/graph/data path:  {self.misc["path"]}")
    seed_dir_name: str = f"rs{self.parameters["random_seed"]}"

    outfo_path: str = \
        create_directories(
            (os.path.pardir, *self.misc["path"]), seed_dir_name,
        )
    outfo: dict = {
        "Parameters" : self.parameters,
        "Analysis" : self.analysis,
        "Misc" : self.misc
    }        
    if not do_dummy:
        _ = export_info(outfo_path, "Outfo", outfo, module,)

    if self.misc["do_export_graphs"]:
        graphs_path: str = \
            create_directories(
                (os.path.pardir,  *self.misc["path"], seed_dir_name,), ".",
            )
        if not do_dummy:
            _ = export_plots(
                    self.graphs.fdict, 
                    graphs_path,
                    do_verbose=self.do_verbose,
                )

    if self.misc["do_export_data"]:
        data_path: str = \
            create_directories(
                (os.path.pardir, *self.misc["path"], seed_dir_name,), ".", 
            )
        if not do_dummy:
            np.savez_compressed(
                os.path.join(data_path, "ρ_t",), 
                t_epochs=self.t_epochs,
                mean_densities=self.mean_densities,
            )
            data_npz: NpzFile = np.load(
                os.path.join(data_path, "ρ_t"+".npz",), 
            )
            data_npz["t_epochs"][-10:], data_npz["mean_densities"][-10:]