from copy import copy, deepcopy
from multiprocessing import Pool
from DnD_5e import encounter
[docs]
class Simulation:
"""
A class to run an Encounter multiple times and display the result
"""
def __init__(self, **kwargs):
"""
Set up everything needed to run the simulation.
"""
self._name = kwargs.get("name", "Simulation")
if not isinstance(self._name, str):
raise ValueError("The name of the Simulation must be a string")
self._encounter = kwargs.get("encounter")
if not isinstance(self._encounter, encounter.Encounter):
raise ValueError("encounter must be an Encounter")
self._stats = {"combatants": {}, "team": {}}
self.reset_stats()
self._aggregate_stats = {"combatants": {}, "team": {}}
[docs]
def get_name(self) -> str:
"""
:return: name
"""
return self._name
[docs]
def set_name(self, name: str):
self._name = name
[docs]
def get_encounter(self):
"""
:return: stats
"""
return self._encounter
[docs]
def collect_stats(self):
"""
Collect the stats from one run of the Encounter and update *self._stats* accordingly
:return:
"""
stats = self.get_encounter().get_stats()
team_stats = stats["team"]
for team_name in self._stats["team"]:
entry = self._stats["team"][team_name]
out = team_stats[team_name]
for stat in entry:
entry[stat] += out[stat]
combatant_stats = stats["combatants"]
for combatant_name in self._stats["combatants"]:
entry = self._stats["combatants"][combatant_name]
out = combatant_stats[combatant_name]
for stat in entry:
entry[stat] += out[stat]
for stat in self.get_encounter().get_encounter_statnames():
self._stats[stat] += stats[stat]
[docs]
def update_stats(self, stats: dict):
"""
Collect stats from the given dictionary and update *self._stats* accordingly
:param stats: results from one run of the Encounter
:return:
"""
team_stats = stats["team"]
for team_name in self._stats["team"]:
entry = self._stats["team"][team_name]
out = team_stats[team_name]
for stat in entry:
entry[stat] += out[stat]
combatant_stats = stats["combatants"]
for combatant_name in self._stats["combatants"]:
entry = self._stats["combatants"][combatant_name]
out = combatant_stats[combatant_name]
for stat in entry:
entry[stat] += out[stat]
for stat in self.get_encounter().get_encounter_statnames():
self._stats[stat] += stats[stat]
[docs]
def calculate_aggregate_stats(self, n=10000):
"""
Calculate the final stats (i.e., take the average of all the runs by dividing by n. Also calculate some other
aggregate statistics such as proportion_end_hp. Store the final stats in aggregate_stats.
:return:
"""
for combatant_name in self._stats["combatants"]:
entry = self._stats["combatants"][combatant_name]
self._aggregate_stats["combatants"][combatant_name] = {}
out = self._aggregate_stats["combatants"][combatant_name]
out["Average damage dealt"] = entry["damage_dealt"] / n
out["Average damage taken"] = entry["damage_taken"] / n
out["Times gone unconscious"] = entry["times_gone_unconscious"]
times_ended_conscious = entry["times_end_conscious"]
out["Times ended conscious"] = times_ended_conscious
out["Probability of ending conscious"] = times_ended_conscious / n
times_ended_unconscious = entry["times_end_unconscious"]
out["Times ended unconscious"] = times_ended_unconscious
out["Probability of ending unconscious"] = times_ended_unconscious / n
times_ended_dead = entry["times_end_dead"]
out["Times ended dead"] = times_ended_dead
out["Probability of ending dead"] = times_ended_dead / n
out["Average hp at encounter end"] = entry["end_hp"] / n
for team_name in self._stats["team"]:
entry = self._stats["team"][team_name]
self._aggregate_stats["team"][team_name] = {}
out = self._aggregate_stats["team"][team_name]
out["Average number of conscious members"] = entry["num_conscious"] / n
out["Average number of unconscious members"] = entry["num_unconscious"] / n
out["Average number of dead members"] = entry["num_dead"] / n
for stat in self.get_encounter().get_encounter_statnames():
self._aggregate_stats[f"Average {stat}"] = self._stats[stat] / n
[docs]
def print_aggregate_stats(self):
"""
Print the aggregate stats calculated in :py:meth:calculate_aggregate_stats
:return:
"""
print("Combatant stats:")
for combatant_name in self._aggregate_stats["combatants"]:
print(combatant_name, self._aggregate_stats["combatants"][combatant_name])
print()
print("Team stats:")
for team_name in self._aggregate_stats["team"]:
print(team_name, self._aggregate_stats["team"][team_name])
print()
print("Encounter stats:")
for stat in self.get_encounter().get_encounter_statnames():
print(f"Average {stat}", self._aggregate_stats["Average %s" % stat]) # pylint: disable=consider-using-f-string
print()
[docs]
def process_run(self, num): # pylint: disable=unused-argument
"""
The method given to an individual process to do work
:param num: which iteration of the loop am I
:return:
"""
self._encounter.run()
result = self._encounter.get_stats()
self.reset_encounter()
return result
[docs]
def mp_run(self, n=1000, num_procs=None):
"""
Run the Encounter n times (using multiprocessing), then print the results
:param n: number of times to run the encounter
:param num_procs: number of processes to use. If None, use max number of processes
:return: aggregate stats
"""
self.reset_stats()
if num_procs:
pool = Pool(num_procs) # use user-specified number of processes
else:
pool = Pool() # use max number of processes as user didn't specify
for result in pool.imap_unordered(self.process_run, range(n), chunksize=max(n // num_procs, 100)):
self.update_stats(result)
pool.close()
pool.join()
self.calculate_aggregate_stats(n)
self.print_aggregate_stats()
return self._aggregate_stats
[docs]
def run(self, n=1000):
"""
Run the Encounter n times, then print the results
:param n:
:return:
"""
self.reset_stats()
for _ in range(n):
self._encounter.run()
self.collect_stats()
self.reset_encounter()
self.calculate_aggregate_stats(n)
self.print_aggregate_stats()
[docs]
def reset_encounter(self):
"""
Reset the Encounter to prepare for another run.
:return:
"""
self.get_encounter().reset()
[docs]
def reset_stats(self):
"""
Reset stats (or set them for the first time)
:return:
"""
self._aggregate_stats = {"combatants": {}, "team": {}}
self.reset_team_stats()
self.reset_combatant_stats()
self.reset_encounter_stats()
[docs]
def reset_encounter_stats(self):
"""
Reset stats that belong to the Encounter (e.g., number of rounds)
:return:
"""
for stat in self.get_encounter().get_encounter_statnames():
self._stats[stat] = 0
[docs]
def reset_combatant_stats(self):
"""
Reset stats to do with Combatants
:return:
"""
for comb in self.get_encounter().get_combatants():
self._stats["combatants"][comb.get_name()] = {"damage_dealt": 0, "damage_taken": 0, "times_gone_unconscious": 0,
"times_end_conscious": 0, "times_end_unconscious": 0, "times_end_dead": 0,
"end_hp": 0}
[docs]
def reset_team_stats(self):
"""
Reset stats to do with Teams
:return:
"""
for current_team in self.get_encounter().get_teams():
self._stats["team"][current_team.get_name()] = {"num_conscious": 0, "num_unconscious": 0, "num_dead": 0}