Coverage for pybeepop/tools.py: 78%
161 statements
« prev ^ index » next coverage.py v7.9.1, created at 2025-06-25 18:27 +0000
1"""
2pybeepop+ Tool and Utility Functions
3"""
5import os
6import io
7import ctypes
8import pandas as pd
10colnames = [ # DataFrame column names for the BeePop+ output
11 "Date",
12 "Colony Size",
13 "Adult Drones",
14 "Adult Workers",
15 "Foragers",
16 "Active Foragers",
17 "Capped Drone Brood",
18 "Capped Worker Brood",
19 "Drone Larvae",
20 "Worker Larvae",
21 "Drone Eggs",
22 "Worker Eggs",
23 "Total Eggs",
24 "DD",
25 "L",
26 "N",
27 "P",
28 "dd",
29 "l",
30 "n",
31 "Free Mites",
32 "Drone Brood Mites",
33 "Worker Brood Mites",
34 "Mites/Drone Cell",
35 "Mites/Worker Cell",
36 "Mites Dying",
37 "Proportion Mites Dying",
38 "Colony Pollen (g)",
39 "Pollen Pesticide Concentration (ug/g)",
40 "Colony Nectar (g)",
41 "Nectar Pesticide Concentration (ug/g)",
42 "Dead Drone Larvae",
43 "Dead Worker Larvae",
44 "Dead Drone Adults",
45 "Dead Worker Adults",
46 "Dead Foragers",
47 "Queen Strength",
48 "Average Temperature (C)",
49 "Rain (mm)",
50 "Min Temp (C)",
51 "Max Temp (C)",
52 "Daylight hours",
53 "Forage Inc",
54 "Forage Day",
55]
58def StringList2CPA(theList):
59 """Utility function to convert a list of strings to bytes readble by the C++ library"""
60 theListBytes = []
61 for i in range(len(theList)):
62 theListBytes.append(bytes(theList[i], "utf-8"))
63 return theListBytes
66class BeePopModel:
67 """Class of background functions to interface with the BeePop+ shared library using CTypes.
69 In most cases users would interact with a PyBeePop object instead of this class.
70 """
72 def __init__(self, library_file, verbose=False):
73 """Initialize the connection to the BeePop+ shared library.
75 Args:
76 library_file (str): Path to the BeePop+ shared library.
77 verbose (bool, optional): Print debugging messages? Defaults to False.
79 Raises:
80 RuntimeError: If BeePop+ passes an error code on initialization.
81 """
82 self.parameters = dict()
83 self.parent = os.path.dirname(os.path.abspath(__file__))
84 self.valid_parameters = pd.read_csv(
85 os.path.join(self.parent, "data/BeePop_exposed_parameters.csv"), skiprows=1
86 )["Exposed Variable Name"].tolist()
87 self.weather_file = None
88 self.contam_file = None
89 self.verbose = verbose
90 self.results = None
91 self.lib = ctypes.CDLL(library_file)
92 self.parent_dir = os.path.dirname(os.path.abspath(__file__))
93 self.lib_status = None
94 if self.lib.InitializeModel(): # Initialize model
95 if self.verbose:
96 print("Model initialized.")
97 else:
98 raise RuntimeError("BeePop+ could not be initialized.")
99 self.clear_buffers()
100 self.send_pars_to_beepop(
101 ["NecPolFileEnable=false"], silent=True
102 ) # disable residue input until given
104 def clear_buffers(self):
105 """Clear C++ buffers in BeePop+"""
106 if not self.lib.ClearResultsBuffer(): # Clear Results and weather lists
107 raise RuntimeError("Error clearing results buffer.")
108 if not self.lib.ClearErrorList():
109 raise RuntimeError("Error clearing error list")
110 if not self.lib.ClearInfoList():
111 raise RuntimeError("Error clearing info")
113 def load_input_file(self, in_file):
114 """Load txt file of BeePop+ parameters."""
115 self.input_file = in_file
116 icf = open(self.input_file)
117 inputs = icf.readlines()
118 icf.close()
119 input_d = dict(x.replace(" ", "").replace("\n", "").split("=") for x in inputs)
120 self.parameter_list_update(input_d) # update parameter dictionary
121 inputlist = []
122 for k, v in self.parameters.items():
123 inputlist.append("{}={}".format(k, v))
124 self.send_pars_to_beepop(inputlist)
125 return self.parameters
127 def parameter_list_update(self, parameters):
128 """Update the internal tracking of set parameters with a dict of
129 parameters: values."""
130 to_add = dict((k.lower(), v) for k, v in parameters.items())
131 self.parameters.update(to_add)
133 def set_parameters(self, parameters=None):
134 """Set BeePop+ parameters based on a dict of parameters: values"""
135 if parameters is not None:
136 self.parameter_list_update(parameters)
137 else:
138 if len(self.parameters) < 1:
139 return
140 inputlist = []
141 for k, v in self.parameters.items():
142 inputlist.append("{}={}".format(k, v))
143 self.send_pars_to_beepop(inputlist)
144 return self.parameters
146 def send_pars_to_beepop(self, parameter_list, silent=False):
147 """Call the BeePop+ interface function to set parameters from a list of
148 parameter=value strings"""
149 for par in parameter_list: # check for invalid parameters
150 par_name = par.split("=")[0].lower()
151 if par_name not in [x.lower() for x in self.valid_parameters]:
152 raise ValueError("{} is not a valid parameter.".format(par_name))
153 CPA = (ctypes.c_char_p * len(parameter_list))()
154 inputlist_bytes = StringList2CPA(parameter_list)
155 CPA[:] = inputlist_bytes
156 if self.lib.SetICVariablesCPA(CPA, len(parameter_list)):
157 if self.verbose and not silent:
158 print("Updated parameters")
159 else:
160 raise RuntimeError("Error setting parameters")
162 def get_parameters(self):
163 """Return the current dict of user defined parameters"""
164 return self.parameters
166 def load_weather(self, weather_file=None):
167 """Load a csv or comma separated txt weather file into BeePop+ using the library interface."""
168 if weather_file is not None:
169 try:
170 wf = open(weather_file)
171 weatherlines = wf.readlines()
172 wf.close()
173 except:
174 raise OSError("Weather file is invalid.")
175 self.weather_file = weather_file
176 CPA = (ctypes.c_char_p * len(weatherlines))()
177 weatherline_bytes = StringList2CPA(weatherlines)
178 CPA[:] = weatherline_bytes
179 if self.lib.SetWeatherCPA(CPA, len(weatherlines)):
180 if self.verbose:
181 print("Loaded Weather")
182 else:
183 raise RuntimeError("Error Loading Weather")
184 else:
185 raise TypeError("Cannot set weather file to None")
187 def load_contam_file(self, contam_file):
188 """Load a csv or comma separated txt of pesticide residues in pollen/nectar using the library interface."""
189 try:
190 ct = open(contam_file)
191 contamlines = ct.readlines()
192 ct.close()
193 self.contam_file = contam_file
194 except:
195 raise OSError("Residue file is invalid.")
196 CPA = (ctypes.c_char_p * len(contamlines))()
197 contamlines_bytes = StringList2CPA(contamlines)
198 CPA[:] = contamlines_bytes
199 if self.lib.SetContaminationTableCPA(CPA, len(contamlines)):
200 if self.verbose:
201 print("Loaded residue file")
202 else:
203 raise RuntimeError("Error loading residue file")
204 self.send_pars_to_beepop(["NecPolFileEnable=true"], silent=True) # enable residue files
206 def set_latitude(self, latitude):
207 """Set the latitude for calculation of day length using the library interface."""
208 c_double_lat = ctypes.c_double(latitude)
209 if self.lib.SetLatitude(c_double_lat):
210 if self.verbose:
211 print("Set Latitude to: {}".format(latitude))
212 else:
213 print("Error setting latitude")
215 def run_beepop(self):
216 """Run the BeePop+ model once using the previously set parameters and weather.
218 Raises:
219 RuntimeError: If BeePop+ passes an error code when running the simulation.
221 Returns:
222 DataFrame: A pandas DataFrame of daily BeePop+ outputs.
223 """
224 if self.lib.RunSimulation():
225 self.lib_status = 1
226 else:
227 self.lib_status = 2
228 raise RuntimeError("Error running BeePop+ simulation.")
229 # fetch results
230 theCount = ctypes.c_int(0)
231 p_Results = ctypes.POINTER(ctypes.c_char_p)()
232 if self.lib.GetResultsCPA(ctypes.byref(p_Results), ctypes.byref(theCount)):
233 # store results
234 n_result_lines = int(theCount.value)
235 self.lib.ClearResultsBuffer()
236 out_lines = []
237 for j in range(0, n_result_lines - 1):
238 out_lines.append(p_Results[j].decode("utf-8", errors="strict"))
239 out_str = io.StringIO("\n".join(out_lines))
240 out_df = pd.read_csv(
241 out_str, sep="\\s+", skiprows=3, names=colnames, dtype={"Date": str}
242 )
243 self.results = out_df
244 else:
245 print("Error running BeePop+ and fetching results.")
246 self.clear_buffers()
247 return self.results
249 def write_results(self, file_path):
250 """Write previously generated BeePop+ outputs to a csv file."""
251 results_file = file_path
252 if self.results is None:
253 raise RuntimeError("There are no results to write. Please run the model first")
254 self.results.to_csv(results_file, index=False)
255 if self.verbose():
256 print("Wrote results to file")
258 def get_errors(self):
259 """Return the BeePop+ error log as a string using the library interface."""
260 p_Errors = ctypes.POINTER(ctypes.c_char_p)()
261 count = ctypes.c_int()
262 if self.lib.GetErrorListCPA(ctypes.byref(p_Errors), ctypes.byref(count)):
263 error_lines = []
264 for j in range(count.value):
265 error_lines.append(p_Errors[j].decode("utf-8", errors="replace"))
266 # self.lib.ClearErrorList()
267 else:
268 raise RuntimeError("Failed to get error log")
269 return "\n".join(error_lines)
271 def get_info(self):
272 """Return the BeePop+ info log as a string using the library interface."""
273 p_Info = ctypes.POINTER(ctypes.c_char_p)()
274 count = ctypes.c_int()
275 if self.lib.GetInfoListCPA(ctypes.byref(p_Info), ctypes.byref(count)):
276 info_lines = []
277 for j in range(count.value):
278 info_lines.append(p_Info[j].decode("utf-8", errors="replace"))
279 # self.lib.ClearInfoList()
280 return "\n".join(info_lines)
282 def get_version(self):
283 """Return the BeePop+ version as a string using the library interface."""
284 buffsize = 16
285 version_buffer = ctypes.create_string_buffer(buffsize)
286 result = self.lib.GetLibVersionCP(version_buffer, buffsize)
287 if result:
288 return version_buffer.value.decode("utf-8")
289 else:
290 raise RuntimeError("Failed to get library version")
292 def close_library(self):
293 """Close connection to the library using CTypes."""
294 dlclose_func = ctypes.CDLL(None).dlclose
295 dlclose_func.argtypes = [ctypes.c_void_p]
296 handle = self.lib._handle
297 self.lib = None
298 del self.lib