class TempDirs(object):
    """Tempdir manager."""
    def __init__(self, tmpdir, prefix="rez_"):
        self.tmpdir = tmpdir
        self.prefix = prefix
        self.dirs = set()
        self.lock = Lock()
        # previous version overloaded TempDirs.__del__ in an unsafe manner;
        # the __del__ method is not guaranteed to be called before sys.modules begins
        # breaking down, so the os.path call was failing with a muted AttributeError,
        # leaving the directory on disk even when the program exited normally; by registering
        # an atexit callback we should ensure the directories are cleared at shutdown
        atexit.register(self.clear)

    def mkdtemp(self, cleanup=True):
        path = mkdtemp(dir=self.tmpdir, prefix=self.prefix)
        if not cleanup:
            return path

        try:
            self.lock.acquire()
            self.dirs.add(path)
        finally:
            self.lock.release()
        return path

    def clear(self):
        dirs = self.dirs
        for path in dirs:
            if os.path.exists(path):
                shutil.rmtree(path)

class ProgressBarLogger:
    def __init__(self, msg, total):
        self.msg = msg
        self.total = total
        self.status = 0
        self.lock = Lock()

    def log(self, *_):
        self.lock.acquire()
        self.status += 1
        self._print_progress_bar(self.status, self.total, prefix=self.msg, bar_length=50)
        self.lock.release()

    # from here http://stackoverflow.com/questions/3173320/text-progress-bar-in-the-console
    # Print iterations progress
    @staticmethod
    def _print_progress_bar(iteration, total, prefix='', suffix='', decimals=2, bar_length=100):
        """
        Call in a loop to create terminal progress bar
        @params:
            iteration   - Required  : current iteration (Int)
            total       - Required  : total iterations (Int)
            prefix      - Optional  : prefix string (Str)
            suffix      - Optional  : suffix string (Str)
            decimals    - Optional  : number of decimals in percent complete (Int)
            bar_length   - Optional  : character length of bar (Int)
        """
        filled_length = int(round(bar_length * iteration / float(total)))
        percents = round(100.00 * (iteration / float(total)), decimals)
        bar_char = '#' * filled_length + '-' * (bar_length - filled_length)
        sys.stdout.write('\r%s |%s| %s%s %s' % (prefix, bar_char, percents, '%', suffix))
        sys.stdout.flush()
        if iteration == total:
            sys.stdout.write('\n')
            sys.stdout.flush()

class _CoreScheduleThread(Thread):
    def __init__(self,threadpool):
        self.scheduletasks = [];
        self.tasklock = Lock();
        self.condition = Condition(Lock())
        self.threadpool = threadpool
        Thread.__init__(self)

    def run(self):
        while True:
            self.condition.acquire()
            if len(self.scheduletasks) == 0:
                self.condition.wait();
            else:
                task = self.scheduletasks.pop(0)
                if dates.current_timestamps()>=task.nexttime: 
                    self.threadpool.execute(task.function,*task.args,**task.kwargs)
                    task.nexttime = dates.current_timestamps()+task.period;
                else:
                    self.condition.wait(task.nexttime-dates.current_timestamps())
                self.addtask(task)
            self.condition.release()

    
    def addtask(self,task): # copy on write
        self.tasklock.acquire()
        tasks = [ t for t in self.scheduletasks ]
        tasks.append(task)
        tasks.sort(key=lambda task:task.nexttime)
        self.scheduletasks = tasks
        self.tasklock.release()

    def __init__(self,
                 component_builder=None,
                 project=None,
                 project_dir=None,
                 remote_storage=None,
                 fingerprint=None,
                 pull_models=None):
        self._component_builder = component_builder
        self._models = {}
        self.status = STATUS_READY
        self.current_training_processes = 0
        self._reader_lock = Lock()
        self._loader_lock = Lock()
        self._writer_lock = Lock()
        self._readers_count = 0
        self._path = None
        self._project = project
        self.remote_storage = remote_storage
        self.fingerprint = fingerprint
        self.pull_models = pull_models
        self.error_message = None

        if project and project_dir:
            self._path = os.path.join(project_dir, project)
        self._search_for_models()

class InMemoryItemValue(object):
  _lock = None
  """:type _lock Lock"""

  def __init__(self, value=None, expire_in=None):
    self._lock = Lock()
    self._value = value
    self._expire_in = None
    self._expire_in_time = None

    self.update_expire_time(expire_in)

  @property
  def value(self):
    return self._value

  @value.setter
  def value(self, val):
    self._lock.acquire()
    self._value = val
    self._expire_in = datetime.now() + timedelta(seconds=float(self._expire_in_time)) if self._expire_in_time else None
    self._lock.release()

  def update_expire_time(self, t):
    self._expire_in_time = t

  @property
  def is_expired(self):
    return (self._expire_in - datetime.now()).days < 0 if self._expire_in else False

 def __init__(self, item_number, person_capacity):
     self.items_remaining_lock = Lock()
     self.space_available_lock = Lock()
     self.item_number = item_number
     self.person_capacity = person_capacity
     self.items_remaining = self.item_number
     self.space_available = self.person_capacity

class Queue:
	"""Command queue class
	"""
	def __init__(self):
		self.lock = Lock()
		self.locks = {}
	
	def queue(self, command, *args):
		check = inspect.getargspec(command)
		cmdname = command.__name__ if command.__name__ else "uknown_cmd"
		if len(check[0]) != len(args):
			logging.warn("Queue command '%s' expected %u args, got %u!" % (cmdname, len(check[0]), len(args)))
			
		# If we have enough args, try running the command
		if len(args) >= len(check[0]):
			args = args[:len(check[0])] # Resize arg list if needed
			ret = None
			server = args[0]
			self.lock.acquire()
			if not server in self.locks:
				self.locks[server] = Lock()
			self.lock.release()
			
			self.locks[server].acquire()
			# Run in an enclosure, so as to be able to release lock if it fails
			try:
				ret = command(*args)
			except Exception as err:
				logging.warn("Queue command returned error: %s" % err)
			self.locks[server].release()
			
			if ret:
				return ret
		return None

class PandoraPool(object):
	def __init__(self, poolSize, proxy=None, expireTime=3600):
		self.size = poolSize
		self.proxy = proxy
		self.expire = expireTime
		self.pool = [self.createPandoraAgent() for i in xrange(self.size)]
		self.mutex = Lock()

	def createPandoraAgent(self):
		return PandoraAgent(datetime.now() + timedelta(0, self.expire), self.proxy)

	def refreshPandoraAgent(self, agent):
		if agent.isExpired():
			agent.authenticate_connection()
			agent.setExpireDate(datetime.now() + timedelta(0, self.expire))
		return agent

	def getAgent(self):
		try:
			return self.refreshPandoraAgent(self.pool.pop())
		except IndexError:
			return self.createPandoraAgent()

	def hasAvailableConnections(self):
		return len(self.pool) > 0

	def releaseAgent(self, agent):
		self.mutex.acquire()
		if len(self.pool) < self.size:
			self.pool.append(agent)
		self.mutex.release()

 def __init__(self, url, bucket, password="", verbose=False):
     self.log = logger.logger("VBucketAwareMemcachedClient")
     self.bucket = bucket
     self.rest_username = bucket
     self.rest_password = password
     self._memcacheds = {}
     self._vBucketMap = {}
     self._vBucketMap_lock = Lock()
     self._vBucketMapFastForward = {}
     self._vBucketMapFastForward_lock = Lock()
     #TODO: use regular expressions to parse the url
     server = {}
     if not bucket:
         raise InvalidArgumentException("bucket can not be an empty string", parameters="bucket")
     if not url:
         raise InvalidArgumentException("url can not be an empty string", parameters="url")
     if url.find("http://") != -1 and url.rfind(":") != -1 and url.find("/pools/default") != -1:
         server["ip"] = url[url.find("http://") + len("http://"):url.rfind(":")]
         server["port"] = url[url.rfind(":") + 1:url.find("/pools/default")]
         server["username"] = self.rest_username
         server["password"] = self.rest_password
     self.servers = [server]
     self.servers_lock = Lock()
     self.rest = RestConnection(server)
     self.reconfig_vbucket_map()
     self.init_vbucket_connections()
     self.dispatcher = CommandDispatcher(self)
     self.dispatcher_thread = Thread(name="dispatcher-thread", target=self._start_dispatcher)
     self.dispatcher_thread.daemon = True
     self.dispatcher_thread.start()
     self.streaming_thread = Thread(name="streaming", target=self._start_streaming, args=())
     self.streaming_thread.daemon = True
     self.streaming_thread.start()
     self.verbose = verbose

class DataWindow(Thread):
    def __init__(self,data_adapter):
        Thread.__init__(self)
        self.win = N.zeros((100,3))
        self.winlock = Lock()
        self.data_adapter = data_adapter
    
    def run(self):
        self.data_adapter.start()
        self.running = True
        while self.running:
            self.winlock.acquire()
            try:
                while 1:
                    newdata = self.data_adapter.q.get(block=False)
                    self.win[:-1,:] = self.win[1:,:]
                    self.win[-1,:] = newdata[1:]
            except Queue.Empty:
                pass
            finally:
                self.winlock.release()
        self.data_adapter.stop()
    
    def stop(self):
        self.running = False

class DebuggingLock:
    def __init__(self, name):
        self.lock = Lock()
        self.name = name

    def acquire(self, blocking = 1):
        self.print_tb("Acquire lock")
        self.lock.acquire(blocking)
        self.logmsg("===== %s: Thread %s acquired lock\n"%
            (self.name, currentThread().getName()))

    def release(self):
        self.print_tb("Release lock")
        self.lock.release()

    def logmsg(self, msg):
        loglock.acquire()
        logfile.write(msg + "\n")
        logfile.flush()
        loglock.release()

    def print_tb(self, msg):
        self.logmsg(".... %s: Thread %s attempting to %s\n"% \
                    (self.name, currentThread().getName(), msg) + \
                    "\n".join(traceback.format_list(traceback.extract_stack())))

def run(conn):
    """Function to handle running implosion generation in separate :py:class:`multithreading.Process`

    :param conn: A connection, i.e. one end of a `Pipe()`
    """
    # Need duck-checking instead of real type-checking...
    assert hasattr(conn, 'send') and hasattr(conn, 'recv')

    # Get the implosion object from the pipe:
    imp = conn.recv()
    assert isinstance(imp, Implosion)

    connLock = Lock()

    # Run in a separate thread in this process:
    def impRun():
        nonlocal imp, conn
        try:
            imp.generate()
        except Exception as e:
            connLock.acquire()
            conn.send(e)
            connLock.release()

    t = Thread(target=impRun)
    t.start()

    while t.is_alive():
        connLock.acquire()
        conn.send(imp.progress())
        connLock.release()
        time.sleep(0.01)

    # When the thread is done, send the Implosion object back:
    conn.send(imp)

 def __init__(self):
     self.lock_mess=Lock()
     self.lock_userlist = Lock()
     self.users = []
     self.message_list = []
     super().__init__()
     self.server = self.get_socket()

class BatchInsertCollector():
    def __init__(self, cur, table_name, header=None, threshold=1000000):
        if not isinstance(cur, MyCursor):
            raise TypeError
        self.cur = cur
        self.table_name = table_name
        print 'lyc here'
        print table_name
        if header is None:
            print 'this way'
            self.header = cur.get_header(table_name)
        else:
            print 'that way'
            self.header = header
        print 'I have a header'
        self.sql_header = ''
        self.cur_len = 0
        self.reset_header()
        self.threshold = threshold
        self.values = []
        self.stat_total = 0
        self.mutex = Lock()
        print 'initial finish'

    def __del__(self):
        self.flush()
        self.cur.con.commit()

    def reset_header(self):
        self.sql_header = 'insert into %s (%s) values ' % (self.table_name, ','.join(self.header))
        self.cur_len = len(self.sql_header)

    def flush(self):
        if len(self.values) == 0:
            return
        self.cur.cur.execute(self.sql_header + ','.join(self.values))
        self.cur_len = len(self.sql_header)
        self.cur.con.commit()
        print 'flush called: %d records, total %d records' % (len(self.values), self.stat_total)
        self.values = []

    def append(self, data):
        assert isinstance(data, DictItem)
        self.mutex.acquire()

        def find(val):
            if val not in data.fields:
                return u"''"
            else:
                return u"'%s'" % unicode(data[val])

        cvalues = u','.join(map(find, self.header))
        val1 = u"(%s)" % cvalues
        # print self.cur_len
        if self.cur_len + len(val1) > self.threshold:
            self.flush()
        self.values.append(val1)
        self.cur_len += len(val1) + 1
        self.stat_total += 1
        self.mutex.release()

 def __init__(self, service, channel, config = {}, _lazy = False):
     #  Close the connection while we set it up
     self._closed = True
     
     #  Setup the config
     self._config = deepcopy(DEFAULT_CONFIG)
     self._config.update(config)
     if self._config["connid"] is None:
         self._config["connid"] = "conn{conn_id}".format(conn_id=self._connection_id_generator.__next__())
     
     self._channel = channel
     
     self._local_root = service(weakref.proxy(self))     # Why does this have to be weak
     self._remote_root = None                            # This is set by root.  Why??????
     
     self._local_objects = Locking_dict()        # {oid: native_obj} dictionary, orginal objects to object ids
     self._proxy_cache = WeakValueDictionary()   # {oid: proxy_obj} oid to proxy objects not owned by this connection
     self._netref_classes_cache = {}             # classes that have been??????
     self._netref_proxy_builtin_cls = netref.PROXY_BUILTIN_TYPE_DICT  # Already created in netref
     
     self._seqcounter = itertools.count()        # With this we will generate the msg seq numbers
     
     self._recvlock = Lock()
     self._sendlock = Lock()
     
     self._sync_replies = {}
     self._async_callbacks = {}
     
     self._last_traceback = None
     
     if not _lazy:                               # Should object automatically be set to go
         self._init_service()
     
     # Open this connection as we have finished setting it up
     self._closed = False

 def __init__(
     self, all_cb=None, new_data_cb=None, ack_packet_finished_cb=None, error_cb=None, client_unreachable=None
 ):
     self._com = ""
     self._serial = serial.Serial
     self._idx = 0
     self._buffer = []
     self._crc = crc8()
     ## @brief comport reading thread
     self._thread = Thread(target=self._loop, name="comport")
     # self._thread.setDaemon(True)
     self._comportContinue = threading.Event()
     self._comportContinue.set()
     self._newdata_cb = new_data_cb
     self._ack_packet_finished_cb = ack_packet_finished_cb
     self._client_unreachable = client_unreachable
     self._error_cb = error_cb
     self._master_cb = all_cb
     self._lock = Lock()
     ## @brief wireless transmition packet queue processing thread
     self._send_thread = Thread(target=self._sending_loop2, name="ack payload sender")
     # self._send_thread.setDaemon(True)
     self._sendContinue = threading.Event()
     self._sendContinue.set()
     self._send_queue = Queue.Queue()
     self._packet_inside = False
     self._buffer = []
     self._log = logging.getLogger("root.serialHardware")
     self._tx_finished_lock = Lock()
     self._loop_idx = 0
     self._loop_reciver_timeout_lock = Lock()

class DPMClient():   
    def __init__(self, uid=None, key=None):
        self._lock = Lock()
        self._uid = uid
        self._key = None
        if key:
            self._key = rsa.PublicKey.load_pkcs1(key)
      
    def request(self, addr, port, buf):
        self._lock.acquire()
        try:
            return self._request(addr, port, buf)
        finally:
            self._lock.release()
    
    def _request(self, addr, port, buf):
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.connect((addr, port))
        try:
            if self._key:
                stream = Stream(sock, uid=self._uid, key=self._key)
            else:
                stream = Stream(sock)
            stream.write( buf)
            if self._key:
                stream = Stream(sock)
            _, _, res = stream.readall()
            return res
        finally:
            sock.close()

    def _setup_to_do_n_cycles(self, number_of_cycles: int, updates_each_cycle: UpdateCollection=None):
        """
        Sets up the test so that the retriever will only do n cycles.
        :param number_of_cycles: the number of cycles to do
        """
        if updates_each_cycle is None:
            updates_each_cycle = UpdateCollection([])

        semaphore = Semaphore(0)
        lock_until_counted = Lock()
        lock_until_counted.acquire()

        def increase_counter(*args) -> UpdateCollection:
            semaphore.release()
            lock_until_counted.acquire()
            return updates_each_cycle

        self.retrieval_manager.update_mapper.get_all_since.side_effect = increase_counter
        self.retrieval_manager.start()

        run_counter = 0
        while run_counter < number_of_cycles:
            semaphore.acquire()
            run_counter += 1
            lock_until_counted.release()
            if run_counter == number_of_cycles:
                self.retrieval_manager.stop()

        self.retrieval_manager.update_mapper.get_all_since.side_effect = None

class Manager:
  def __init__(self):
    self.networks = set()
    self.logs = []
    self.next_log_id = 0
    self.log_lock = Lock()

  def notifyNetUp(self, pno, net_name):
    self.networks.add(net_name)
    sys.stdout.write(">> network: %s is up (%d)\n" % (net_name, len(self.networks)))

  def getNetworks(self):
    return self.networks

  def getLogs(self, since=-1):
    if since >= 0:
      return filter(lambda l: l['id'] > since, self.logs)
    else:
      return self.logs

  def putLog(self, host, log):
    self.log_lock.acquire()
    l = {'id': self.next_log_id, 'host': host, 'log': log}
    self.next_log_id = self.next_log_id + 1
    sys.stdout.write(">> log: %s\n" % json.dumps(l))
    self.logs.append(l)
    self.log_lock.release()

class ProcessThread(Thread):
	# Initialize this thread
	def __init__(self):
		Thread.__init__(self)
		self.stopped = Event() # Cancel Event
		self.mutex = Lock()
		self.data = None

	# Threaded code
	def run(self):
		while not self.stopped.isSet():
			# Check if status data is available and process it
			data = None
			self.mutex.acquire()
			if self.data:
				data = self.data
				self.data = None
			self.mutex.release()

			if data:
				global outputfile
				try:
					fp = file(outputfile, 'wu')
					fp.write(data.encode('utf-8'))
					fp.close()
				except Exception, e:
					console.log(LOG_PYTHON, "Couldn't write status to '%s': %s.\n" % (outputfile, str(e)))

			self.stopped.wait(0.5)