benjamintli/cosqa-llm-filtered-hard-negatives huggingface dataset:
Query: python get the size of a type

Relevant code:
def size(dtype):
  """Returns the number of bytes to represent this `dtype`."""
  dtype = tf.as_dtype(dtype)
  if hasattr(dtype, 'size'):
    return dtype.size
  return np.dtype(dtype).itemsize

Similar but not relevant code:
def codes_get_size(handle, key):
    # type: (cffi.FFI.CData, str) -> int
    """
    Get the number of coded value from a key.
    If several keys of the same name are present, the total sum is returned.

    :param bytes key: the keyword to get the size of

    :rtype: int
    """
    size = ffi.new('size_t *')
    _codes_get_size(handle, key.encode(ENC), size)
    return size[0]

Similar but not relevant code:
def memsize(self):
        """ Total array cell + indexes size
        """
        return self.size + 1 + TYPE.size(gl.BOUND_TYPE) * len(self.bounds)

Similar but not relevant code:
def get_type_len(self):
        """Retrieve the type and length for a data record."""
        # Check types and set type/len
        self.get_sql()
        return self.type, self.len, self.len_decimal
Query: dot product of vectors in python

Relevant code:
def dot_v3(v, w):
    """Return the dotproduct of two vectors."""

    return sum([x * y for x, y in zip(v, w)])

Similar but not relevant code:
def cross_product_matrix(vec):
    """Returns a 3x3 cross-product matrix from a 3-element vector."""
    return np.array([[0, -vec[2], vec[1]],
                     [vec[2], 0, -vec[0]],
                     [-vec[1], vec[0], 0]])
Query: python get the text on the page

Relevant code:
def get_page_text(self, page):
        """
        Downloads and returns the full text of a particular page
        in the document.
        """
        url = self.get_page_text_url(page)
        return self._get_url(url)

Similar but not relevant code:
def _get_minidom_tag_value(station, tag_name):
    """get a value from a tag (if it exists)"""
    tag = station.getElementsByTagName(tag_name)[0].firstChild
    if tag:
        return tag.nodeValue

    return None
Query: python get windows user id

Relevant code:
def get_nt_system_uid():
    """Get the MachineGuid from
    HKEY_LOCAL_MACHINE\Software\Microsoft\Cryptography\MachineGuid
    """
    try:
        import _winreg as winreg
    except ImportError:
        import winreg
    lm = winreg.ConnectRegistry(None, winreg.HKEY_LOCAL_MACHINE)
    try:
        key = winreg.OpenKey(lm, r"Software\Microsoft\Cryptography")
        try:
            return winreg.QueryValueEx(key, "MachineGuid")[0]
        finally:
            key.Close()
    finally:
        lm.Close()

Similar but not relevant code:
def setAsApplication(myappid):
    """
    Tells Windows this is an independent application with an unique icon on task bar.

    id is an unique string to identify this application, like: 'mycompany.myproduct.subproduct.version'
    """

    if os.name == 'nt':
        import ctypes
        ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID(myappid)

Similar but not relevant code:
def get_user_id_from_email(self, email):
        """ Uses the get-all-user-accounts Portals API to retrieve the
        user-id by supplying an email. """
        accts = self.get_all_user_accounts()

        for acct in accts:
            if acct['email'] == email:
                return acct['id']
        return None
Query: python gevent pool repeat

Relevant code:
def fetch_event(urls):
    """
    This parallel fetcher uses gevent one uses gevent
    """
    rs = (grequests.get(u) for u in urls)
    return [content.json() for content in grequests.map(rs)]

Similar but not relevant code:
def start(self):
        """
        Starts the loop. Calling a running loop is an error.
        """
        assert not self.has_started(), "called start() on an active GeventLoop"
        self._stop_event = Event()
        # note that we don't use safe_greenlets.spawn because we take care of it in _loop by ourselves
        self._greenlet = gevent.spawn(self._loop)

Similar but not relevant code:
def compute_capture(args):
    x, y, w, h, params = args
    """Callable function for the multiprocessing pool."""
    return x, y, mandelbrot_capture(x, y, w, h, params)

Similar but not relevant code:
def parallel(processes, threads):
    """
    execute jobs in processes using N threads
    """
    pool = multithread(threads)
    pool.map(run_process, processes)
    pool.close()
    pool.join()

Similar but not relevant code:
def compute(args):
    x, y, params = args
    """Callable function for the multiprocessing pool."""
    return x, y, mandelbrot(x, y, params)

Similar but not relevant code:
def multiprocess_mapping(func, iterable):
    """Multiprocess mapping the given function on the given iterable.

    This only works in Linux and Mac systems since Windows has no forking capability. On Windows we fall back on
    single processing. Also, if we reach memory limits we fall back on single cpu processing.

    Args:
        func (func): the function to apply
        iterable (iterable): the iterable with the elements we want to apply the function on
    """
    if os.name == 'nt':  # In Windows there is no fork.
        return list(map(func, iterable))
    try:
        p = multiprocessing.Pool()
        return_data = list(p.imap(func, iterable))
        p.close()
        p.join()
        return return_data
    except OSError:
        return list(map(func, iterable))
Query: python gevent several group task

Relevant code:
def fetch_event(urls):
    """
    This parallel fetcher uses gevent one uses gevent
    """
    rs = (grequests.get(u) for u in urls)
    return [content.json() for content in grequests.map(rs)]

Similar but not relevant code:
def start(self):
        """
        Starts the loop. Calling a running loop is an error.
        """
        assert not self.has_started(), "called start() on an active GeventLoop"
        self._stop_event = Event()
        # note that we don't use safe_greenlets.spawn because we take care of it in _loop by ourselves
        self._greenlet = gevent.spawn(self._loop)

Similar but not relevant code:
def gevent_monkey_patch_report(self):
        """
        Report effective gevent monkey patching on the logs.
        """
        try:
            import gevent.socket
            import socket

            if gevent.socket.socket is socket.socket:
                self.log("gevent monkey patching is active")
                return True
            else:
                self.notify_user("gevent monkey patching failed.")
        except ImportError:
            self.notify_user("gevent is not installed, monkey patching failed.")
        return False

Similar but not relevant code:
def run(self, *args, **kwargs):
        """ Connect and run bot in event loop. """
        self.eventloop.run_until_complete(self.connect(*args, **kwargs))
        try:
            self.eventloop.run_forever()
        finally:
            self.eventloop.stop()
Query: epoch to gmt datetime python 3

Relevant code:
def _dt_to_epoch(dt):
        """Convert datetime to epoch seconds."""
        try:
            epoch = dt.timestamp()
        except AttributeError:  # py2
            epoch = (dt - datetime(1970, 1, 1)).total_seconds()
        return epoch

Similar but not relevant code:
def excel_datetime(timestamp, epoch=None):
    """Return datetime object from timestamp in Excel serial format.

    Convert LSM time stamps.

    >>> excel_datetime(40237.029999999795)
    datetime.datetime(2010, 2, 28, 0, 43, 11, 999982)

    """
    if epoch is None:
        epoch = datetime.datetime.fromordinal(693594)
    return epoch + datetime.timedelta(timestamp)

Similar but not relevant code:
def after_epoch(self, **_) -> None:
        """Save/override the latest model after every epoch."""
        SaveEvery.save_model(model=self._model, name_suffix=self._OUTPUT_NAME, on_failure=self._on_save_failure)
Query: python go to parent directory

Relevant code:
def go_to_parent_directory(self):
        """Go to parent directory"""
        self.chdir(osp.abspath(osp.join(getcwd_or_home(), os.pardir)))

Similar but not relevant code:
def root_parent(self, category=None):
        """ Returns the topmost parent of the current category. """
        return next(filter(lambda c: c.is_root, self.hierarchy()))

Similar but not relevant code:
def in_directory(path):
    """Context manager (with statement) that changes the current directory
    during the context.
    """
    curdir = os.path.abspath(os.curdir)
    os.chdir(path)
    yield
    os.chdir(curdir)
Query: epython name is notdefined

Relevant code:
def page_guiref(arg_s=None):
    """Show a basic reference about the GUI Console."""
    from IPython.core import page
    page.page(gui_reference, auto_html=True)

Similar but not relevant code:
def method_header(method_name, nogil=False, idx_as_arg=False):
    """Returns the Cython method header for methods without arguments except
    `self`."""
    if not config.FASTCYTHON:
        nogil = False
    header = 'cpdef inline void %s(self' % method_name
    header += ', int idx)' if idx_as_arg else ')'
    header += ' nogil:' if nogil else ':'
    return header

Similar but not relevant code:
def IPYTHON_MAIN():
    """Decide if the Ipython command line is running code."""
    import pkg_resources

    runner_frame = inspect.getouterframes(inspect.currentframe())[-2]
    return (
        getattr(runner_frame, "function", None)
        == pkg_resources.load_entry_point("ipython", "console_scripts", "ipython").__name__
    )

Similar but not relevant code:
def show_intro(self):
        """Show intro to IPython help"""
        from IPython.core.usage import interactive_usage
        self.main.help.show_rich_text(interactive_usage)

Similar but not relevant code:
def _show_traceback(method):
    """decorator for showing tracebacks in IPython"""
    def m(self, *args, **kwargs):
        try:
            return(method(self, *args, **kwargs))
        except Exception as e:
            ip = get_ipython()
            if ip is None:
                self.log.warning("Exception in widget method %s: %s", method, e, exc_info=True)
            else:
                ip.showtraceback()
    return m

Similar but not relevant code:
def debug_on_error(type, value, tb):
    """Code due to Thomas Heller - published in Python Cookbook (O'Reilley)"""
    traceback.print_exc(type, value, tb)
    print()
    pdb.pm()
Query: equivalent of throw in python

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 on_error(e):  # pragma: no cover
    """Error handler

    RuntimeError or ValueError exceptions raised by commands will be handled
    by this function.
    """
    exname = {'RuntimeError': 'Runtime error', 'Value Error': 'Value error'}
    sys.stderr.write('{}: {}\n'.format(exname[e.__class__.__name__], str(e)))
    sys.stderr.write('See file slam_error.log for additional details.\n')
    sys.exit(1)

Similar but not relevant code:
def assert_raises(ex_type, func, *args, **kwargs):
    r"""
    Checks that a function raises an error when given specific arguments.

    Args:
        ex_type (Exception): exception type
        func (callable): live python function

    CommandLine:
        python -m utool.util_assert assert_raises --show

    Example:
        >>> # ENABLE_DOCTEST
        >>> from utool.util_assert import *  # NOQA
        >>> import utool as ut
        >>> ex_type = AssertionError
        >>> func = len
        >>> # Check that this raises an error when something else does not
        >>> assert_raises(ex_type, assert_raises, ex_type, func, [])
        >>> # Check this does not raise an error when something else does
        >>> assert_raises(ValueError, [].index, 0)
    """
    try:
        func(*args, **kwargs)
    except Exception as ex:
        assert isinstance(ex, ex_type), (
            'Raised %r but type should have been %r' % (ex, ex_type))
        return True
    else:
        raise AssertionError('No error was raised')

Similar but not relevant code:
def make_exception_message(exc):
    """
    An exception is passed in and this function
    returns the proper string depending on the result
    so it is readable enough.
    """
    if str(exc):
        return '%s: %s\n' % (exc.__class__.__name__, exc)
    else:
        return '%s\n' % (exc.__class__.__name__)
Query: python graphviz executables not found

Relevant code:
def _check_graphviz_available(output_format):
    """check if we need graphviz for different output format"""
    try:
        subprocess.call(["dot", "-V"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    except OSError:
        print(
            "The output format '%s' is currently not available.\n"
            "Please install 'Graphviz' to have other output formats "
            "than 'dot' or 'vcg'." % output_format
        )
        sys.exit(32)

Similar but not relevant code:
def _text_to_graphiz(self, text):
        """create a graphviz graph from text"""
        dot = Source(text, format='svg')
        return dot.pipe().decode('utf-8')

Similar but not relevant code:
def setup_detect_python2():
        """
        Call this before using the refactoring tools to create them on demand
        if needed.
        """
        if None in [RTs._rt_py2_detect, RTs._rtp_py2_detect]:
            RTs._rt_py2_detect = RefactoringTool(py2_detect_fixers)
            RTs._rtp_py2_detect = RefactoringTool(py2_detect_fixers,
                                                  {'print_function': True})

Similar but not relevant code:
def to_graphviz(graph):
    """

    :param graph:
    :return:
    """
    ret = ['digraph g {']
    vertices = []

    node_ids = dict([(name, 'node' + idx) for (idx, name) in enumerate(list(graph))])

    for node in list(graph):
        ret.append('  "%s" [label="%s"];' % (node_ids[node], node))
        for target in graph[node]:
            vertices.append('  "%s" -> "%s";' % (node_ids[node], node_ids[target]))

    ret += vertices
    ret.append('}')
    return '\n'.join(ret)
Query: escape string for sql python

Relevant code:
def _escape(s):
    """ Helper method that escapes parameters to a SQL query. """
    e = s
    e = e.replace('\\', '\\\\')
    e = e.replace('\n', '\\n')
    e = e.replace('\r', '\\r')
    e = e.replace("'", "\\'")
    e = e.replace('"', '\\"')
    return e

Similar but not relevant code:
def decode_mysql_string_literal(text):
    """
    Removes quotes and decodes escape sequences from given MySQL string literal
    returning the result.

    :param text: MySQL string literal, with the quotes still included.
    :type text: str

    :return: Given string literal with quotes removed and escape sequences
             decoded.
    :rtype: str
    """
    assert text.startswith("'")
    assert text.endswith("'")

    # Ditch quotes from the string literal.
    text = text[1:-1]

    return MYSQL_STRING_ESCAPE_SEQUENCE_PATTERN.sub(
        unescape_single_character,
        text,
    )

Similar but not relevant code:
def atlasdb_format_query( query, values ):
    """
    Turn a query into a string for printing.
    Useful for debugging.
    """
    return "".join( ["%s %s" % (frag, "'%s'" % val if type(val) in [str, unicode] else val) for (frag, val) in zip(query.split("?"), values + ("",))] )

Similar but not relevant code:
def sql(self, sql: str, *qmark_params, **named_params):
        """
        :deprecated: use self.statement to execute properly-formatted sql statements
        """
        statement = SingleSqlStatement(sql)
        return self.statement(statement).execute(*qmark_params, **named_params)

Similar but not relevant code:
def get_sql(query):
    """ Returns the sql query """
    sql = str(query.statement.compile(dialect=sqlite.dialect(),
                                      compile_kwargs={"literal_binds": True}))
    return sql

Similar but not relevant code:
def tsv_escape(x: Any) -> str:
    """
    Escape data for tab-separated value (TSV) format.
    """
    if x is None:
        return ""
    x = str(x)
    return x.replace("\t", "\\t").replace("\n", "\\n")
Query: python how do i inspect code of builtin

Relevant code:
def activate(self):
        """Store ipython references in the __builtin__ namespace."""

        add_builtin = self.add_builtin
        for name, func in self.auto_builtins.iteritems():
            add_builtin(name, func)

Similar but not relevant code:
def get_code(module):
    """
    Compile and return a Module's code object.
    """
    fp = open(module.path)
    try:
        return compile(fp.read(), str(module.name), 'exec')
    finally:
        fp.close()

Similar but not relevant code:
def _import(module, cls):
    """
    A messy way to import library-specific classes.
    TODO: I should really make a factory class or something, but I'm lazy.
    Plus, factories remind me a lot of java...
    """
    global Scanner

    try:
        cls = str(cls)
        mod = __import__(str(module), globals(), locals(), [cls], 1)
        Scanner = getattr(mod, cls)
    except ImportError:
        pass

Similar but not relevant code:
def method_header(method_name, nogil=False, idx_as_arg=False):
    """Returns the Cython method header for methods without arguments except
    `self`."""
    if not config.FASTCYTHON:
        nogil = False
    header = 'cpdef inline void %s(self' % method_name
    header += ', int idx)' if idx_as_arg else ')'
    header += ' nogil:' if nogil else ':'
    return header

Similar but not relevant code:
def debug_src(src, pm=False, globs=None):
    """Debug a single doctest docstring, in argument `src`'"""
    testsrc = script_from_examples(src)
    debug_script(testsrc, pm, globs)
Query: python how do i save cookies between each requests

Relevant code:
def save_config_value(request, response, key, value):
    """Sets value of key `key` to `value` in both session and cookies."""
    request.session[key] = value
    response.set_cookie(key, value, expires=one_year_from_now())
    return response

Similar but not relevant code:
def cookies(self) -> Dict[str, str]:
        """The parsed cookies attached to this request."""
        cookies = SimpleCookie()
        cookies.load(self.headers.get('Cookie', ''))
        return {key: cookie.value for key, cookie in cookies.items()}

Similar but not relevant code:
def parse_cookies(self, req, name, field):
        """Pull the value from the cookiejar."""
        return core.get_value(req.COOKIES, name, field)
Query: python how long does it take to check if two sets are equal

Relevant code:
def sets_are_rooted_compat(one_set, other):
    """treats the 2 sets are sets of taxon IDs on the same (unstated)
    universe of taxon ids.
    Returns True clades implied by each are compatible and False otherwise
    """
    if one_set.issubset(other) or other.issubset(one_set):
        return True
    return not intersection_not_empty(one_set, other)

Similar but not relevant code:
def allsame(list_, strict=True):
    """
    checks to see if list is equal everywhere

    Args:
        list_ (list):

    Returns:
        True if all items in the list are equal
    """
    if len(list_) == 0:
        return True
    first_item = list_[0]
    return list_all_eq_to(list_, first_item, strict)

Similar but not relevant code:
def same(*values):
    """
    Check if all values in a sequence are equal.

    Returns True on empty sequences.

    Examples
    --------
    >>> same(1, 1, 1, 1)
    True
    >>> same(1, 2, 1)
    False
    >>> same()
    True
    """
    if not values:
        return True
    first, rest = values[0], values[1:]
    return all(value == first for value in rest)

Similar but not relevant code:
def issubset(self, other):
        """
        Report whether another set contains this set.

        Example:
            >>> OrderedSet([1, 2, 3]).issubset({1, 2})
            False
            >>> OrderedSet([1, 2, 3]).issubset({1, 2, 3, 4})
            True
            >>> OrderedSet([1, 2, 3]).issubset({1, 4, 3, 5})
            False
        """
        if len(self) > len(other):  # Fast check for obvious cases
            return False
        return all(item in other for item in self)

Similar but not relevant code:
def issuperset(self, other):
        """Report whether this RangeSet contains another set."""
        self._binary_sanity_check(other)
        return set.issuperset(self, other)

Similar but not relevant code:
def __eq__(self, anotherset):
        """Tests on itemlist equality"""
        if not isinstance(anotherset, LR0ItemSet):
            raise TypeError
        if len(self.itemlist) != len(anotherset.itemlist):
            return False
        for element in self.itemlist:
            if element not in anotherset.itemlist:
                return False
        return True
Query: expanding addresses with python

Relevant code:
def ip_address_list(ips):
    """ IP address range validation and expansion. """
    # first, try it as a single IP address
    try:
        return ip_address(ips)
    except ValueError:
        pass
    # then, consider it as an ipaddress.IPv[4|6]Network instance and expand it
    return list(ipaddress.ip_network(u(ips)).hosts())

Similar but not relevant code:
def get_ips(self, instance_id):
        """Retrieves all IP addresses associated to a given instance.

        :return: tuple (IPs)
        """
        instance = self._load_instance(instance_id)
        IPs = sum(instance.networks.values(), [])
        return IPs

Similar but not relevant code:
def get_ip_address(ifname):
    """ Hack to get IP address from the interface """
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

    return socket.inet_ntoa(fcntl.ioctl(
        s.fileno(),
        0x8915,  # SIOCGIFADDR
        struct.pack('256s', ifname[:15])
    )[20:24])

Similar but not relevant code:
def _get_ipv6_from_binary(self, bin_addr):
        """Converts binary address to Ipv6 format."""

        hi = bin_addr >> 64
        lo = bin_addr & 0xFFFFFFFF
        return socket.inet_ntop(socket.AF_INET6, struct.pack("!QQ", hi, lo))

Similar but not relevant code:
def _get_binary_from_ipv4(self, ip_addr):
        """Converts IPv4 address to binary form."""

        return struct.unpack("!L", socket.inet_pton(socket.AF_INET,
                                                    ip_addr))[0]

Similar but not relevant code:
def is_valid_ipv6(ip_str):
    """
    Check the validity of an IPv6 address
    """
    try:
        socket.inet_pton(socket.AF_INET6, ip_str)
    except socket.error:
        return False
    return True