benjamintli/cosqa-llm-filtered-hard-negatives huggingface dataset:
Query: returning multiple values in a queue python multiprocesss

Relevant code:
def create(parallel):
    """Create a queue based on the provided parallel arguments.

    TODO Startup/tear-down. Currently using default queue for testing
    """
    queue = {k: v for k, v in parallel.items() if k in ["queue", "cores_per_job", "mem"]}
    yield queue

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))

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()
Query: python 3 print excetion message

Relevant code:
def format_exc(limit=None):
    """Like print_exc() but return a string. Backport for Python 2.3."""
    try:
        etype, value, tb = sys.exc_info()
        return ''.join(traceback.format_exception(etype, value, tb, limit))
    finally:
        etype = value = tb = None

Similar but not relevant code:
def _show(self, message, indent=0, enable_verbose=True):  # pragma: no cover
        """Message printer.
        """
        if enable_verbose:
            print("    " * indent + message)

Similar but not relevant code:
def err(msg):
    """Pretty-print an error."""
    click.echo(click.style(msg, fg="red", bold=True))

Similar but not relevant code:
def error(*args):
    """Display error message via stderr or GUI."""
    if sys.stdin.isatty():
        print('ERROR:', *args, file=sys.stderr)
    else:
        notify_error(*args)

Similar but not relevant code:
def print_error(msg):
    """ Print an error message """
    if IS_POSIX:
        print(u"%s[ERRO] %s%s" % (ANSI_ERROR, msg, ANSI_END))
    else:
        print(u"[ERRO] %s" % (msg))
Query: python 3 replace every n instacnesstring

Relevant code:
def subn_filter(s, find, replace, count=0):
    """A non-optimal implementation of a regex filter"""
    return re.gsub(find, replace, count, s)

Similar but not relevant code:
def replace(s, replace):
    """Replace multiple values in a string"""
    for r in replace:
        s = s.replace(*r)
    return s

Similar but not relevant code:
def multi_replace(instr, search_list=[], repl_list=None):
    """
    Does a string replace with a list of search and replacements

    TODO: rename
    """
    repl_list = [''] * len(search_list) if repl_list is None else repl_list
    for ser, repl in zip(search_list, repl_list):
        instr = instr.replace(ser, repl)
    return instr

Similar but not relevant code:
def replace_all(text, dic):
    """Takes a string and dictionary. replaces all occurrences of i with j"""

    for i, j in dic.iteritems():
        text = text.replace(i, j)
    return text

Similar but not relevant code:
def myreplace(astr, thefind, thereplace):
    """in string astr replace all occurences of thefind with thereplace"""
    alist = astr.split(thefind)
    new_s = alist.split(thereplace)
    return new_s
Query: round to zero decimals python

Relevant code:
def price_rounding(price, decimals=2):
    """Takes a decimal price and rounds to a number of decimal places"""
    try:
        exponent = D('.' + decimals * '0')
    except InvalidOperation:
        # Currencies with no decimal places, ex. JPY, HUF
        exponent = D()
    return price.quantize(exponent, rounding=ROUND_UP)

Similar but not relevant code:
def round_to_int(number, precision):
    """Round a number to a precision"""
    precision = int(precision)
    rounded = (int(number) + precision / 2) // precision * precision
    return rounded

Similar but not relevant code:
def round_to_n(x, n):
    """
    Round to sig figs
    """
    return round(x, -int(np.floor(np.log10(x))) + (n - 1))
Query: python 3 xml pretty

Relevant code:
def pp_xml(body):
    """Pretty print format some XML so it's readable."""
    pretty = xml.dom.minidom.parseString(body)
    return pretty.toprettyxml(indent="  ")

Similar but not relevant code:
def xml_str_to_dict(s):
    """ Transforms an XML string it to python-zimbra dict format

    For format, see:
      https://github.com/Zimbra-Community/python-zimbra/blob/master/README.md

    :param: a string, containing XML
    :returns: a dict, with python-zimbra format
    """
    xml = minidom.parseString(s)
    return pythonzimbra.tools.xmlserializer.dom_to_dict(xml.firstChild)

Similar but not relevant code:
def xmltreefromfile(filename):
    """Internal function to read an XML file"""
    try:
        return ElementTree.parse(filename, ElementTree.XMLParser(collect_ids=False))
    except TypeError:
        return ElementTree.parse(filename, ElementTree.XMLParser())
Query: run ansible playbook from python with host pass as variable

Relevant code:
def callPlaybook(self, playbook, ansibleArgs, wait=True, tags=["all"]):
        """
        Run a playbook.

        :param playbook: An Ansible playbook to run.
        :param ansibleArgs: Arguments to pass to the playbook.
        :param wait: Wait for the play to finish if true.
        :param tags: Control tags for the play.
        """
        playbook = os.path.join(self.playbooks, playbook)  # Path to playbook being executed
        verbosity = "-vvvvv" if logger.isEnabledFor(logging.DEBUG) else "-v"
        command = ["ansible-playbook", verbosity, "--tags", ",".join(tags), "--extra-vars"]
        command.append(" ".join(["=".join(i) for i in ansibleArgs.items()]))  # Arguments being passed to playbook
        command.append(playbook)

        logger.debug("Executing Ansible call `%s`", " ".join(command))
        p = subprocess.Popen(command)
        if wait:
            p.communicate()
            if p.returncode != 0:
                # FIXME: parse error codes
                raise RuntimeError("Ansible reported an error when executing playbook %s" % playbook)

Similar but not relevant code:
def server(port):
    """Start the Django dev server."""
    args = ['python', 'manage.py', 'runserver']
    if port:
        args.append(port)
    run.main(args)

Similar but not relevant code:
def init():
    """
    Execute init tasks for all components (virtualenv, pip).
    """
    print(yellow("# Setting up environment...\n", True))
    virtualenv.init()
    virtualenv.update_requirements()
    print(green("\n# DONE.", True))
    print(green("Type ") + green("activate", True) + green(" to enable your virtual environment."))

Similar but not relevant code:
def start():
    """Starts the web server."""
    global app
    bottle.run(app, host=conf.WebHost, port=conf.WebPort,
               debug=conf.WebAutoReload, reloader=conf.WebAutoReload,
               quiet=conf.WebQuiet)

Similar but not relevant code:
def vsh(cmd, *args, **kw):
    """ Execute a command installed into the active virtualenv.
    """
    args = '" "'.join(i.replace('"', r'\"') for i in args)
    easy.sh('"%s" "%s"' % (venv_bin(cmd), args))

Similar but not relevant code:
def docker_environment(env):
    """
    Transform dictionary of environment variables into Docker -e parameters.

    >>> result = docker_environment({'param1': 'val1', 'param2': 'val2'})
    >>> result in ['-e "param1=val1" -e "param2=val2"', '-e "param2=val2" -e "param1=val1"']
    True
    """
    return ' '.join(
        ["-e \"%s=%s\"" % (key, value.replace("$", "\\$").replace("\"", "\\\"").replace("`", "\\`"))
         for key, value in env.items()])
Query: python a list of all constants

Relevant code:
def _get_all_constants():
    """
    Get list of all uppercase, non-private globals (doesn't start with ``_``).

    Returns:
        list: Uppercase names defined in `globals()` (variables from this \
              module).
    """
    return [
        key for key in globals().keys()
        if all([
            not key.startswith("_"),          # publicly accesible
            key.upper() == key,               # uppercase
            type(globals()[key]) in _ALLOWED  # and with type from _ALLOWED
        ])
    ]

Similar but not relevant code:
def types(self):
        """
        Return a list of all the variable types that exist in the
        Variables object.
        """
        output = set()
        for var in self.values():
            if var.has_value():
                output.update(var.types())
        return list(output)

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 items(self):
    """Return a list of the (name, value) pairs of the enum.

    These are returned in the order they were defined in the .proto file.
    """
    return [(value_descriptor.name, value_descriptor.number)
            for value_descriptor in self._enum_type.values]

Similar but not relevant code:
def parameter_vector(self):
        """An array of all parameters (including frozen parameters)"""
        return np.array([getattr(self, k) for k in self.parameter_names])
Query: run python file in java using runtime

Relevant code:
def test_python_java_rt():
    """ Run Python test cases against Java runtime classes. """
    sub_env = {'PYTHONPATH': _build_dir()}

    log.info('Executing Python unit tests (against Java runtime classes)...')
    return jpyutil._execute_python_scripts(python_java_rt_tests,
                                           env=sub_env)

Similar but not relevant code:
def stub_main():
    """setuptools blah: it still can't run a module as a script entry_point"""
    from google.apputils import run_script_module
    import butcher.main
    run_script_module.RunScriptModule(butcher.main)

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 execfile(fname, variables):
    """ This is builtin in python2, but we have to roll our own on py3. """
    with open(fname) as f:
        code = compile(f.read(), fname, 'exec')
        exec(code, variables)

Similar but not relevant code:
def code(self):
    """Returns the code object for this BUILD file."""
    return compile(self.source(), self.full_path, 'exec', flags=0, dont_inherit=True)

Similar but not relevant code:
def generate(env):
    """Add Builders and construction variables for SGI MIPS C++ to an Environment."""

    cplusplus.generate(env)

    env['CXX']         = 'CC'
    env['CXXFLAGS']    = SCons.Util.CLVar('-LANG:std')
    env['SHCXX']       = '$CXX'
    env['SHOBJSUFFIX'] = '.o'
    env['STATIC_AND_SHARED_OBJECTS_ARE_THE_SAME'] = 1
Query: running python by code block on spyder

Relevant code:
def test():        
    """Local test."""
    from spyder.utils.qthelpers import qapplication
    app = qapplication()
    dlg = ProjectDialog(None)
    dlg.show()
    sys.exit(app.exec_())

Similar but not relevant code:
def start(self):
        """Start the receiver.
        """
        if not self._is_running:
            self._do_run = True
            self._thread.start()
        return self

Similar but not relevant code:
def stub_main():
    """setuptools blah: it still can't run a module as a script entry_point"""
    from google.apputils import run_script_module
    import butcher.main
    run_script_module.RunScriptModule(butcher.main)

Similar but not relevant code:
async def _thread_coro(self, *args):
        """ Coroutine called by MapAsync. It's wrapping the call of
        run_in_executor to run the synchronous function as thread """
        return await self._loop.run_in_executor(
            self._executor, self._function, *args)

Similar but not relevant code:
def asynchronous(function, event):
    """
    Runs the function asynchronously taking care of exceptions.
    """
    thread = Thread(target=synchronous, args=(function, event))
    thread.daemon = True
    thread.start()

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 active window title

Relevant code:
def title(self):
        """ The title of this window """
        with switch_window(self._browser, self.name):
            return self._browser.title

Similar but not relevant code:
def get_active_window(self):
        """
        The current active :class:`.Window`.
        """
        app = get_app()

        try:
            return self._active_window_for_cli[app]
        except KeyError:
            self._active_window_for_cli[app] = self._last_active_window or self.windows[0]
            return self.windows[0]

Similar but not relevant code:
def inFocus(self):
        """Set GUI on-top flag"""
        previous_flags = self.window.flags()
        self.window.setFlags(previous_flags |
                             QtCore.Qt.WindowStaysOnTopHint)

Similar but not relevant code:
def getFileDialogTitle(msg, title):
    """
    Create nicely-formatted string based on arguments msg and title
    :param msg: the msg to be displayed
    :param title: the window title
    :return: None
    """
    if msg and title:
        return "%s - %s" % (title, msg)
    if msg and not title:
        return str(msg)
    if title and not msg:
        return str(title)
    return None
Query: python add a colormap width

Relevant code:
def add_matplotlib_cmap(cm, name=None):
    """Add a matplotlib colormap."""
    global cmaps
    cmap = matplotlib_to_ginga_cmap(cm, name=name)
    cmaps[cmap.name] = cmap

Similar but not relevant code:
def _xls2col_widths(self, worksheet, tab):
        """Updates col_widths in code_array"""

        for col in xrange(worksheet.ncols):
            try:
                xls_width = worksheet.colinfo_map[col].width
                pys_width = self.xls_width2pys_width(xls_width)
                self.code_array.col_widths[col, tab] = pys_width

            except KeyError:
                pass

Similar but not relevant code:
def _column_resized(self, col, old_width, new_width):
        """Update the column width."""
        self.dataTable.setColumnWidth(col, new_width)
        self._update_layout()

Similar but not relevant code:
def palettebar(height, length, colormap):
    """Return the channels of a palettebar.
    """
    cbar = np.tile(np.arange(length) * 1.0 / (length - 1), (height, 1))
    cbar = (cbar * (colormap.values.max() + 1 - colormap.values.min())
            + colormap.values.min())

    return colormap.palettize(cbar)
Query: save http get to file python

Relevant code:
def download_file(save_path, file_url):
    """ Download file from http url link """

    r = requests.get(file_url)  # create HTTP response object

    with open(save_path, 'wb') as f:
        f.write(r.content)

    return save_path

Similar but not relevant code:
def rest_put_stream(self, url, stream, headers=None, session=None, verify=True, cert=None):
        """
        Perform a chunked PUT request to url with requests.session
        This is specifically to upload files.
        """
        res = session.put(url, headers=headers, data=stream, verify=verify, cert=cert)
        return res.text, res.status_code

Similar but not relevant code:
def write(url, content, **args):
    """Put an object into a ftps URL."""
    with FTPSResource(url, **args) as resource:
        resource.write(content)

Similar but not relevant code:
def write(self):
        """Write content back to file."""
        with open(self.path, 'w') as file_:
            file_.write(self.content)
Query: save python graphviz as jpeg

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 fig2x(figure, format):
    """Returns svg from matplotlib chart"""

    # Save svg to file like object svg_io
    io = StringIO()
    figure.savefig(io, format=format)

    # Rewind the file like object
    io.seek(0)

    data = io.getvalue()
    io.close()

    return data

Similar but not relevant code:
def create_movie(fig, update_figure, filename, title, fps=15, dpi=100):
    """Helps us to create a movie."""
    FFMpegWriter = manimation.writers['ffmpeg']
    metadata     = dict(title=title)
    writer       = FFMpegWriter(fps=fps, metadata=metadata)

    with writer.saving(fig, filename, dpi):
        t = 0
        while True:
            if update_figure(t):
                writer.grab_frame()
                t += 1
            else:
                break

Similar but not relevant code:
def save_as_png(self, filename, width=300, height=250, render_time=1):
        """Open saved html file in an virtual browser and save a screen shot to PNG format."""
        self.driver.set_window_size(width, height)
        self.driver.get('file://{path}/{filename}'.format(
            path=os.getcwd(), filename=filename + ".html"))
        time.sleep(render_time)
        self.driver.save_screenshot(filename + ".png")
Query: python add path to modle

Relevant code:
def _load_mod_ui_libraries(self, path):
        """
        :param Path path:
        """
        path = path / Path('mod')
        sys.path.append(str(path))

Similar but not relevant code:
def replacing_symlink(source, link_name):
    """Create symlink that overwrites any existing target.
    """
    with make_tmp_name(link_name) as tmp_link_name:
        os.symlink(source, tmp_link_name)
        replace_file_or_dir(link_name, tmp_link_name)

Similar but not relevant code:
def _extension(modpath: str) -> setuptools.Extension:
    """Make setuptools.Extension."""
    return setuptools.Extension(modpath, [modpath.replace(".", "/") + ".py"])

Similar but not relevant code:
def fixpath(path):
    """Uniformly format a path."""
    return os.path.normpath(os.path.realpath(os.path.expanduser(path)))

Similar but not relevant code:
def symlink(source, destination):
    """Create a symbolic link"""
    log("Symlinking {} as {}".format(source, destination))
    cmd = [
        'ln',
        '-sf',
        source,
        destination,
    ]
    subprocess.check_call(cmd)
Query: scala equivileant of python dictionary

Relevant code:
def get_python_dict(scala_map):
    """Return a dict from entries in a scala.collection.immutable.Map"""
    python_dict = {}
    keys = get_python_list(scala_map.keys().toList())
    for key in keys:
        python_dict[key] = scala_map.apply(key)
    return python_dict

Similar but not relevant code:
def conv_dict(self):
        """dictionary of conversion"""
        return dict(integer=self.integer, real=self.real, no_type=self.no_type)

Similar but not relevant code:
def inverted_dict(d):
    """Return a dict with swapped keys and values

    >>> inverted_dict({0: ('a', 'b'), 1: 'cd'}) == {'cd': 1, ('a', 'b'): 0}
    True
    """
    return dict((force_hashable(v), k) for (k, v) in viewitems(dict(d)))

Similar but not relevant code:
def from_dict(cls, d):
        """Create an instance from a dictionary."""
        return cls(**{k: v for k, v in d.items() if k in cls.ENTRIES})

Similar but not relevant code:
def to_dict(dictish):
    """
    Given something that closely resembles a dictionary, we attempt
    to coerce it into a propery dictionary.
    """
    if hasattr(dictish, 'iterkeys'):
        m = dictish.iterkeys
    elif hasattr(dictish, 'keys'):
        m = dictish.keys
    else:
        raise ValueError(dictish)

    return dict((k, dictish[k]) for k in m())
Query: python adjacency matrix dict

Relevant code:
def get_adjacent_matrix(self):
        """Get adjacency matrix.

        Returns:
            :param adj: adjacency matrix
            :type adj: np.ndarray
        """
        edges = self.edges
        num_edges = len(edges) + 1
        adj = np.zeros([num_edges, num_edges])

        for k in range(num_edges - 1):
            adj[edges[k].L, edges[k].R] = 1
            adj[edges[k].R, edges[k].L] = 1

        return adj

Similar but not relevant code:
def to_bipartite_matrix(A):
    """Returns the adjacency matrix of a bipartite graph whose biadjacency
    matrix is `A`.

    `A` must be a NumPy array.

    If `A` has **m** rows and **n** columns, then the returned matrix has **m +
    n** rows and columns.

    """
    m, n = A.shape
    return four_blocks(zeros(m, m), A, A.T, zeros(n, n))

Similar but not relevant code:
def get_edge_relations(graph: BELGraph) -> Mapping[Tuple[BaseEntity, BaseEntity], Set[str]]:
    """Build a dictionary of {node pair: set of edge types}."""
    return group_dict_set(
        ((u, v), d[RELATION])
        for u, v, d in graph.edges(data=True)
    )

Similar but not relevant code:
def get_triangles(graph: DiGraph) -> SetOfNodeTriples:
    """Get a set of triples representing the 3-cycles from a directional graph.

    Each 3-cycle is returned once, with nodes in sorted order.
    """
    return {
        tuple(sorted([a, b, c], key=str))
        for a, b in graph.edges()
        for c in graph.successors(b)
        if graph.has_edge(c, a)
    }

Similar but not relevant code:
def to_networkx(graph):
    """ Convert a Mapper 1-complex to a networkx graph.

    Parameters
    -----------

    graph: dictionary, graph object returned from `kmapper.map`

    Returns
    --------

    g: graph as networkx.Graph() object

    """

    # import here so networkx is not always required.
    import networkx as nx

    nodes = graph["nodes"].keys()
    edges = [[start, end] for start, ends in graph["links"].items() for end in ends]

    g = nx.Graph()
    g.add_nodes_from(nodes)
    nx.set_node_attributes(g, dict(graph["nodes"]), "membership")

    g.add_edges_from(edges)

    return g