Zadání
Základem algoritmu je tedy umět vzít N intervalů a zjistit, které se překrývají. Vyzkoušel jsem různé postupy a podělím se s vámi o výsledky v pořadí od komplikovaných postupů po nejjednodušší.
Vzorky
Prvně si vygeneruji 100 náhodně rozmístěných intervalů mezi 0 a 1000, délky 10
import random
nsamples = 100
#list of (start position, length, name)
samples = [(random.randint(0,990), 10, i) for i in xrange(nsamples)]
Algoritmy založené na vkládání
Hlavní myšlenkou je rozdělit si intervaly pomocí binárního stromu a testovat kam vložit nový interval. V podstatě se jedná o 3 implementace binárního stromu. Ovšem pro usnadnění testování jsem si vytvořil pomocnou třídu interval.
class Interval:
def __init__(self, lower, upper):
self.upper = upper
self.lower = lower
def union(self, interval):
'''
self := self union interval
'''
self.lower = min(self.lower, interval.lower)
self.upper = max(self.upper, interval.upper)
def __cmp__(self, interval):
if self.upper <= interval.lower:
return -1
elif self.lower >= interval.upper:
return 1
else:
return 0
Všechny implementace binárního stromu se budou používat stejně.
tree = BTreeCluster()
for s, e, i in samples:
tree.append(s, e, i)
tree.write()
1. Implementace
První implementace je jen jednoduše napsaný binární strom, který si vytvoří pod-uzly až teprve po přidání prvního prvku. Tento přístup bude paměťově nejnáročnější. Zato implementace v C++ by byla poměrně přímočará.
class BTreeCluster1:
'''
classical Binary tree with interval and intersecting objects in each node
'''
def __init__(self):
#lazy inicialization
self.left = None
self.right = None
self.inter = 0
self.group = []
def append(self, pos, lenght, name):
if not self.group:
self.group = [name]
self.inter = Interval(pos, pos+lenght)
self.left = BTreeCluster1()
self.right = BTreeCluster1()
else:
inter = Interval(pos, pos+lenght)
if self.inter == inter:
self.group.append(name)
self.inter.union(inter)
elif self.inter > inter:
self.right.append(pos, lenght, name)
else:
self.left.append(pos, lenght, name)
def write(self):
if self.left:
self.left.write()
if len(self.group) > 2:
print 'collision', self.group
if self.right:
self.right.write()
2. Implementace
Druhá implementace se snaží více šetřit pamětí a přiřazuje uzlům indexy. Uzel s indexem n má pod-uzly 2n a 2n+1. Protože intervaly mohou do funkce přijít třeba seřazené a vzniklý strom může být značně nevyvážený používám na uložení uzlů hash tabulku (slovník) s indexy na místě klíčů. Protože ve standardní knihovně C++ není implementace hash tabulky, bylo by přepsání složitější. Stejně jako v prvním případě by bylo vhodné občas strom vyvážit, ale pro jednoduchost a lenost to vynechám.
class BTreeCluster2:
'''
Binary tree with interval and intersecting objects in each node
encoded in dictionary (hash table)
node number i has subnodes 2i, 2i+1
hash table saves memory even for unballanced tree
'''
def __init__(self):
self.groups = {} #number:interval, group
def append(self, pos, lenght, name):
inter = Interval(pos, pos+lenght)
index = 1
while True:
if index not in self.groups:
self.groups[index] = Interval(pos, pos+lenght), [name]
return
iinter, igroup = self.groups[index]
if inter == iinter:
igroup.append(name)
iinter.union(inter)
return
elif inter > iinter:
index = 2*index + 1
else:
index = 2*index
def write(self):
for i,g in self.groups.values():
if len(g) > 2:
print 'collision', g
3. Implementace
Narozdíl od prvních dvou jsem se rozhodl strom přerovnávat v průběhu přidávání (halda by byla účinnější). Uzly jsou uloženy v poli srovnaném podle pozice intervalu a pole je prohledáváno pomocí půlení intervalu.
class BTreeCluster3:
'''
Binary tree with interval and intersecting objects in each node
encoded in sorted array of disjunctive intervals
sorted is archieved by inserting new nodes to right position
'''
def __init__(self):
self.groups = [] #number:interval, group
def append(self, pos, lenght, name):
'''
finding the right position is archieved by iterative interval dividing
'''
inter = Interval(pos, pos+lenght)
if not self.groups:
self.groups = [(inter, [name])]
return
upper = len(self.groups)-1
lower = 0
while True:
##print lower, upper
index = (upper + lower) / 2
iinter, igroup = self.groups[index]
if inter == iinter:
#exteding existing one
igroup.append(name)
iinter.union(inter)
return
elif inter > iinter:
if lower == upper or index == upper:
self.groups.insert(index+1, (inter, [name]))
print 'including after'
return
lower = index + 1
else:
if lower == upper or index == lower:
self.groups.insert(index, (inter, [name]))
print 'including before'
return
upper = index - 1
def write(self):
for i,g in self.groups:
if len(g) > 2:
print 'collision', g
Všechny tyto implementace jsou poměrně rozsáhlé, ale umožňují přidání nového prvku kdykoliv v průběhu programu. Následující postup je jednodušší, ale danou věc neumožňuje. Místo prvních 3 pokusů by možná stálo za to implementovat haldu.
Postupy založené na syntaktickém analyzátoru
Pokud intervaly seřadíme podle počátku lze je sekvenčně procházet, jednoduše testovat kolize a vracet intervaly, které se protínají. Je potřeba si však poradit s tím, kdy začít novou skupinu a také s přečtením posledního prvku.
1. Implementace
Je napsána jednoduše a jistě by šla snadno převést do C++, speciálně při využití knihovny <sort>.
EOF = 'EOF'
def readgroup(token, it):
'''
c-like parsing function
returns last read token and group of intersecting intervals
'''
s, e, i = token
end = s + e
group = [i]
for token in it:
s, e, i = token
if end <= s:
return token, group
else:
group.append(i)
end = s + e
return EOF, group
samples.sort()
it = iter(samples)
token = it.next()
while token != EOF:
token, g = readgroup(token, it)
if len(g) > 1:
print 'collision', g
Implementace nepočítá s prázdným vstupním seznamem. Po načtení posledního prvku signalizuje konec speciálním tokenem EOF. Protože se v průběhu funkce readgroup vrací vygenerované skupiny je potřeba vracet i naposledy přečtený token a navázat ve zpracování funkce ve stejném místě. Pokud by funkce vracela seznam seznamů, tak by její implementace byla jednodušší ale paměťově náročnější. Realizace výpisu přímo při nalezení kolize by funkci zjednodušila. Cílem však není výpis ale umožnit s nalezenými skupinami dále pracovat.
2. Implementace
Je první algoritmus přepsaný do Pythonu a využívající techniky generátorů.
def readgroup(samples):
group = []
for s, e, i in samples:
if group and end <= s:
yield group
group = []
group.append(i)
end = s + e
yield group
samples.sort()
for g in readgroup(samples):
if len(g) > 1:
print 'collision', g
Závěr
Možná se zdá, že jsem mohl napsat jen poslední příspěvek. Chtěl jsem však ukázat, jak v závislosti na požadavcích, použitých konstruktech jazyka a přijaté abstrakci lze vytvořit různé implementace, které se liší v rychlosti, použité paměti a délce zdrojového kódu. Předpokládám, že nejsnadněji by v C++ šla vytvořit předposlední implementace. A jaký je Váš názor?
Žádné komentáře:
Okomentovat