pyduino.spectra

  1from pyduino.optimization.nelder_mead import NelderMeadBounded
  2from pyduino.pyduino2 import ReactorManager, chunks, PATHS
  3import numpy as np
  4import json
  5import os
  6import pandas as pd
  7import numpy as np
  8import time
  9from tensorboardX import SummaryWriter
 10from datetime import date, datetime
 11from pyduino.data_parser import RangeParser
 12from collections import OrderedDict
 13from pyduino.utils import bcolors, get_param, partition
 14from pyduino.paths import PATHS
 15from pyduino.log import datetime_to_str, y_to_table, to_markdown_table
 16import traceback
 17import warnings
 18import threading
 19from datetime import datetime
 20import logging
 21
 22__location__ = os.path.realpath(os.path.join(os.getcwd(), os.path.dirname(__file__)))
 23
 24#Path to spectrum.json
 25SPECTRUM_PATH = os.path.join(__location__,"spectrum.json")
 26#Path to irradiance values
 27IRRADIANCE_PATH = os.path.join(__location__,"irradiance.yaml")
 28
 29logging.basicConfig(
 30    filename='pyduino.log',
 31    filemode='w',
 32    level=PATHS.SYSTEM_PARAMETERS.get('log_level', logging.INFO),
 33)
 34
 35
 36def update_dict(D,A,key):
 37    """
 38    Updates dictionary `D` with values in `A` with key name `key`.
 39    
 40    Args:
 41        D (:obj:dict of :obj:dict): Input dictionary.
 42        A (dict): Dict of values to be added with the same keys as in `D`.
 43        key (str): Key name to assign to values of `A`.
 44    """
 45    for k,v in A.items():
 46        D[k].update({key:v})
 47
 48
 49def dict_subset(D,keys):
 50    """
 51    Extracts a subdictionary from a given dictionary.
 52
 53    Args:
 54        D (dict): Input dictionary.
 55        keys (:obj:list of :obj:str): A list of keys pertaining to the dictionary.
 56    """
 57    return {k:D[k] for k in keys if k in D}
 58
 59def row_subset(rows,keys):
 60    """
 61    Extracts a subset from a `rows` table.
 62
 63    Args: 
 64        rows (:obj:`dict` of :obj:`OrderedDict`): Data obtained from `ReactorManager.log_dados`.
 65        keys (:obj:list of :obj:str): A list of keys.
 66    """
 67    return pd.DataFrame(rows).T.loc[:,keys].T.astype(float).astype(str).to_dict()
 68
 69def seval(x):
 70    try:
 71        return eval(x)
 72    except:
 73        return x
 74
 75def parse_dados(X,param):
 76    """
 77    Parses data from `ReactorManager.log_dados` into an array while attempting to convert the input string into an appropriate data type.
 78
 79    Args:
 80        X (:obj:`dict` of :obj:`OrderedDict`): Data obtained from `ReactorManager.log_dados`.
 81        param (str): Parameter name to be extracted from `X`.
 82    """
 83    return np.array(list(map(seval,map(lambda x: x[1].get(param,0),sorted(X.items(),key=lambda x: x[0])))))
 84
 85class Spectra(RangeParser,ReactorManager,NelderMeadBounded):
 86    def __init__(self,
 87        ranges,
 88        density_param,
 89        brilho_param=None,
 90        maximize=True,
 91        log_name=None,
 92        reset_density=False,
 93        summary_writer=None,
 94        **kwargs
 95        ):
 96        """
 97        Initializes the Spectra class.
 98
 99        Args:
100            ranges (dict): A dictionary of parameters with a two-element list containing the minimum and maximum attainable values for each parameter.
101            density_param (str): The name of the parameter to be used as the density count.
102            brilho_param (float, optional): When nonzero, it will be used to turn optimization on or off. Defaults to None.
103            maximize (bool, optional): Whether to maximize the fitness function. Defaults to True.
104            log_name (str, optional): The name of the log. Defaults to None.
105            reset_density (bool, optional): Whether to reset density values on the reactors at each iteration. Defaults to False.
106            **kwargs: Additional keyword arguments.
107
108        Attributes:
109            spectrum (dict): A dictionary containing the spectrum data.
110            parameters (list): A list of relevant system parameters.
111            titled_parameters (list): A list of relevant system parameters with their titles capitalized.
112            irradiance (str): The irradiance value.
113            ids (list): A list of reactor IDs.
114            sorted_ids (list): A sorted list of reactor IDs.
115            tensorboard_path (str): The path to the tensorboard log.
116            writer (SummaryWriter): The tensorboard summary writer.
117            payload (dict): The payload data.
118            data (None): Placeholder for data.
119            do_gotod (bool): Whether to reset density values.
120            dt (float): The time step value.
121
122        Raises:
123            AssertionError: If the spectrum file does not exist.
124
125        """
126        assert os.path.exists(SPECTRUM_PATH)
127        with open(SPECTRUM_PATH) as jfile:
128            self.spectrum = json.loads(jfile.read())
129        
130        self.parameters = PATHS.SYSTEM_PARAMETERS['relevant_parameters']
131        self.titled_parameters = list(map(lambda x: x.title(),self.parameters))
132        RangeParser.__init__(self,ranges,self.parameters)
133
134        self.irradiance = PATHS.SYSTEM_PARAMETERS['irradiance']
135
136        ReactorManager.__init__(self, kwargs.get('include', None))
137        NelderMeadBounded.__init__(
138            self,
139            population_size=len(self.parameters),
140            ranges=self.ranges_as_list(),
141            rng_seed=kwargs.get('rng_seed',0)
142        )
143        self.sorted_ids = sorted(self.ids)
144        self.log_init(name=log_name)
145        if summary_writer is None:
146            self.tensorboard_path = os.path.join(self.log.prefix, "runs", datetime.now().strftime("%Y-%m-%d_%H-%M-%S"))
147            self.writer = SummaryWriter(self.tensorboard_path)
148            logging.info(f"{bcolors.OKGREEN}Created tensorboard log at {self.tensorboard_path}{bcolors.ENDC}")
149        else:
150            self.writer = summary_writer
151            self.tensorboard_path = summary_writer.logdir
152        self.payload = self.population_as_dict if self.payload is None else self.payload
153        self.data = None
154        self.do_gotod = reset_density
155        self.density_param = density_param
156        self.brilho_param = brilho_param
157        self.maximize = maximize
158        self.dt = np.nan
159    def assign_to_reactors(self, x):
160        """
161        Assigns a list of parameters to the reactors.
162
163        Parameters:
164        x (list): The input list to be converted.
165
166        Returns:
167        OrderedDict: An ordered dictionary where the keys are the IDs and the values are the ranges.
168
169        """
170        ids = self.ids[:len(x)]
171        return OrderedDict(
172            zip(
173                ids,
174                map(
175                    lambda u: self.ranges_as_keyed(u),
176                    list(np.round(self.view(x),2))
177                )
178            )
179        )
180    @property
181    def population_as_dict(self):
182        """
183        Converts genome matrix into an appropriate format to send to the reactors.
184        """
185        return self.assign_to_reactors(self.population)
186    
187    @property
188    def power(self):
189        return {reactor_ids: sum(vals[key]*self.irradiance[key] for key in self.irradiance.keys()) for reactor_ids, vals in self.payload.items()}
190
191    def payload_to_matrix(self):
192        return np.nan_to_num(
193            np.array(
194                [[self.payload[i].get(u,np.nan) for u in self.keyed_ranges.keys()] for i in self.ids]
195                ).astype(float)
196            )
197    def data_dict_to_matrix(self,D):
198        return np.nan_to_num(
199            np.array(
200                [[self.D[i].get(u,np.nan) for u in self.parameters] for i in self.ids]
201                ).astype(float)
202            )
203    def F_get(self):
204        """
205        Extracts relevant data from Arduinos.
206        """
207        return self.dados()
208    def F_set(self,x):
209        """
210        Sets parameters to Arduinos.
211        Args:
212            x (:obj:`dict` of :obj:`dict`): Dictionary having reactor id as keys
213            and a dictionary of parameters and their values as values.
214        """
215
216        for _id,params in x.items():
217            for chk in chunks(list(params.items()),3):
218                self.reactors[_id].set(dict(chk))
219                time.sleep(1)
220    def init(self):
221        """
222        Sets payload to the reactors.
223        """
224        self.F_set(self.payload)
225
226    def set_spectrum(self,preset):
227        """
228        Sets all reactors with a preset spectrum contained in `SPECTRUM_PATH`.
229        """
230        command = f"set({','.join(map(lambda x: f'{x[0]},{x[1]}',self.spectrum[preset].items()))})"
231        self.send(command,await_response=False)
232
233    def set_preset_state_spectra(self,*args,**kwargs):
234        self.set_preset_state(*args,**kwargs)
235        self.population = self.inverse_view(self.payload_to_matrix()).astype(int)   
236    
237    def GET(self,tag):
238        """
239        Collects data from Arduinos and logs it to disk.
240        """
241        print("[INFO]","GET",datetime.now().strftime("%c"))
242        self.past_data = self.data.copy() if self.data is not None else pd.DataFrame(self.payload)
243        self.data = self.F_get()
244        self.log.log_many_rows(pd.DataFrame(self.data),tags={'growth_state':tag})
245
246    def log_data(self, i, tags={}):
247        """
248        Logs the tensor values and fitness scores.
249
250        This method iterates over the tensor values and fitness scores and logs them using the writer object.
251        """
252        logging.info(f"LOGGING {datetime.now().strftime('%c')}")
253        data = self.F_get()
254        logging.debug(f"DATA\n{pd.DataFrame(data)}")
255        y = get_param(data, self.density_param, self.reactors)
256        y = {k:float(v) for k,v in y.items()}
257        additional_parameters = {}
258        if PATHS.TENSORBOARD is not None and "additional_parameters" in PATHS.TENSORBOARD:
259            additional_parameters_source = PATHS.TENSORBOARD["additional_parameters"]
260            for param in additional_parameters_source:
261                additional_parameters[param] = get_param(data, param, self.reactors.keys())
262
263        logging.debug(f"ADDITIONAL PARAMETERS {additional_parameters}")
264        
265        P = self.view_g()
266        #Log main parameters
267        for k,(rid, ry) in enumerate(y.items()):
268            self.writer.add_scalar(f'reactor_{rid}/y', float(ry), i)
269            for r_param_id, rparam in enumerate(self.parameters):
270                self.writer.add_scalar(f'reactor_{rid}/{rparam}', float(P[k][r_param_id]), i)
271            for param, value in additional_parameters.items():
272                self.writer.add_scalar(f'reactor_{rid}/{param}', float(value[rid]), i)
273        if self.maximize:
274            self.writer.add_scalar('optima', max(y.values()), i)
275        else:
276            self.writer.add_scalar('optima', min(y.values()), i)
277
278        # Log the DataFrame as a table in text format
279        self.writer.add_text("reactor_state", text_string=to_markdown_table(data), global_step=i)
280        self.log.log_many_rows(data,tags=tags)
281
282    def gotod(self):
283        self.t_gotod_1 = datetime.now()
284        self.send("gotod",await_response=False)
285        logging.debug(f"gotod sent")
286        time.sleep(self.deltaTgotod)
287        self.dt = (datetime.now()-self.t_gotod_1).total_seconds()
288        logging.debug(f"gotod DT {self.dt}")
289
290    # === Optimizer methods ===
291    def ask_oracle(self, X) -> np.ndarray:
292        """
293        Asks the oracle for the fitness of the given input.
294
295        Parameters:
296        X (np.ndarray): The input for which the fitness is to be calculated. Must be already mapped to codomain.
297
298        Returns:
299        np.ndarray: The fitness value calculated by the oracle.
300        """
301
302        assert X.shape[1] == len(self.parameters)
303        assert X.ndim == 2, "X must be a 2D array."
304        partitions = partition(X, len(self.reactors))
305
306        y = pd.Series([])
307
308        for partition in partitions:
309            self.payload = self.assign_to_reactors(partition)
310            reactors = self.payload.keys()
311
312            if self.deltaTgotod is not None and isinstance(self.deltaTgotod, int):
313                self.gotod()
314            data0 = self.F_get()
315            f0 = pd.Series(get_param(data0, self.density_param, reactors))
316
317            self.F_set(self.payload)
318            time.sleep(self.deltaT)
319            data = self.F_get()
320            f = pd.Series(get_param(data, self.density_param, reactors))
321
322            #yield_rate = np.array([(float(f[id])/float(f[id]) - 1)/self.deltaT/self.power[id] for id in reactors]).astype(float)
323            
324            fitness = pd.Series(dict([(id, float(self.power[id])) for id in reactors]))
325
326            partial_y = np.array(((-1)**(self.maximize))*(fitness))
327            y = y.append(partial_y)
328        
329        assert len(y) == len(self.parameters), f"Got different shapes for y: {len(y)} and n params: {len(self.parameters)}"
330        
331        return np.array([y[id] for id in reactors]) 
332    # === * ===
333
334    def run(
335        self,
336        deltaT: float,
337        mode: str = 'optimize',
338        deltaTgotod: int = None
339    ):
340        """
341        Run the bioreactor simulation.
342
343        Args:
344            deltaT (float): The time step for the simulation.
345            mode (str, optional): The mode of operation. Defaults to 'optimize'.
346            deltaTgotod (int, optional): The time interval for performing optimization. Defaults to None.
347
348        Notes:
349            - If mode is 'optimize' and deltaTgotod is less than or equal to 300, a warning will be raised.
350            - If mode is 'free', the number of rows in X must be equal to the number of reactors.
351
352        """
353        if hasattr(self, 'thread') and self.thread.is_alive():
354            self.thread.kill()
355        self.thread = threading.Thread(target=self._run, args=(deltaT, mode, deltaTgotod))
356        self.thread.start()
357    
358    def _run(self, deltaT: float, mode: str = 'optimize', deltaTgotod: int = None):
359        # Checking if gotod time is at least five minutes
360        if mode == "optimize" and isinstance(deltaTgotod, int) and deltaTgotod <= 300:
361            warnings.warn("deltaTgotod should be at least 5 minutes.")
362
363        if mode == "free":
364            assert self.population.shape[0] == len(self.reactors), "X must have the same number of rows as reactors in free mode."
365
366        self.deltaT = deltaT
367        self.deltaTgotod = deltaTgotod
368        self.iteration_counter = 1
369
370        with open("error_traceback.log", "w") as log_file:
371            log_file.write(datetime_to_str(self.log.timestamp) + '\n')
372            try:
373                logging.debug("START")
374                while True:
375                    # growing
376                    self.t_grow_1 = datetime.now()
377                    time.sleep(max(2, deltaT))
378                    self.dt = (datetime.now() - self.t_grow_1).total_seconds()
379                    logging.debug(f"DT {self.dt}")
380                    # Optimizer
381                    if mode == "optimize":
382                        if self.brilho_param is None:
383                            self.step()
384                        else:
385                            brilhos = np.array(list(self.brilho.values()))
386                            if np.all(brilhos > 0):
387                                self.step()
388                            else:
389                                logging.info(f"{self.brilho_param} is off. No optimization steps are being performed.")
390                        if self.deltaTgotod is not None and isinstance(self.deltaTgotod, int):
391                            self.gotod()
392                    self.log_data(self.iteration_counter)
393                    self.iteration_counter += 1
394            except Exception as e:
395                traceback.print_exc(file=log_file)
396                raise(e)
397
398class SpectraManager():
399    """
400    A class that manages multiple Spectra instances.
401
402    Attributes:
403        g (dict): A dictionary containing Spectra instances.
404
405    Methods:
406        call_method: Calls a method on all Spectra instances.
407        get_attr: Retrieves an attribute from all Spectra instances.
408        run: Runs the Spectra instances with specified parameters.
409        reactors: Returns a dictionary of all reactors from Spectra instances.
410    """
411
412    def __init__(self, g:dict):
413        self.g = g
414
415    def __getitem__(self, key):
416        """
417        Allows the SpectraManager class to be indexable.
418
419        Args:
420            key: The key to access the Spectra instance.
421
422        Returns:
423            Spectra: The Spectra instance corresponding to the key.
424        """
425        return self.g[key]
426
427    def call_method(self, method, *args, **kwargs):
428        """
429        Calls a method on all Spectra instances.
430
431        Args:
432            method (str): The name of the method to call.
433            *args: Variable length argument list for the method.
434            **kwargs: Arbitrary keyword arguments for the method.
435        """
436        returns = {}
437        for s in self.g.values():
438            r = getattr(s, method)(*args, **kwargs)
439            returns.update({s:r})
440        return returns
441    
442    def get_attr(self, attr):
443        """
444        Retrieves an attribute from all Spectra instances.
445
446        Args:
447            attr (str): The name of the attribute to retrieve.
448
449        Returns:
450            dict: A dictionary containing the attribute values for each Spectra instance.
451        """
452        return {k:getattr(v, attr) for k,v in self.g.items()}
453    
454    def run(self, deltaT, mode, deltaTgotod):
455        """
456        Runs the Spectra instances with specified parameters.
457
458        Args:
459            deltaT (float): The time step for the simulation.
460            mode (str): The mode of operation for the Spectra instances.
461            deltaTgotod (float): The time step for the go-to-deltaT operation.
462        """
463        for s in self.g.values():
464            s.run(deltaT, mode, deltaTgotod)
465    
466    @property
467    def reactors(self):
468        """
469        Returns a dictionary of all reactors from Spectra instances.
470
471        Returns:
472            dict: A dictionary containing all reactors from Spectra instances.
473        """
474        reactors = {}
475        for s in self.g.values():
476            reactors.update(s.reactors)
477        return reactors
478
479    def __repr__(self) -> str:
480        rstr = bcolors.BOLD + bcolors.OKGREEN + "Main Manager\n" + bcolors.ENDC
481        for v in self.g.values():
482            rstr += f"├── {repr(v)}" + "\n"
483        return rstr
484
485
486if __name__ == "__main__":
487    g = Spectra(**PATHS.HYPERPARAMETERS)