<commit_before>import time
import threading
import warnings
import requests
from .persistence import NoPersistenceLayer


class LongshotThread(threading.Thread):
    def __init__(self, device, api_root):
        threading.Thread.__init__(self)
        self.device = device
        self.api_root = api_root
        self.terminating = False

    def _syncpaths(self):
        """synchronize pathpoints against the server"""
        all_pathpoints = list(self.device.pathpoints.keys())

        r = requests.post(self.api_root + '/v1/redefine_paths/',
                          json={
                              'device_id': self.device.device_id,
                              'secret': self.device.secret,
                              'paths': all_pathpoints
                          })

        if r.status_code != 200:
            raise IOError('Failed to redefine paths')

        self.device.paths_synced = True

        r = r.json()

        # Now we need to check default values for sensors that are registered first time
        # in this session

        for path, tsv in r['values'].iteritems():

            try:
                p = self.device.pathpoints[path]
            except KeyError:     # it might have been deleted while we were syncing!
                continue

            if p.registered:        # No need to check it.
                continue

            candidate_ts, candidate_v = tsv
            candidate_ts /= 1000    # server expresses them in milliseconds

            did_server_value_win = False        # because if it does, we need to schedule a sync...
            if p.value is None:     # server value automatically wins...
                did_server_value_win = True
            else: # there is a conflict
                if candidate_ts > p.timestamp:
                    did_server_value_win = True

            if did_server_value_win:
                p._change(candidate_ts, candidate_v)
            else:
                p._schedulesync(p.timestamp, p.value)

            p.registered = True

    def _check_write_queue(self):
        #todo hold writes for invalid paths
        r = requests.post(self.api_root + '/v1/get_writes/',
                          json={'device_id': self.device.device_id, 'secret': self.device.secret})
        if r.status_code != 200:
            raise IOError('Failed to get writes')

        r = r.json()
        paths_updated = []

        for pathpoint, value in r['writes'].iteritems():

            timestamp, value = value
            timestamp = timestamp / 1000    # server counts in ms

            try:
                pp = self.device.pathpoints[pathpoint]
            except KeyError:
                continue
            else:
                paths_updated.append(pathpoint)

            pp._change(timestamp, value)

        if paths_updated:
            r = requests.post(self.api_root + '/v1/confirm_writes/', json={'device_id': self.device.device_id,
                                                                           'secret': self.device.secret,
                                                                           'paths_updated': paths_updated})


    def _syncvalues(self):
        # we did not break - this means there is a need to sync
        sync_dict = {}

        for pathpoint in self.device.pathpoints.values():
            if pathpoint.values_to_store:
                sync_dict[pathpoint.path] = sorted(pathpoint.values_to_store)
                pathpoint.values_to_store = []

        print 'syncing %s' % (sync_dict, )

        r = requests.post(self.api_root + '/v1/sync_values/', json={'device_id': self.device.device_id,
                                                                    'secret': self.device.secret,
                                                                    'values': sync_dict})

        if r.status_code != 200:
            # If you failed syncing, return the values to pool and try another time
            for path, values in sync_dict.iteritems():
                try:
                    self.device.pathpoints[path].values_to_store.extend(values)<commit_msg>fixed timestamp bug in store values<commit_after>import time
import threading
import warnings
import requests
from .persistence import NoPersistenceLayer


class LongshotThread(threading.Thread):
    def __init__(self, device, api_root):
        threading.Thread.__init__(self)
        self.device = device
        self.api_root = api_root
        self.terminating = False

    def _syncpaths(self):
        """synchronize pathpoints against the server"""
        all_pathpoints = list(self.device.pathpoints.keys())

        r = requests.post(self.api_root + '/v1/redefine_paths/',
                          json={
                              'device_id': self.device.device_id,
                              'secret': self.device.secret,
                              'paths': all_pathpoints
                          })

        if r.status_code != 200:
            raise IOError('Failed to redefine paths')

        self.device.paths_synced = True

        r = r.json()

        # Now we need to check default values for sensors that are registered first time
        # in this session

        for path, tsv in r['values'].iteritems():

            try:
                p = self.device.pathpoints[path]
            except KeyError:     # it might have been deleted while we were syncing!
                continue

            if p.registered:        # No need to check it.
                continue

            candidate_ts, candidate_v = tsv
            candidate_ts /= 1000    # server expresses them in milliseconds

            did_server_value_win = False        # because if it does, we need to schedule a sync...
            if p.value is None:     # server value automatically wins...
                did_server_value_win = True
            else: # there is a conflict
                if candidate_ts > p.timestamp:
                    did_server_value_win = True

            if did_server_value_win:
                p._change(candidate_ts, candidate_v)
            else:
                p._schedulesync(p.timestamp, p.value)

            p.registered = True

    def _check_write_queue(self):
        #todo hold writes for invalid paths
        r = requests.post(self.api_root + '/v1/get_writes/',
                          json={'device_id': self.device.device_id, 'secret': self.device.secret})
        if r.status_code != 200:
            raise IOError('Failed to get writes')

        r = r.json()
        paths_updated = []

        for pathpoint, value in r['writes'].iteritems():

            timestamp, value = value
            timestamp = timestamp / 1000    # server counts in ms

            try:
                pp = self.device.pathpoints[pathpoint]
            except KeyError:
                continue
            else:
                paths_updated.append(pathpoint)

            pp._change(timestamp, value)

        if paths_updated:
            r = requests.post(self.api_root + '/v1/confirm_writes/', json={'device_id': self.device.device_id,
                                                                           'secret': self.device.secret,
                                                                           'paths_updated': paths_updated})


    def _syncvalues(self):
        # we did not break - this means there is a need to sync
        sync_dict = {}

        for pathpoint in self.device.pathpoints.values():
            if pathpoint.values_to_store:
                    # server deals in MS
                q = (ts * 1000, v for ts, v in pathpoint.values_to_store)

                sync_dict[pathpoint.path] = sorted(q)
                pathpoint.values_to_store = []

        r = requests.post(self.api_root + '/v1/sync_values/', json={'device_id': self.device.device_id,
                                                                    'secret': self.device.secret,
                                                                    'values': sync_dict})

        if r.status_code != 200:
            # If you failed syncing, return the values to pool and try another time
            for path, values in sync_dict.iteritems():
                try:
                    self.device.pathpoints[path].values_to_store.extend(values)
{'fraction_non_alphanumeric': 0.08078865111805722, 'fraction_numerical': 0.005530175522962251, 'mean_word_length': 1.4525353773584906, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 20, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1310993', 'n_tokens_mistral': 2142, 'n_tokens_neox': 2026, 'n_words': 561}