pyduino.pyduino2

  1from typing import Union
  2import logging
  3from time import sleep, time
  4import pandas as pd
  5from collections import OrderedDict
  6from pyduino.log import Log
  7import pandas as pd
  8from multiprocessing import Pool, Process
  9from functools import partial
 10import os
 11from pyduino.utils import bcolors, get_servers, get_meta
 12import requests
 13from urllib.parse import urljoin
 14import logging
 15from datetime import datetime
 16from pyduino.paths import PATHS
 17
 18__location__ = os.path.realpath(os.path.join(os.getcwd(), os.path.dirname(__file__)))
 19logging.basicConfig(
 20    filename='pyduino.log',
 21    filemode='w',
 22    level=PATHS.SYSTEM_PARAMETERS.get('log_level', logging.INFO),
 23)
 24
 25
 26STEP = 1/8
 27HEADER_DELAY = 5
 28COLN = 48 #Number of columns to parse from Arduino (used for sanity tests)
 29CACHEPATH = "cache.csv"
 30
 31#From https://stackoverflow.com/a/312464/6451772
 32def chunks(lst, n):
 33    """Yield successive n-sized chunks from lst."""
 34    for i in range(0, len(lst), n):
 35        yield lst[i:i + n]
 36
 37class Reactor:
 38    """
 39    Master of HTTP Server to Serial handler. 
 40    """
 41
 42    def __init__(self, url):
 43        self.connected = False
 44        self.url = url
 45        self.meta = get_meta(url)
 46        self.id = self.meta["id"]
 47
 48    def http_get(self,route):
 49        return requests.get(urljoin(self.url,route))
 50    
 51    def http_post(self,route,command,await_response):
 52        return requests.post(urljoin(self.url,route),json={
 53            "command": command,
 54            "await_response": await_response
 55        })
 56
 57    def connect(self):
 58        """
 59        Starts connection
 60        """
 61        resp = self.http_get("connect")
 62        if resp.ok:
 63            self.connected = True
 64        return resp.ok
 65    
 66    def reboot(self,retry_time=5):
 67        try:
 68            resp = self.http_get("reboot")
 69        except requests.ConnectionError:
 70            self.connected = False
 71        print(bcolors.WARNING,"Rebooting reactor",self.id,"at",self.url,bcolors.ENDC)
 72        while not self.connected:
 73            try:
 74                self.connect()
 75            except requests.ConnectionError:
 76                print(bcolors.WARNING,"Waiting for reactor",self.id,bcolors.ENDC)
 77                sleep(retry_time)
 78
 79    def reset(self):
 80        """
 81        Resets connection.
 82        """
 83        resp = self.http_get("reset")
 84        return resp.ok
 85
 86    def close(self):
 87        """
 88        Sends 'fim'.
 89        """
 90        self.http_post("send","fim",False,0)
 91
 92    def send(self, msg):
 93        """
 94        Sends command and awaits for a response
 95        """
 96        resp = self.http_post("send",msg,True)
 97        return resp.json()["response"]
 98
 99    def _send(self, msg):
100        """
101        Sends command and doesn't await for a response
102        """
103        resp = self.http_post("send",msg,False)
104        return resp.ok
105    
106    def set_in_chunks(self,params,chunksize=4):
107        """
108        Sets params into chunks.
109        """
110        ch = chunks(params,chunksize)
111        for chunk in ch:
112            self.set(dict(chunk))
113            sleep(2)
114    
115    def __repr__(self):
116        return f"{bcolors.OKCYAN}<Reactor {self.id} at {self.meta['hostname']}({self.url})>{bcolors.ENDC}"
117
118    def set(self, data=None, **kwargs):
119        """
120        Reactor.set({"440": 50, "brilho": 100})
121        """
122        data = {**(data or {}), **kwargs}
123        args = ",".join(f'{k},{v}' for k, v in data.items())
124        cmd = f"set({args})"
125        self._send(cmd)
126    
127    def horacerta(self):
128        """
129        Synchronizes Arduino clock with the client computer.
130        """
131        now = datetime.now()
132        logging.info(f"Set clock on {self.meta['hostname']}")
133        self.set_in_chunks([["ano",now.year],["mes",now.month],["dia",now.day],["hora",now.hour],["minuto",now.minute]],chunksize=2)
134        sleep(2)
135        logging.debug(f"Clock set on {self.meta['hostname']}")
136        self._send("horacerta")
137
138
139def send_wrapper(reactor,command,await_response):
140    id, reactor = reactor
141    if await_response:
142        return (id,reactor.send(command))
143    else:
144        return (id,reactor._send(command))
145
146def reboot_wrapper(reactor):
147    id,reactor = reactor
148    reactor.reboot(retry_time=PATHS.SYSTEM_PARAMETERS["reboot_wait_time"])
149    return True
150
151def set_in_chunks(X):
152    reactor,row,chunksize = X
153    reactor.set_in_chunks(list(row.items()),chunksize)
154
155class ReactorManager:
156    """
157    A class that manages multiple reactors.
158
159    Attributes:
160        pinged (bool): Indicates if the reactors have been pinged.
161        network (str): The network address of the reactors.
162        port (int): The port number of the reactors.
163        exclude (list): A list of reactors to exclude.
164        reactors (dict): A dictionary of reactor objects.
165        servers (dict): A dictionary of server addresses.
166        header (list): A list of header values.
167        payload (dict): A dictionary of payload values.
168        connected (bool): Indicates if the reactors are connected.
169        log (log): A log object for logging data.
170
171    Methods:
172        __init__(self, include: dict = None): Initializes the ReactorManager object.
173        ids(self) -> list: Returns a list of reactor IDs.
174        send(self, command, await_response=True, **kwargs): Sends a command to the reactors.
175        send_parallel(self, command, delay, await_response=True): Sends a command to the reactors in parallel.
176        set(self, data=None, **kwargs): Sets data on the reactors.
177        get(self, key=None): Gets data from the reactors.
178        connect(self): Connects to the reactors.
179        reset(self): Resets the reactors.
180        reboot(self): Reboots the reactors.
181        horacerta(self): Updates Arduino clocks with the clock of the current system.
182        log_init(self, **kwargs): Creates log directories for each Arduino.
183        dados(self, save_cache=True): Gets data from the Arduinos.
184        log_dados(self, save_cache=True): Logs output of `dados` in CSV format.
185        set_preset_state(self, path="preset_state.csv", sep="\t", chunksize=4, params=PATHS.REACTOR_PARAMETERS, **kwargs): Prepares Arduinos with preset parameters from a CSV file.
186        calibrate(self, deltaT=120, dir="calibrate"): Runs `curva` and dumps the result into txts.
187    """
188    
189    pinged = False
190    def __init__(self, include: dict = None):
191        """
192        Initializes the ReactorManager object.
193
194        Args:
195            include (dict): A dictionary of reactor IDs and their corresponding server addresses.
196        """
197        self.network = PATHS.SLAVE_PARAMETERS["network"]
198        self.port = PATHS.SLAVE_PARAMETERS["port"]
199        self.exclude = PATHS.SLAVE_PARAMETERS.get("exclude", None)
200        self.reactors = {}
201
202        if include is None:
203            self.servers = get_servers(self.network,self.port,self.exclude)
204        else:
205            self.servers = include
206
207        for id, host in self.servers.items():
208            self.reactors[id] = Reactor(host)
209        
210        logging.info("Connection completed")
211
212        #Info
213        print("\n".join(map(lambda x: f"Reactor {bcolors.OKCYAN}{x.id}{bcolors.ENDC} at {bcolors.OKCYAN}{x.meta['hostname']}({x.url}){bcolors.ENDC}",self.reactors.values())))
214        self.header = None
215        self.payload = None
216
217        if PATHS.SYSTEM_PARAMETERS.get("sync_clocks", True):
218            self.horacerta()
219        self.ids = sorted(list(self.reactors.keys()))
220
221    def send(self,command,await_response=True,**kwargs):
222        out = {}
223        for k,r in self.reactors.items():
224            if await_response:
225                out[k] = r.send(command,**kwargs)
226            else:
227                r._send(command)
228        return out
229    
230    def send_parallel(self,command,await_response=True):
231        with Pool(7) as p:
232            out = p.map(partial(send_wrapper,command=command,await_response=await_response),list(self.reactors.items()))
233        return dict(out)
234
235    def set(self, data=None, **kwargs):
236        for k,r in self.reactors.items():
237            r.set(data=data, **kwargs)
238    def get(self,key=None):
239        for k,r in self.reactors.items():
240            r.get(key=key)
241    def connect(self):
242        for k,r in self.reactors.items():
243            r.connect()
244    def reset(self):
245        for k,r in self.reactors.items():
246            r.reset()
247        self.connected = True
248    def reboot(self):
249        self.connected = False
250        with Pool(7) as p:
251            response = p.map(reboot_wrapper,list(self.reactors.items()))
252        self.connected = True
253    def horacerta(self):
254        """
255        Updates Arduino clocks with clock of current system.
256        """
257        print("[INFO]", f"{bcolors.BOLD}Syncronizing clocks{bcolors.ENDC}")
258        for k,r in self.reactors.items():
259            r.horacerta()
260
261    #Logging
262
263    def log_init(self,**kwargs):
264        """
265        Creates log directories for each Arduino.
266
267        Args:
268            name (str): Name of the subdirectory in the log folder where the files will be saved.
269        """
270        self.log = Log(subdir=list(self.reactors.keys()),**kwargs)
271        print(f"Log will be saved on: {bcolors.OKGREEN}{self.log.prefix}{bcolors.ENDC}")
272    
273    @property
274    def brilho(self):
275        """
276        Convenience method to get brilho from reactors.
277        """
278        out = self.send_parallel(f"get({self.brilho_param.lower()})")
279        out = {k: float(v.strip()) for k,v in out.items()}
280        return out
281
282    def dados(self,save_cache=True):
283        """
284        Get data from Arduinos.
285
286        Args:
287            save_cache (bool): Whether or not so save a cache file with the last reading with `log.log.cache_data`.
288        """
289
290        if self.header is None:
291            self.header = list(self.reactors.values())[0].send("cabecalho").split(" ")
292
293        len_empty = None
294        while len_empty!=0:
295            rows = self.send_parallel("dados",await_response=True).items()
296            #Checking if any reactor didn't respond.
297            empty = list(filter(lambda x: x[1] is None,rows))
298            len_empty = len(empty)
299            if len_empty!=0:
300                #Error treatment in case some reactor fails to respond.
301                empty = list(map(lambda x: x[0],empty))
302                print(bcolors.FAIL+"[FAIL]","The following reactors didn't respond:"+"\n\t"+"\n\t".join(list(map(str,empty))))
303                print("Resetting serial")
304                sleep(10)
305                for i in empty:
306                    self.reactors[i].reset()
307                    self.reactors[i].connect()
308                print("Recovering last state")
309                self.set_preset_state(path=PATHS.INITIAL_STATE_PATH)
310                self.set_preset_state(path=CACHEPATH)
311                sleep(10)
312                print("Done"+bcolors.ENDC)
313
314        rows = dict(map(lambda x: (x[0],OrderedDict(zip(self.header,x[1].split(" ")))),rows))
315        if save_cache:
316            self.log.cache_data(rows) #Index set to False because ID already exists in rows.
317        return rows
318
319    def log_dados(self,save_cache=True):
320        """
321        Logs output of `dados` in csv format.
322
323        Args:
324            save_cache (bool): Whether or not so save a cache file with the last reading with `log.log.cache_data`.
325        """
326        self.garbage()
327        header = list(self.reactors.values())[0].send("cabecalho").split(" ")
328
329        rows = self.send_parallel("dados",delay=13)
330        rows = list(map(lambda x: (x[0],OrderedDict(zip(header,x[1].split(" ")))),rows))
331
332        for _id,row in rows:
333            self.log.log_rows(rows=[row],subdir=_id)
334        rows = dict(rows)
335        if save_cache:
336            self.log.cache_data(rows) #Index set to False because ID already exists in rows.
337        return rows
338    
339    def set_preset_state(self,path="preset_state.csv",sep="\t",chunksize=4, params=PATHS.REACTOR_PARAMETERS, **kwargs):
340        """
341        Prepare Arduinos with preset parameters from a csv file.
342        Args:
343            path (str): Path to the csv file.
344            chunksize (int): How many to commands to send in a single line. A large value can cause Serial errors.
345            sep (str): Column separator used in the csv.
346        """
347        df = pd.read_csv(path,sep=sep,index_col='ID',**kwargs)
348        df = df[df.index.isin(self.ids)]
349        df.columns = df.columns.str.lower() #Commands must be sent in lowercase
350        #Dropping empty columns
351        df.dropna(axis=1,inplace=True)
352        if params:
353            cols = list(set(df.columns)&set(params))
354            df = df.loc[:,cols]
355        #Setting schema
356        schema = set(df.columns)&PATHS.SCHEMA.keys()
357        schema = {k:PATHS.SCHEMA[k] for k in schema}
358        df = df.astype(schema)
359        with Pool(7) as p:
360            p.map(set_in_chunks,map(lambda x: (self.reactors[x[0]],x[1],chunksize),df.to_dict(orient="index").items()))
361        #Saving relevant parameters' values
362        cols = list(set(df.columns)&set(PATHS.RELEVANT_PARAMETERS))
363        self.preset_state = df.loc[:,cols]
364        self.payload = self.preset_state.to_dict('index').copy()
365
366    def calibrate(self,deltaT=120,dir="calibrate"):
367        """
368        Runs `curva` and dumps the result into txts.
369        """
370        if not os.path.exists(dir):
371            os.mkdir(dir)
372        out = {}
373        self.send("curva",await_response=False)
374        sleep(deltaT)
375        for name,reactor in self.reactors.items():
376            out[name] = reactor._conn.read_until('*** fim da curva dos LEDs ***'.encode('ascii'))
377            with open(os.path.join(dir,f"reator_{name}.txt"),"w") as f:
378                f.write(out[name].decode('ascii'))
379        return out
380    
381    def __repr__(self):
382        return f"{bcolors.OKCYAN}<Manager of reactors {''.join(str(self.ids))}>{bcolors.ENDC}"
383
384
385if __name__ == '__main__':
386    r = ReactorManager()