<commit_before>from __future__ import print_function, division
import serial
import json
import logging
log = logging.getLogger("rfm_ecomanager_logger")
import select
import time
import sys

class NanodeError(Exception):
    """Base class for errors from the Nanode."""


class NanodeRestart(NanodeError):
    """Nanode has restarted."""
    
    
class NanodeTooManyRetries(NanodeError):
    """Nanode has restarted."""
    

class Data(object):
    """Struct for storing data from Nanode"""


class Nanode(object):
    """Used to manage a Nanode running the rfm_edf_ecomanager code."""
    
    MAX_RETRIES = 20
    MAX_ACCEPTABLE_LATENCY = 0.015 # in seconds
    TIME_OFFSET_UPDATE_PERIOD = 60*10 # in seconds
    MAX_ACCEPTABLE_DRIFT = 0.5 # in seconds
    
    def __init__(self, args):
        self.abort = False        
        self.args = args
        self._open_port()
        try:
            self.init_nanode()
        except NanodeRestart:
            self.init_nanode() # re-send init commands after restart
        
    def init_nanode(self):
        log.info("Sending init commands to Nanode...")
        retries = 2
        while retries > 0 and not self.abort:
            retries -= 1
            self.clear_serial()
            self._serial.write("\r")
            
            # Turn off LOGGING on the Nanode if necessary
            try:
                self.send_command("v", 4) # don't show any debug log messages
            except NanodeRestart:
                continue # retry
            except NanodeError:
                pass # if Nanode code was compiled without LOGGING
            
            # Other Nanode config commands...
            self.send_command("m") # manual pairing mode
            self.send_command("k") # Only print data from known transmitters
            self._time_offset = None                    
            self._last_nanode_time = self._get_nanode_time()[1]
            self._set_time_offset()
            break
    
    def _set_time_offset(self):
        retries = 0
        while retries < Nanode.MAX_RETRIES and not self.abort:
            retries += 1
            start_time, nanode_time, end_time = self._get_nanode_time()
            if nanode_time:
                # Nanode sends time 10ms after receipt ocf the 't' command
                new_time_offset = end_time - (nanode_time / 1000)
                
                # Detect rollover
                if nanode_time < self._last_nanode_time:
                    roll_over_detected = True
                    log.info("Rollover detected.")            
                else: 
                    roll_over_detected = False 
                
                # Test if new_time_offset is stupidly different from the
                # old time offset
                if self._time_offset and not roll_over_detected and (
                  new_time_offset > self._time_offset+Nanode.MAX_ACCEPTABLE_DRIFT or
                  new_time_offset < self._time_offset-Nanode.MAX_ACCEPTABLE_DRIFT):
                    log.debug("new_time_offset is too dissimilar to self._time_offset")
                    continue  
                
                # Test
                test_time = new_time_offset + (self._get_nanode_time()[1] / 1000)                
                if (test_time > time.time()+Nanode.MAX_ACCEPTABLE_DRIFT or
                    test_time < time.time()-Nanode.MAX_ACCEPTABLE_DRIFT):
                    log.debug("test_time too dissimilar to time.time(). diff={}"
                                  .format(test_time - time.time()) )
                    continue
                
                # Log time offset details
                log.debug("Updated time_offset to {}".format(new_time_offset))
                if self._time_offset:
                    log.debug("  was {}, diff is {}"
                                  .format(self._time_offset, 
                                          new_time_offset-self._time_offset))
                
                # If we get to here then new_time_offset is sane so save it
                self._time_offset = new_time_offset
                self._deadline_to_update_time_offset = start_time + \
                                     Nanode.TIME_OFFSET_UPDATE_PERIOD
                break
        if nanode_time:
            self._last_nanode_time = nanode_time
    
    def _get_nanode_time(self):
        retries = 0
        while retries < Nanode.MAX_RETRIES and not self.abort:
            retries += 1
            self._serial.flushInput()
            start_time = time.time()
            self._serial.write("t")
            nanode_time = self._readline()
            end_time = time.time()
            latency = end_time - start_time
            log.debug("latency = {}".format(latency))
            
            if latency > Nanode.MAX_ACCEPTABLE_LATENCY:
                log.debug("Latency {} too high".format(latency))
                nanode_time = None
                continue            

            try:
                nanode_time = int(nanode_time)
            except:
                log.debug("Failed to convert {} to an int.".format(nanode_time))
                nanode_time = None
                continue
            else:
                break
        
        return start_time, nanode_time, end_time
    
    def clear_serial(self):
        self._serial.flushInput()
    
    def read_sensor_data(self, retries=MAX_RETRIES):           
        json_line = self._readjson(retries=retries)
        if json_line:
            log.debug("LINE: {}".format(json_line))
            t = time.time()
            data = Data()
            
            # Handle "pair with" responses
            if json_line.get("pw"):
                data.pair_ack = True
                data.tx_id = json_line.get("pw").get("id")
                data.tx_type = json_line.get("pw").get("type")
                return data
            else:
                data.pair_ack = False
            
            data.is_pairing_request = True if json_line.get("pr") else False
            if data.is_pairing_request:
                json_line = json_line.get("pr")
            else:
                # Handle time
                nanode_time = json_line.get("t")
                if self._deadline_to_update_time_offset < time.time():
                    self._set_time_offset()
                elif nanode_time < self._last_nanode_time: # roll-over of Nanode's clock  
                    self._set_time_offset()
                
                self._last_nanode_time = nanode_time
                
                data.timecode = self._time_offset + (nanode_time / 1000)
                log.debug("ETA={:.3f}, time received={:.3f}, diff={:.3f}"
                      .format(data.timecode, t, data.timecode-t))           
                data.timecode = int(round(data.timecode))
                
                data.sensors  = json_line.get("sensors")
                
            data.tx_id    = json_line.get("id")
            data.tx_type  = json_line.get("type")
            data.state    = json_line.get("state")
            return data
        else:
            return None
    
    def _readjson(self, retries=MAX_RETRIES):
        line = self._readline(retries=retries)
        if line and line[0] == "{":
            return json.loads(line)
        
    def _readline_with_exception_handling(self):
        """Wrap serial.readline() with exception handling."""
        try:
            log.debug("Waiting for line from Nanode")
            line = self._serial.readline().strip()
        except select.error:
            if self.abort:
                log.debug("Caught select.error but this is nothing to "
                              "worry about because it was caused by keyboard "<commit_msg>Check if data is available before flushing input when checking Nanode time. This should eliminate the possibility of destroying valid data.<commit_after>from __future__ import print_function, division
import serial
import json
import logging
log = logging.getLogger("rfm_ecomanager_logger")
import select
import time
import sys

class NanodeError(Exception):
    """Base class for errors from the Nanode."""


class NanodeRestart(NanodeError):
    """Nanode has restarted."""
    
    
class NanodeTooManyRetries(NanodeError):
    """Nanode has restarted."""
    
class NanodeDataWaiting(NanodeError):
    """Data is waiting yet this function needs a clear input buffer.
    Caller must read and process data or flushInput before calling this function again"""

class Data(object):
    """Struct for storing data from Nanode"""


class Nanode(object):
    """Used to manage a Nanode running the rfm_edf_ecomanager code."""
    
    MAX_RETRIES = 20
    MAX_ACCEPTABLE_LATENCY = 0.015 # in seconds
    TIME_OFFSET_UPDATE_PERIOD = 60*10 # in seconds
    MAX_ACCEPTABLE_DRIFT = 0.5 # in seconds
    
    def __init__(self, args):
        self.abort = False        
        self.args = args
        self._open_port()
        try:
            self.init_nanode()
        except NanodeRestart:
            self.init_nanode() # re-send init commands after restart
        
    def init_nanode(self):
        log.info("Sending init commands to Nanode...")
        retries = 2
        while retries > 0 and not self.abort:
            retries -= 1
            self._serial.flushInput()
            self._serial.write("\r")
            
            # Turn off LOGGING on the Nanode if necessary
            try:
                self.send_command("v", 4) # don't show any debug log messages
            except NanodeRestart:
                continue # retry
            except NanodeError:
                pass # if Nanode code was compiled without LOGGING
            
            # Other Nanode config commands...
            self.send_command("m") # manual pairing mode
            self.send_command("k") # Only print data from known transmitters
            self._time_offset = None
            break
        
        # Set time offset
        retries = 5
        while retries > 0 and not self.abort:
            retries -= 1
            log.debug("Setting _last_nanode_time and _time_offset for first time. Retries left={}".format(retries))
            self._serial.flushInput()            
            try:
                self._last_nanode_time = self._get_nanode_time()[1]
                self._set_time_offset()
            except NanodeDataWaiting, e:
                log.debug(e)
            else:
                break
    
    def _set_time_offset(self):
        """
        Returns nothing but sets self._last_nanode_time if successful.
        
        Raises:
                NanodeDataWaiting: Data is available on the serial port,
                caller must empty input buffer and retry.
        """
        
        retries = 0
        while retries < Nanode.MAX_RETRIES and not self.abort:
            retries += 1
            start_time, nanode_time, end_time = self._get_nanode_time() # don't catch NanodeDataWaiting exception
            if nanode_time:
                # Nanode sends time 10ms after receipt of the 't' command
                new_time_offset = end_time - (nanode_time / 1000)
                
                # Detect rollover
                if nanode_time < self._last_nanode_time:
                    roll_over_detected = True
                    log.info("Rollover detected.")            
                else: 
                    roll_over_detected = False 
                
                # Test if new_time_offset is stupidly different from the
                # old time offset
                if self._time_offset and not roll_over_detected and (
                  new_time_offset > self._time_offset+Nanode.MAX_ACCEPTABLE_DRIFT or
                  new_time_offset < self._time_offset-Nanode.MAX_ACCEPTABLE_DRIFT):
                    log.debug("new_time_offset is too dissimilar to self._time_offset")
                    continue  
                
                # Test
                test_time = new_time_offset + (self._get_nanode_time()[1] / 1000)                
                if (test_time > time.time()+Nanode.MAX_ACCEPTABLE_DRIFT or
                    test_time < time.time()-Nanode.MAX_ACCEPTABLE_DRIFT):
                    log.debug("test_time too dissimilar to time.time(). diff={}"
                                  .format(test_time - time.time()) )
                    continue
                
                # Log time offset details
                log.debug("Updated time_offset to {}".format(new_time_offset))
                if self._time_offset:
                    log.debug("  was {}, diff is {}"
                                  .format(self._time_offset, 
                                          new_time_offset-self._time_offset))
                
                # If we get to here then new_time_offset is sane so save it
                self._time_offset = new_time_offset
                self._deadline_to_update_time_offset = start_time + \
                                     Nanode.TIME_OFFSET_UPDATE_PERIOD
                break

        if nanode_time:
            self._last_nanode_time = nanode_time
    
    def _get_nanode_time(self):
        """
        Asks the Nanode for the number of milliseconds since it started.
        
        Returns:
            start_time (float): UNIX time immediately before asking Nanode 
                for its time
                
            nanode_time (int): Number of milliseconds since Nanode started
            
            end_time (float): UNIX time immediately after receiving 
                Nanode's time
        
        Raises:
            NanodeDataWaiting: Data is available on the serial port,
                caller must empty input buffer and retry.
        """        
        retries = 0
        while retries < Nanode.MAX_RETRIES and not self.abort:
            retries += 1
            n_waiting = self._serial.inWaiting() # check if any data is waiting for us
            if n_waiting > 0:
                raise NanodeDataWaiting("{} chars waiting".format(n_waiting))
            start_time = time.time()
            self._serial.write("t")
            nanode_time = self._readline()
            end_time = time.time()
            latency = end_time - start_time
            log.debug("latency = {}".format(latency))
            
            if latency > Nanode.MAX_ACCEPTABLE_LATENCY:
                log.debug("Latency {} too high".format(latency))
                nanode_time = None
                continue            

            try:
                nanode_time = int(nanode_time)
            except:
                log.debug("Failed to convert {} to an int.".format(nanode_time))
                nanode_time = None
                continue
            else:
                break
        
        return start_time, nanode_time, end_time
    
    def read_sensor_data(self, retries=MAX_RETRIES):           
        json_line = self._readjson(retries=retries)
        if json_line:
            log.debug("LINE: {}".format(json_line))
            t = time.time()
            data = Data()
            
            # Handle "pair with" responses
            if json_line.get("pw"):
                data.pair_ack = True
                data.tx_id = json_line.get("pw").get("id")
                data.tx_type = json_line.get("pw").get("type")
                return data
            else:
                data.pair_ack = False
            
            data.is_pairing_request = True if json_line.get("pr") else False
            if data.is_pairing_request:
                json_line = json_line.get("pr")
            else:
                # Handle time
                nanode_time = json_line.get("t")
                
                # Decide if we need to update self._time_offset
                if time.time() > self._deadline_to_update_time_offset or \
                  nanode_time < self._last_nanode_time: # roll-over of Nanode's clock
                    
                    try:
                        self._set_time_offset()
                    except NanodeDataWaiting, e:
                        log.debug(e)
                        log.debug("Data is waiting so won't update time on this cycle")
                        
                self._last_nanode_time = nanode_time
                
                data.timecode = self._time_offset + (nanode_time / 1000)
                log.debug("ETA={:.3f}, time received={:.3f}, diff={:.3f}"
                      .format(data.timecode, t, data.timecode-t))           
                data.timecode = int(round(data.timecode))
                
                data.sensors  = json_line.get("sensors")
                
            data.tx_id    = json_line.get("id")
            data.tx_type  = json_line.get("type")
            data.state    = json_line.get("state")
            return data
        else:
            return None
    
    def _readjson(self, retries=MAX_RETRIES):
        line = self._readline(retries=retries)
        if line and line[0] == "{":
            return json.loads(line)
        
    def _readline_with_exception_handling(self):
        """Wrap serial.readline() with exception handling."""
        try:
            log.debug("Waiting for line from Nanode")
            line = self._serial.readline().strip()
        except select.error:
            if self.abort:
                log.debug("Caught select.error but this is nothing to "
                              "worry about because it was caused by keyboard "
{'fraction_non_alphanumeric': 0.07740538075695394, 'fraction_numerical': 0.0047879616963064295, 'mean_word_length': 1.4621105809710917, 'pattern_counts': {'":': 2, '<': 16, '<?xml version=': 0, '>': 14, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5193490', 'n_tokens_mistral': 4887, 'n_tokens_neox': 4566, 'n_words': 1147}