heapq with custom compare predicate
Asked Answered
L

9

148

I am trying to build a heap with a custom sort predicate. Since the values going into it are of "user-defined" type, I cannot modify their built-in comparison predicate.

Is there a way to do something like:

h = heapq.heapify([...], key=my_lt_pred)
h = heapq.heappush(h, key=my_lt_pred)

Or even better, I could wrap the heapq functions in my own container so I don't need to keep passing the predicate.

Lester answered 16/1, 2012 at 4:16 Comment(2)
possible duplicate of https://mcmap.net/q/82557/-min-heap-in-pythonDipody
possible duplicate of How to make heapq evaluate the heap off of a specific attribute?Mexico
P
168

According to the heapq documentation, the way to customize the heap order is to have each element on the heap to be a tuple, with the first tuple element being one that accepts normal Python comparisons.

The functions in the heapq module are a bit cumbersome (since they are not object-oriented), and always require our heap object (a heapified list) to be explicitly passed as the first parameter. We can kill two birds with one stone by creating a very simple wrapper class that will allow us to specify a key function, and present the heap as an object.

The class below keeps an internal list, where each element is a tuple, the first member of which is a key, calculated at element insertion time using the key parameter, passed at Heap instantiation:

# -*- coding: utf-8 -*-
import heapq

class MyHeap(object):
    def __init__(self, initial=None, key=lambda x:x):
        self.key = key
        self.index = 0
        if initial:
            self._data = [(key(item), i, item) for i, item in enumerate(initial)]
            self.index = len(self._data)
            heapq.heapify(self._data)
        else:
            self._data = []

    def push(self, item):
        heapq.heappush(self._data, (self.key(item), self.index, item))
        self.index += 1

    def pop(self):
        return heapq.heappop(self._data)[2]

(The extra self.index part is to avoid clashes when the evaluated key value is a draw and the stored value is not directly comparable - otherwise heapq could fail with TypeError)

Provencher answered 16/1, 2012 at 4:38 Comment(10)
Very nice! You could even go further and use triples (self.key(item), id, item), where id could be an integer handled as a class attribute, and incremented after each push. That way, you avoid the exception raised when key(item1) = key(item2). Because keys would be unique.Tshombe
I actually tried to push this (or something based on this) into Python's stdlib,and the suggestion got declined.Provencher
pity, fits the object-oriented style of most Python features, and the key argument provides extra flexibility.Tshombe
I have used list instead of tuple for e.g. [self.key(item), id, item] and it works just fine as long as first index is key.Hannahhannan
@DeepakYadav things might go messy, as lists are mutable, and tuples are not. If you'll change the priority (the [0]) the heap will be broken.Debbi
@Provencher sorry for a stupid question but I'm new to python and looking at your example, I don't see how the custom predicate is defined. Shouldn't there be some function that defines a strict weak less than ordering? I assume self.key(item) is used to for the predicate is this is where you can define custom behavior but you just returned the value as is in your example, correct? But I don't see how to consume this when you need a custom ordering, like for example if you want to order based on both values of pair you'll need to have a comparison of two pairs and return a true/false for <.Tapes
The key argument should be a callable object (ex. a function), that taking the main item as input, yields another object that is directly comparable by Python with > and <= in the desired order. The idea is the same of the key parameter on the built-in sorted method - docs.python.org/3/library/functions.html#sortedProvencher
This would fail if elements are not comparable and there are ties in key values. I'd put id(item) as a middle element of the tuple to break ties.Solatium
@GeorgiYanchev - I added that now. Thanks for the feedback.Provencher
I think the more specified way to do this is to use docs.python.org/3.6/library/…Tummy
T
161

Define a class, in which override the __lt__() function. See example below (works in Python 3.7):

import heapq

class Node(object):
    def __init__(self, val: int):
        self.val = val

    def __repr__(self):
        return f'Node value: {self.val}'

    def __lt__(self, other):
        return self.val < other.val

heap = [Node(2), Node(0), Node(1), Node(4), Node(2)]
heapq.heapify(heap)
print(heap)  # output: [Node value: 0, Node value: 2, Node value: 1, Node value: 4, Node value: 2]

heapq.heappop(heap)
print(heap)  # output: [Node value: 1, Node value: 2, Node value: 2, Node value: 4]

Thersathersites answered 28/1, 2020 at 19:58 Comment(6)
This seems like the cleanest solution by far!Gazehound
Absolutely agree with the previous two comments. This seems to be a better, cleaner solution for Python 3.Inrush
Also, here is very similar solution to a similar question: #2501957Inrush
I tested this using __gt__ instead and it works as well. Why doesn't it matter which magic method we use? I can't find anything in heapq's documentation. Maybe it's related to how Python does comparisons in general?Birthwort
When doing a comparison in heapq, Python looks for __lt__() first. If it is not defined, it will look for __gt__(). If neither is defined, it throws TypeError: '<' not supported between instances of 'Node' and 'Node'. This can be confirmed by defining both __lt__() and __gt__(), placing a print statement in each, and having __lt__() return NotImplemented.Thersathersites
To make this solution a complete one, there needs to be a tie breaker. In order to break the tie when "self.val == other.val" in the "lt" function, one option is to introduce an other field (priority or something that is pertinent to your business domain) into Node class, so that we could compare this field and make sure there are not equal values regarding this field.Tantalate
E
29

The heapq documentation suggests that heap elements could be tuples in which the first element is the priority and defines the sort order.

More pertinent to your question, however, is that the documentation includes a discussion with sample code of how one could implement their own heapq wrapper functions to deal with the problems of sort stability and elements with equal priority (among other issues).

In a nutshell, their solution is to have each element in the heapq be a triple with the priority, an entry count and the element to be inserted. The entry count ensures that elements with the same priority a sorted in the order they were added to the heapq.

Extravascular answered 16/1, 2012 at 4:37 Comment(3)
This is the correct solution, both heappush and heappushpop works directly with tuplesCrean
this solution is clean but can not cover all custom algorithm, for example, a max heap of string.Blasted
i really don't understand how people find submitting an entry count a clean solution. gods of python hear my prayers and make a normal priority queue class. thanks!Transept
K
27
setattr(ListNode, "__lt__", lambda self, other: self.val <= other.val)

Use this for comparing values of objects in heapq

Krill answered 7/7, 2020 at 11:21 Comment(4)
Interesting way to avoid redefining/re-encapsulating the object!Isothere
Thanks! this is exactly what I am looking forDeracinate
Though this may work for Leetcode, this doesn't work with heapqExposition
Thanks, this answer deserves to be at the topHautbois
O
3

The limitation with both answers is that they don't allow ties to be treated as ties. In the first, ties are broken by comparing items, in the second by comparing input order. It is faster to just let ties be ties, and if there are a lot of them it could make a big difference. Based on the above and on the docs, it is not clear if this can be achieved in heapq. It does seem strange that heapq does not accept a key, while functions derived from it in the same module do.
P.S.: If you follow the link in the first comment ("possible duplicate...") there is another suggestion of defining le which seems like a solution.

Obligee answered 13/7, 2015 at 23:19 Comment(1)
The limitation with writing "both answers" is that it is no longer clear which those are.Showcase
T
2

Simple little trick:

Say you have this list of (name,age) as

a = [('Tim',4), ('Radha',9), ('Rob',7), ('Krsna',3)]

And you want to sort this list based on their ageby adding them to a min-heap, instead of writing all the custom comparator stuff, you can just flip the order of the contents of the tuple just before pushing it to the queue. This is because heapq.heappush() sorts by the first element of the tuple by default. Like this:

import heapq
heap = []
heapq.heapify(heap)
for element in a:
    heapq.heappush(heap, (element[1],element[0]))

This is a simple trick if this does your job and you don't want to get into writing the custom comparator mess.

Similarly it sorts the values in ascending order by default. If you want to sort in descending order of age, flip the contents and make the value of the first element of the tuple a negative:

import heapq
heap = []
heapq.heapify(heap)
for element in a:
    heapq.heappush(heap, (-element[1],element[0]))
Tuba answered 27/6, 2023 at 1:13 Comment(0)
A
1

In python3, you can use cmp_to_key from functools module. cpython source code.

Suppose you need a priority queue of triplets and specify the priority use the last attribute.

from heapq import *
from functools import cmp_to_key
def mycmp(triplet_left, triplet_right):
    key_l, key_r = triplet_left[2], triplet_right[2]
    if key_l > key_r:
        return -1  # larger first
    elif key_l == key_r:
        return 0  # equal
    else:
        return 1


WrapperCls = cmp_to_key(mycmp)
pq = []
myobj = tuple(1, 2, "anystring")
# to push an object myobj into pq
heappush(pq, WrapperCls(myobj))
# to get the heap top use the `obj` attribute
inner = pq[0].obj

Performance Test:

Environment

python 3.10.2

Code

from functools import cmp_to_key
from timeit import default_timer as time
from random import randint
from heapq import *

class WrapperCls1:
    __slots__ = 'obj'
    def __init__(self, obj):
        self.obj = obj
    def __lt__(self, other):
        kl, kr = self.obj[2], other.obj[2]
        return True if kl > kr else False

def cmp_class2(obj1, obj2):
    kl, kr = obj1[2], obj2[2]
    return -1 if kl > kr else 0 if kl == kr else 1

WrapperCls2 = cmp_to_key(cmp_class2)

triplets = [[randint(-1000000, 1000000) for _ in range(3)] for _ in range(100000)]
# tuple_triplets = [tuple(randint(-1000000, 1000000) for _ in range(3)) for _ in range(100000)]

def test_cls1():
    pq = []
    for triplet in triplets:
        heappush(pq, WrapperCls1(triplet))
        
def test_cls2():
    pq = []
    for triplet in triplets:
        heappush(pq, WrapperCls2(triplet))

def test_cls3():
    pq = []
    for triplet in triplets:
        heappush(pq, (-triplet[2], triplet))

start = time()
for _ in range(10):
    test_cls1()
    # test_cls2()
    # test_cls3()
print("total running time (seconds): ", -start+(start:=time()))

Results

use list instead of tuple, per function:

  • WrapperCls1: 16.2ms
  • WrapperCls1 with __slots__: 9.8ms
  • WrapperCls2: 8.6ms
  • move the priority attribute into the first position ( don't support custom predicate ): 6.0ms.

Therefore, this method is slightly faster than using a custom class with an overridden __lt__() function and the __slots__ attribute.

Autotoxin answered 28/2, 2022 at 12:49 Comment(2)
Have you check those results? I got something like[<functools.KeyWrapper> ...] instead of the value.Lade
@Lade Did you forget to use the .obj attribute?Autotoxin
U
0

Simple and Recent

A simple solution is to store entries as a list of tuples for each tuple define the priority in your desired order if you need a different order for each item within the tuple just make it the negative for descending order.

See the official heapq python documentation in this topic Priority Queue Implementation Notes

Underclay answered 24/12, 2022 at 6:11 Comment(0)
G
0

Using the answer from Fanchen Bao above, I created a Max Priority Queue by extending tuple:

import heapq

class MaxTuple(tuple):
    def __lt__(self, other):
        return self[0] > other[0]

my_tuples = [(2, "orange"), (1, "red"), (5, "blue"), (3, "yellow"), (4, "green")]

my_queue = [MaxTuple(t) for t in my_tuples]

heapq.heapify(my_queue)
while my_queue:
    print(heapq.heappop(my_queue))

Which pops the heap from max to min:

(5, 'blue')
(4, 'green')
(3, 'yellow')
(2, 'orange')
(1, 'red')
Gyno answered 29/3 at 8:18 Comment(0)

© 2022 - 2024 — McMap. All rights reserved.