benjamintli/cosqa-llm-filtered-hard-negatives huggingface dataset:
Query: terminate while loop after predefined time python process

Relevant code:
def timeout_thread_handler(timeout, stop_event):
    """A background thread to kill the process if it takes too long.

    Args:
        timeout (float): The number of seconds to wait before killing
            the process.
        stop_event (Event): An optional event to cleanly stop the background
            thread if required during testing.
    """

    stop_happened = stop_event.wait(timeout)
    if stop_happened is False:
        print("Killing program due to %f second timeout" % timeout)

    os._exit(2)

Similar but not relevant code:
def wait_until_exit(self):
        """ Wait until thread exit

            Used for testing purpose only
        """

        if self._timeout is None:
            raise Exception("Thread will never exit. Use stop or specify timeout when starting it!")

        self._thread.join()
        self.stop()

Similar but not relevant code:
def wait_until_exit(self):
        """ Wait until all the threads are finished.

        """
        [t.join() for t in self.threads]

        self.threads = list()
Query: test if a row in a file is empty python

Relevant code:
def file_empty(fp):
    """Determine if a file is empty or not."""
    # for python 2 we need to use a homemade peek()
    if six.PY2:
        contents = fp.read()
        fp.seek(0)
        return not bool(contents)

    else:
        return not fp.peek()

Similar but not relevant code:
def isfile_notempty(inputfile: str) -> bool:
        """Check if the input filename with path is a file and is not empty."""
        try:
            return isfile(inputfile) and getsize(inputfile) > 0
        except TypeError:
            raise TypeError('inputfile is not a valid type')

Similar but not relevant code:
def isemptyfile(filepath):
    """Determine if the file both exists and isempty

    Args:
        filepath (str, path): file path

    Returns:
        bool
    """
    exists = os.path.exists(safepath(filepath))
    if exists:
        filesize = os.path.getsize(safepath(filepath))
        return filesize == 0
    else:
        return False

Similar but not relevant code:
def allZero(buffer):
    """
    Tries to determine if a buffer is empty.
    
    @type buffer: str
    @param buffer: Buffer to test if it is empty.
        
    @rtype: bool
    @return: C{True} if the given buffer is empty, i.e. full of zeros,
        C{False} if it doesn't.
    """
    allZero = True
    for byte in buffer:
        if byte != "\x00":
            allZero = False
            break
    return allZero

Similar but not relevant code:
def non_zero_row(arr):
    """
        0.  Empty row returns False.

            >>> arr = array([])
            >>> non_zero_row(arr)

            False

        1.  Row with a zero returns False.

            >>> arr = array([1, 4, 3, 0, 5, -1, -2])
            >>> non_zero_row(arr)

            False
        2.  Row with no zeros returns True.

            >>> arr = array([-1, -0.1, 0.001, 2])
            >>> non_zero_row(arr)

            True

        :param arr: array
        :type arr: numpy array
        :return empty: If row is completely free of zeros
        :rtype empty: bool
    """

    if len(arr) == 0:
        return False

    for item in arr:
        if item == 0:
            return False

    return True
Query: python catch and raise same excepiton

Relevant code:
def reraise(error):
    """Re-raises the error that was processed by prepare_for_reraise earlier."""
    if hasattr(error, "_type_"):
        six.reraise(type(error), error, error._traceback)
    raise error

Similar but not relevant code:
def excepthook(self, etype, value, tb):
      """One more defense for GUI apps that call sys.excepthook.

      GUI frameworks like wxPython trap exceptions and call
      sys.excepthook themselves.  I guess this is a feature that
      enables them to keep running after exceptions that would
      otherwise kill their mainloop. This is a bother for IPython
      which excepts to catch all of the program exceptions with a try:
      except: statement.

      Normally, IPython sets sys.excepthook to a CrashHandler instance, so if
      any app directly invokes sys.excepthook, it will look to the user like
      IPython crashed.  In order to work around this, we can disable the
      CrashHandler and replace it with this excepthook instead, which prints a
      regular traceback using our InteractiveTB.  In this fashion, apps which
      call sys.excepthook will generate a regular-looking exception from
      IPython, and the CrashHandler will only be triggered by real IPython
      crashes.

      This hook should be used sparingly, only in places which are not likely
      to be true IPython errors.
      """
      self.showtraceback((etype,value,tb),tb_offset=0)
Query: python caught sigwinch, shutting down gracefully

Relevant code:
def signal_handler(signal_name, frame):
    """Quit signal handler."""
    sys.stdout.flush()
    print("\nSIGINT in frame signal received. Quitting...")
    sys.stdout.flush()
    sys.exit(0)

Similar but not relevant code:
def sigterm(self, signum, frame):
        """
        These actions will be done after SIGTERM.
        """
        self.logger.warning("Caught signal %s. Stopping daemon." % signum)
        sys.exit(0)

Similar but not relevant code:
def wait_for_shutdown_signal(
        self,
        please_stop=False,  # ASSIGN SIGNAL TO STOP EARLY
        allow_exit=False,  # ALLOW "exit" COMMAND ON CONSOLE TO ALSO STOP THE APP
        wait_forever=True  # IGNORE CHILD THREADS, NEVER EXIT.  False => IF NO CHILD THREADS LEFT, THEN EXIT
    ):
        """
        FOR USE BY PROCESSES THAT NEVER DIE UNLESS EXTERNAL SHUTDOWN IS REQUESTED

        CALLING THREAD WILL SLEEP UNTIL keyboard interrupt, OR please_stop, OR "exit"

        :param please_stop:
        :param allow_exit:
        :param wait_forever:: Assume all needed threads have been launched. When done
        :return:
        """
        self_thread = Thread.current()
        if self_thread != MAIN_THREAD or self_thread != self:
            Log.error("Only the main thread can sleep forever (waiting for KeyboardInterrupt)")

        if isinstance(please_stop, Signal):
            # MUTUAL SIGNALING MAKES THESE TWO EFFECTIVELY THE SAME SIGNAL
            self.please_stop.on_go(please_stop.go)
            please_stop.on_go(self.please_stop.go)
        else:
            please_stop = self.please_stop

        if not wait_forever:
            # TRIGGER SIGNAL WHEN ALL CHILDREN THEADS ARE DONE
            with self_thread.child_lock:
                pending = copy(self_thread.children)
            children_done = AndSignals(please_stop, len(pending))
            children_done.signal.on_go(self.please_stop.go)
            for p in pending:
                p.stopped.on_go(children_done.done)

        try:
            if allow_exit:
                _wait_for_exit(please_stop)
            else:
                _wait_for_interrupt(please_stop)
        except KeyboardInterrupt as _:
            Log.alert("SIGINT Detected!  Stopping...")
        except SystemExit as _:
            Log.alert("SIGTERM Detected!  Stopping...")
        finally:
            self.stop()

Similar but not relevant code:
def _turn_sigterm_into_systemexit(): # pragma: no cover
    """
    Attempts to turn a SIGTERM exception into a SystemExit exception.
    """
    try:
        import signal
    except ImportError:
        return
    def handle_term(signo, frame):
        raise SystemExit
    signal.signal(signal.SIGTERM, handle_term)
Query: python chaining filter functions

Relevant code:
def filter(self, func):
        """Returns a SndRcv list filtered by a truth function"""
        return self.__class__( [ i for i in self.res if func(*i) ], name='filtered %s'%self.listname)

Similar but not relevant code:
def BROADCAST_FILTER_NOT(func):
        """
        Composes the passed filters into an and-joined filter.
        """
        return lambda u, command, *args, **kwargs: not func(u, command, *args, **kwargs)
Query: testing the symbol set with nested brackets in python

Relevant code:
def is_SYMBOL(token, *symbols):
    """ Returns True if ALL of the given argument are AST nodes
    of the given token (e.g. 'BINARY')
    """
    from symbols.symbol_ import Symbol
    assert all(isinstance(x, Symbol) for x in symbols)
    for sym in symbols:
        if sym.token != token:
            return False

    return True

Similar but not relevant code:
def bound_symbols(self):
        """Set of bound SymPy symbols contained within the equation."""
        try:
            lhs_syms = self.lhs.bound_symbols
        except AttributeError:
            lhs_syms = set()
        try:
            rhs_syms = self.rhs.bound_symbols
        except AttributeError:
            rhs_syms = set()
        return lhs_syms | rhs_syms

Similar but not relevant code:
def init_checks_registry():
    """Register all globally visible functions.

    The first argument name is either 'physical_line' or 'logical_line'.
    """
    mod = inspect.getmodule(register_check)
    for (name, function) in inspect.getmembers(mod, inspect.isfunction):
        register_check(function)

Similar but not relevant code:
def contains(self, element):
        """
        Ensures :attr:`subject` contains *other*.
        """
        self._run(unittest_case.assertIn, (element, self._subject))
        return ChainInspector(self._subject)

Similar but not relevant code:
def functions(self):
        """
        A list of functions declared or defined in this module.
        """
        return [v for v in self.globals.values()
                if isinstance(v, values.Function)]

Similar but not relevant code:
def instance_contains(container, item):
    """Search into instance attributes, properties and return values of no-args methods."""
    return item in (member for _, member in inspect.getmembers(container))
Query: python chang fraction to decimal number in string

Relevant code:
def format_float(value): # not used
    """Modified form of the 'g' format specifier.
    """
    string = "{:g}".format(value).replace("e+", "e")
    string = re.sub("e(-?)0*(\d+)", r"e\1\2", string)
    return string

Similar but not relevant code:
def fixed(ctx, number, decimals=2, no_commas=False):
    """
    Formats the given number in decimal format using a period and commas
    """
    value = _round(ctx, number, decimals)
    format_str = '{:f}' if no_commas else '{:,f}'
    return format_str.format(value)

Similar but not relevant code:
def print_float(self, value, decimal_digits=2, justify_right=True):
        """Print a numeric value to the display.  If value is negative
        it will be printed with a leading minus sign.  Decimal digits is the
        desired number of digits after the decimal point.
        """
        format_string = '{{0:0.{0}F}}'.format(decimal_digits)
        self.print_number_str(format_string.format(value), justify_right)

Similar but not relevant code:
def _saferound(value, decimal_places):
    """
    Rounds a float value off to the desired precision
    """
    try:
        f = float(value)
    except ValueError:
        return ''
    format = '%%.%df' % decimal_places
    return format % f

Similar but not relevant code:
def trim_decimals(s, precision=-3):
        """
        Convert from scientific notation using precision
        """
        encoded = s.encode('ascii', 'ignore')
        str_val = ""
        if six.PY3:
            str_val = str(encoded, encoding='ascii', errors='ignore')[:precision]
        else:
            # If precision is 0, this must be handled seperately
            if precision == 0:
                str_val = str(encoded)
            else:
                str_val = str(encoded)[:precision]
        if len(str_val) > 0:
            return float(str_val)
        else:
            return 0

Similar but not relevant code:
def truncate(value: Decimal, n_digits: int) -> Decimal:
    """Truncates a value to a number of decimals places"""
    return Decimal(math.trunc(value * (10 ** n_digits))) / (10 ** n_digits)
Query: python change backward slash to forward slash

Relevant code:
def norm_slash(name):
    """Normalize path slashes."""

    if isinstance(name, str):
        return name.replace('/', "\\") if not is_case_sensitive() else name
    else:
        return name.replace(b'/', b"\\") if not is_case_sensitive() else name

Similar but not relevant code:
def slugify(string):
    """
    Removes non-alpha characters, and converts spaces to hyphens. Useful for making file names.


    Source: http://stackoverflow.com/questions/5574042/string-slugification-in-python
    """
    string = re.sub('[^\w .-]', '', string)
    string = string.replace(" ", "-")
    return string

Similar but not relevant code:
def camelcase_to_slash(name):
    """ Converts CamelCase to camel/case

    code ripped from http://stackoverflow.com/questions/1175208/does-the-python-standard-library-have-function-to-convert-camelcase-to-camel-cas
    """

    s1 = re.sub('(.)([A-Z][a-z]+)', r'\1/\2', name)
    return re.sub('([a-z0-9])([A-Z])', r'\1/\2', s1).lower()

Similar but not relevant code:
def normalise_string(string):
    """ Strips trailing whitespace from string, lowercases it and replaces
        spaces with underscores
    """
    string = (string.strip()).lower()
    return re.sub(r'\W+', '_', string)
Query: textwrap python doensnt work

Relevant code:
def wrap(text, indent='    '):
    """Wrap text to terminal width with default indentation"""
    wrapper = textwrap.TextWrapper(
        width=int(os.environ.get('COLUMNS', 80)),
        initial_indent=indent,
        subsequent_indent=indent
    )
    return '\n'.join(wrapper.wrap(text))

Similar but not relevant code:
def toggle_word_wrap(self):
        """
        Toggles document word wrap.

        :return: Method success.
        :rtype: bool
        """

        self.setWordWrapMode(not self.wordWrapMode() and QTextOption.WordWrap or QTextOption.NoWrap)
        return True
Query: the order of an element in an index python

Relevant code:
def get_feature_order(dataset, features):
    """ Returns a list with the order that features requested appear in
    dataset """
    all_features = dataset.get_feature_names()

    i = [all_features.index(f) for f in features]

    return i

Similar but not relevant code:
def sortlevel(self, level=None, ascending=True, sort_remaining=None):
        """
        For internal compatibility with with the Index API.

        Sort the Index. This is for compat with MultiIndex

        Parameters
        ----------
        ascending : boolean, default True
            False to sort in descending order

        level, sort_remaining are compat parameters

        Returns
        -------
        Index
        """
        return self.sort_values(return_indexer=True, ascending=ascending)
Query: time in am/pm in python to utc time

Relevant code:
def datetime_local_to_utc(local):
    """
    Simple function to convert naive :std:`datetime.datetime` object containing
    local time to a naive :std:`datetime.datetime` object with UTC time.
    """
    timestamp = time.mktime(local.timetuple())
    return datetime.datetime.utcfromtimestamp(timestamp)

Similar but not relevant code:
def convert_timeval(seconds_since_epoch):
    """Convert time into C style timeval."""
    frac, whole = math.modf(seconds_since_epoch)
    microseconds = math.floor(frac * 1000000)
    seconds = math.floor(whole)
    return seconds, microseconds

Similar but not relevant code:
def current_timestamp():
    """Returns current time as ISO8601 formatted string in the Zulu TZ"""
    now = datetime.utcnow()
    timestamp = now.isoformat()[0:19] + 'Z'

    debug("generated timestamp: {now}".format(now=timestamp))

    return timestamp

Similar but not relevant code:
def _elapsed_time(begin_time, end_time):
    """Assuming format YYYY-MM-DD hh:mm:ss

    Returns the elapsed time in seconds
    """

    bt = _str2datetime(begin_time)
    et = _str2datetime(end_time)

    return float((et - bt).seconds)

Similar but not relevant code:
def run_time() -> timedelta:
    """

    :return:
    """

    delta = start_time if start_time else datetime.utcnow()
    return datetime.utcnow() - delta

Similar but not relevant code:
def ms_to_datetime(ms):
    """
    Converts a millisecond accuracy timestamp to a datetime
    """
    dt = datetime.datetime.utcfromtimestamp(ms / 1000)
    return dt.replace(microsecond=(ms % 1000) * 1000).replace(tzinfo=pytz.utc)
Query: python change function values with slider

Relevant code:
def _update_plot(self, _):
        """Callback to redraw the plot to reflect the new parameter values."""
        # Since all sliders call this same callback without saying who they are
        # I need to update the values for all parameters. This can be
        # circumvented by creating a seperate callback function for each
        # parameter.
        for param in self.model.params:
            param.value = self._sliders[param].val
        for indep_var, dep_var in self._projections:
            self._update_specific_plot(indep_var, dep_var)

Similar but not relevant code:
def __call__(self, _):
        """Update the progressbar."""
        if self.iter % self.step == 0:
            self.pbar.update(self.step)

        self.iter += 1

Similar but not relevant code:
def p(self):
        """
        Helper property containing the percentage this slider is "filled".
        
        This property is read-only.
        """
        return (self.n-self.nmin)/max((self.nmax-self.nmin),1)

Similar but not relevant code:
def stepBy(self, steps):
        """steps value up/down by a single step. Single step is defined in singleStep().

        Args:
            steps (int): positiv int steps up, negativ steps down
        """
        self.setValue(self.value() + steps*self.singleStep())
Query: to impute missing values in python

Relevant code:
def impute_data(self,x):
        """Imputes data set containing Nan values"""
        imp = Imputer(missing_values='NaN', strategy='mean', axis=0)
        return imp.fit_transform(x)

Similar but not relevant code:
def dropna(self, subset=None):
        """Remove missing values according to Baloo's convention.

        Parameters
        ----------
        subset : list of str, optional
            Which columns to check for missing values in.

        Returns
        -------
        DataFrame
            DataFrame with no null values in columns.

        """
        subset = check_and_obtain_subset_columns(subset, self)
        not_nas = [v.notna() for v in self[subset]._iter()]
        and_filter = reduce(lambda x, y: x & y, not_nas)

        return self[and_filter]
Query: python change user to root

Relevant code:
def _is_root():
    """Checks if the user is rooted."""
    import os
    import ctypes
    try:
        return os.geteuid() == 0
    except AttributeError:
        return ctypes.windll.shell32.IsUserAnAdmin() != 0
    return False

Similar but not relevant code:
def get_user_name():
    """Get user name provide by operating system
    """

    if sys.platform == 'win32':
        #user = os.getenv('USERPROFILE')
        user = os.getenv('USERNAME')
    else:
        user = os.getenv('LOGNAME')

    return user

Similar but not relevant code:
def make_file_read_only(file_path):
    """
    Removes the write permissions for the given file for owner, groups and others.

    :param file_path: The file whose privileges are revoked.
    :raise FileNotFoundError: If the given file does not exist.
    """
    old_permissions = os.stat(file_path).st_mode
    os.chmod(file_path, old_permissions & ~WRITE_PERMISSIONS)

Similar but not relevant code:
def chmod(f):
    """ change mod to writeable """
    try:
        os.chmod(f, S_IWRITE)  # windows (cover all)
    except Exception as e:
        pass
    try:
        os.chmod(f, 0o777)  # *nix
    except Exception as e:
        pass

Similar but not relevant code:
def chmod(self, mode):
        """
        Change the mode (permissions) of this file.  The permissions are
        unix-style and identical to those used by python's C{os.chmod}
        function.

        @param mode: new permissions
        @type mode: int
        """
        self.sftp._log(DEBUG, 'chmod(%s, %r)' % (hexlify(self.handle), mode))
        attr = SFTPAttributes()
        attr.st_mode = mode
        self.sftp._request(CMD_FSETSTAT, self.handle, attr)
Query: python change variables in outer scope

Relevant code:
def scopes_as(self, new_scopes):
        """Replace my :attr:`scopes` for the duration of the with block.

        My global scope is not replaced.

        Args:
            new_scopes (list of dict-likes): The new :attr:`scopes` to use.
        """
        old_scopes, self.scopes = self.scopes, new_scopes
        yield
        self.scopes = old_scopes

Similar but not relevant code:
def vars_(self):
        """ Returns symbol instances corresponding to variables
        of the current scope.
        """
        return [x for x in self[self.current_scope].values() if x.class_ == CLASS.var]

Similar but not relevant code:
def append_scope(self):
        """Create a new scope in the current frame."""
        self.stack.current.append(Scope(self.stack.current.current))

Similar but not relevant code:
def _method_scope(input_layer, name):
  """Creates a nested set of name and id scopes and avoids repeats."""
  global _in_method_scope
  # pylint: disable=protected-access

  with input_layer.g.as_default(), \
       scopes.var_and_name_scope(
           None if _in_method_scope else input_layer._scope), \
       scopes.var_and_name_scope((name, None)) as (scope, var_scope):
    was_in_method_scope = _in_method_scope
    yield scope, var_scope
    _in_method_scope = was_in_method_scope

Similar but not relevant code:
def __init__(self, scope, parent):
        """Constructor for try block structures.

        Args:
            scope (CodeEntity): The program scope where this object belongs.
            parent (CodeEntity): This object's parent in the program tree.
        """
        CodeStatement.__init__(self, scope, parent)
        self.body = CodeBlock(scope, self, explicit=True)
        self.catches = []
        self.finally_body = CodeBlock(scope, self, explicit=True)
Query: python check array of strings

Relevant code:
def all_strings(arr):
        """
        Ensures that the argument is a list that either is empty or contains only strings
        :param arr: list
        :return:
        """
        if not isinstance([], list):
            raise TypeError("non-list value found where list is expected")
        return all(isinstance(x, str) for x in arr)

Similar but not relevant code:
def check_lengths(*arrays):
    """
    tool to ensure input and output data have the same number of samples

    Parameters
    ----------
    *arrays : iterable of arrays to be checked

    Returns
    -------
    None
    """
    lengths = [len(array) for array in arrays]
    if len(np.unique(lengths)) > 1:
        raise ValueError('Inconsistent data lengths: {}'.format(lengths))

Similar but not relevant code:
def check_consistent_length(*arrays):
    """Check that all arrays have consistent first dimensions.

    Checks whether all objects in arrays have the same shape or length.

    Parameters
    ----------
    arrays : list or tuple of input objects.
        Objects that will be checked for consistent length.
    """

    uniques = np.unique([_num_samples(X) for X in arrays if X is not None])
    if len(uniques) > 1:
        raise ValueError("Found arrays with inconsistent numbers of samples: %s"
                         % str(uniques))