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