"""
Enthaelt alle Datenstrukturen um die einen genetischen Algorithmus darzustellen
"""
from random import *
from itertools import chain
from copy import deepcopy
from functools import total_ordering
import ast
import configparser, os
def _StringAusdruck(text):
"""wertet einen einfachen String-Ausdruck sicher aus.
zyklus.cfg nutzt kompakte Schreibweisen wie
'e,'+'s,z,m,j,z,'*3+'s'
also String-Literale, mit '+' verkettet und mit '*' vervielfacht.
ast.literal_eval kann das nicht (nur reine Literale), eval() waere
unsicher. Dieser Mini-Auswerter erlaubt ausschliesslich str-Literale,
ganze Zahlen sowie die Operatoren + und * - kein Funktionsaufruf,
keine Namen, kein beliebiger Code.
"""
knoten = ast.parse(text, mode='eval').body
def rechne(n):
if isinstance(n, ast.Constant) and isinstance(n.value, (str, int)):
return n.value
if isinstance(n, ast.BinOp) and isinstance(n.op, (ast.Add, ast.Mult)):
links, rechts = rechne(n.left), rechne(n.right)
return links + rechts if isinstance(n.op, ast.Add) else links * rechts
raise ValueError("unerlaubter Ausdruck in zyklus.cfg: %r" % text)
ergebnis = rechne(knoten)
if not isinstance(ergebnis, str):
raise ValueError("zyklus.cfg-Ausdruck ergibt keinen String: %r" % text)
return ergebnis
__author__ = "Michael Stangl"
__version__ = "$Revision: $"
__date__ = "$Date: $"
__copyright__ = "Copyright (c) 2005 Michael Stangl"
__license__ = "Python"
@total_ordering
class Loesung:
"""ein einfaches Loesungsobjekt. Taugt um es als Superklasse zu ueberladen"""
def __init__(self, *Argumente, **keywords):
self.Id = -1 * randint(1,100)
self.value = self.Id
print(" initialisiere Loesung " + repr(self.Id))
def mutieren(self):
print(" mutiere Loesung " + repr(self.Id))
self.Id = -1 * randint(1,100)
self.value = self.Id
return self
def kreuzen(self,obj):
print(" kreuze Loesung " + repr(self.Id))
def laden(self):
print(" lade Loesung " + repr(self.Id))
def speichern(self):
print(" speichere Loesung " + repr(self.Id))
def getGruppenId(self):
"""Id fuer den Dedup-Vergleich (Basis: jede Loesung eigenstaendig).
Unterklassen ueberladen dies mit einer inhaltlichen Signatur, so
dass gleiche Loesungen dieselbe Id bekommen und nicht mehrfach
gehalten werden."""
return id(self)
def getPersonenId(self):
"""feinere Id fuer wirklich identische Loesungen (Basis: eigenstaendig)"""
return id(self)
def __repr__(self):
return ' L '+ repr(self.Id)
def __eq__(self,other):
return self.value == other.value
def __lt__(self,other):
return self.value < other.value
class Zyklus:
"""Definiert wie ein Zyklus ablaufen soll:
wann sollen Objekte generiert, gekreuzt, mutiert und selektiert werden
wieviele sind davon jeweils betroffen
"""
# adaptive Vorgaben, falls das Profil kein eigenes Feld setzt
AdaptivDefaults = {
'Start': 40, # Groesse der Startpopulation (Anfangswert)
'MutationProLoesung': 2,# wie oft jede Loesung je Generation mutiert wird
'Schrumpfung': 0.85, # Faktor, um den die Population je Generation faellt
'MinPopulation': 5, # untere Schranke der Population
'MaxGenerationen': 50, # harte Obergrenze der Generationen
'Geduld': 3, # Generationen ohne Verbesserung bis zum Abbruch
'Schwelle': 1.0 } # als Verbesserung zaehlende Mindestdifferenz
def __init__(self, Name='default',
Abfolge='e,m,s',
Anzahl='10,10,10'):
self.Name = Name
self.Abfolge = Abfolge.split(',')
self.Anzahl = Anzahl.split(',')
# 'statisch' = klassische Abfolge/Anzahl abspielen;
# 'adaptiv' = Generationen-Schleife mit Abbruch bei Konvergenz.
self.Modus = 'statisch'
self.Adaptiv = {}
def __repr__(self):
if self.Modus == 'adaptiv':
return( "Zyklus: %s (adaptiv)\n"
" Parameter: %s\n" % ( self.Name, self.Adaptiv ) )
return( "Zyklus: %s\n"
" Zyklusabfolge: %s\n"
" Anzahl : %s\n"
% ( self.Name,
self.Abfolge,
self.Anzahl ) )
def laden(self, FileName, ZyklusName):
config = configparser.ConfigParser()
ZyklusPath= os.path.split( FileName )
ZyklusPath= os.path.join( FileName )
config.read( ZyklusPath )
self.Name = ZyklusName
# Adaptives Profil, sobald ein 'Start'-Feld vorhanden ist. Dann wird
# die Aktionsabfolge nicht aus der Config gelesen, sondern von der
# Farm generationenweise erzeugt (mit Abbruch bei Konvergenz).
if config.has_option( ZyklusName, 'Start' ):
self.Modus = 'adaptiv'
self.Adaptiv = dict(self.AdaptivDefaults)
for Feld in self.AdaptivDefaults:
if config.has_option( ZyklusName, Feld ):
self.Adaptiv[Feld] = ast.literal_eval(
config.get( ZyklusName, Feld ) )
return
self.Modus = 'statisch'
Abfolge = _StringAusdruck( config.get( ZyklusName, 'Abfolge' ))
Anzahl = _StringAusdruck( config.get( ZyklusName, 'Anzahl' ))
self.Abfolge = Abfolge.split(',')
self.Anzahl = Anzahl.split(',')
class Entwicklung:
"""Beobachtungsobjekt: zeichnet den Verlauf der Loesungsfindung auf.
Nach jeder Zyklus-Aktion wird ein Messpunkt erfasst (bester,
durchschnittlicher und schlechtester Wert im aktuellen Pool).
alsSVG() zeichnet daraus eine Verlaufsgrafik: Zeitachse (Aktions-
index) nach rechts, Wertigkeit auf der Y-Achse. Domaenen-agnostisch;
funktioniert mit jeder Loesung, die ein Attribut .value hat.
"""
# Buchstaben der Zyklus-Abfolge -> lesbare Namen fuer die Beschriftung
AktionsNamen = {
'e': 'erzeugen', 'm': 'mutieren', 's': 'selektieren',
'S': 'zufaellig', 'j': 'behalten', 'z': 'Generation' }
def __init__(self):
# je Messpunkt: (Schrittindex, Aktion, bester, schnitt, schlechtester, N)
self.Punkte = []
def erfassen(self, Aktion, Pool):
"""einen Messpunkt aus dem aktuellen Pool aufnehmen"""
if not Pool:
return
Werte = [ L.value for L in Pool ]
self.Punkte.append( (
len(self.Punkte),
Aktion,
max(Werte),
sum(Werte) / len(Werte),
min(Werte),
len(Werte) ) )
def __repr__(self):
r = "-- Entwicklung (%d Messpunkte):\n" % len(self.Punkte)
for (i, Aktion, best, schnitt, schlecht, n) in self.Punkte:
r = r + (" %3d %-12s best=%s schnitt=%.1f schlecht=%s (N=%d)\n"
% (i, self.AktionsNamen.get(Aktion, Aktion),
best, schnitt, schlecht, n))
return r
def alsSVG(self, OutputSVG, Breite=900, Hoehe=650, Rand=60, PoolHoehe=150):
"""zeichnet den aufgezeichneten Verlauf als SVG-Liniendiagramm
Oberes Panel: Wertigkeit (beste/durchschnittliche/schlechteste
Loesung) ueber die Zyklus-Aktionen. Unteres Panel (PoolHoehe hoch):
Anzahl der erzeugten Verteilungen im Pool je Schritt, auf derselben
Zeitachse. Beide Panels teilen sich die X-Achse.
"""
if not self.Punkte:
raise ValueError("keine Messpunkte aufgezeichnet")
AlleWerte = []
for (_i, _a, best, schnitt, schlecht, _n) in self.Punkte:
AlleWerte.extend( (best, schnitt, schlecht) )
WMin, WMax = min(AlleWerte), max(AlleWerte)
if WMax == WMin:
WMax = WMin + 1 # flache Linie nicht auf den Rand kleben
# Pool-Groessen (N je Messpunkt, Index 5)
NMax = max( p[5] for p in self.Punkte )
if NMax == 0:
NMax = 1
N = len(self.Punkte)
PlotBreite = Breite - 2 * Rand
# Oberes Panel (Wertigkeit) reicht von Rand bis kurz vor das
# untere Panel; dazwischen ein kleiner Zwischenraum (Luecke).
Luecke = 50
OberUnten = Hoehe - Rand - PoolHoehe - Luecke # Unterkante oberes Panel
OberHoehe = OberUnten - Rand
UnterOben = OberUnten + Luecke # Oberkante unteres Panel
UnterUnten = Hoehe - Rand # Unterkante unteres Panel
def X(i):
if N == 1:
return Rand + PlotBreite / 2
return Rand + PlotBreite * i / (N - 1)
def Y(wert):
# hoher (besserer) Wert oben, niedriger unten
return Rand + OberHoehe * (WMax - wert) / (WMax - WMin)
def YPool(n):
# 0 an der Unterkante des unteren Panels, NMax an der Oberkante
return UnterUnten - (UnterUnten - UnterOben) * n / NMax
def Polyline(Index, Farbe, Beschriftung):
punkte = ' '.join( '%.1f,%.1f' % (X(i), Y(p[Index]))
for i, p in enumerate(self.Punkte) )
return ('\n' % (punkte, Farbe), Beschriftung, Farbe)
Teile = []
Teile.append( '\n' )
Teile.append(
'\n' )
fout = open( OutputSVG, 'w' )
fout.writelines( Teile )
fout.close()
class Abstammung:
"""Beobachtungsobjekt: zeichnet den Stammbaum aller Loesungen auf.
Jede Loesung bekommt von der Farm eine fortlaufende Abstammungs-Id.
Jede Loesung hat eine fortlaufende Nummer (Id 1..N); die Abstammung
ergibt sich aus dem gemerkten Vorgaenger (Eltern-Id), nicht aus der Id
selbst. Beim Erzeugen entsteht ein Wurzelknoten (keine Eltern), beim
Mutieren ein Kindknoten mit Verweis auf die Eltern-Id. alsSVG() zeichnet
daraus einen Baum: Generationen als Zeilen (oben Start, nach unten
spaeter), Knoten als Punkte (Farbe = Wertigkeit), Mutationskanten
Eltern->Kind. Domaenen-agnostisch; funktioniert mit jeder Loesung, die
.value hat.
"""
def __init__(self):
# je Knoten: Id -> dict(eltern, generation, value, aktion)
self.Knoten = {}
def erfassen(self, Id, Eltern, Generation, value, Aktion):
self.Knoten[Id] = { 'eltern': Eltern, 'generation': Generation,
'value': value, 'aktion': Aktion }
def __repr__(self):
return "-- Abstammung (%d Loesungen)\n" % len(self.Knoten)
def alsSVG(self, OutputSVG, HervorId=None, Breite=1200, Rand=60,
KnotenAbstand=26, ZeilenHoehe=90):
"""zeichnet den aufgezeichneten Stammbaum als SVG.
HervorId (optional): Id der hervorzuhebenden Loesung (z.B. der
finalen besten). Ihr Ast (die Kette ihrer Vorgaenger) wird kraeftig
gezeichnet und der Knoten mit seiner Nummer beschriftet, so dass man
die ganze Abstammung im Baum verfolgen kann."""
if not self.Knoten:
raise ValueError("keine Abstammung aufgezeichnet")
# Ahnenkette der hervorgehobenen Loesung bestimmen (fuer die
# kraeftige Darstellung ihres Astes)
Ahnen = set()
Id = HervorId
while Id is not None and Id in self.Knoten:
Ahnen.add(Id)
Id = self.Knoten[Id]['eltern']
# Knoten nach Generation gruppieren
Generationen = {}
for Id, K in self.Knoten.items():
Generationen.setdefault( K['generation'], [] ).append(Id)
GenListe = sorted(Generationen)
# je Generation die Knoten stabil sortieren (nach Id), damit das Bild
# reproduzierbar ist
for g in Generationen:
Generationen[g].sort()
MaxBreiteKnoten = max( len(v) for v in Generationen.values() )
PlotBreite = max( Breite - 2 * Rand, MaxBreiteKnoten * KnotenAbstand )
Breite = PlotBreite + 2 * Rand
Hoehe = Rand * 2 + (len(GenListe) - 1) * ZeilenHoehe + 40
# Position (x,y) je Knoten-Id bestimmen
Pos = {}
for gi, g in enumerate(GenListe):
Ids = Generationen[g]
n = len(Ids)
y = Rand + gi * ZeilenHoehe
for k, Id in enumerate(Ids):
# horizontal mittig ausrichten
if n == 1:
x = Rand + PlotBreite / 2
else:
x = Rand + PlotBreite * k / (n - 1)
Pos[Id] = (x, y)
# Wertebereich fuer die Knotenfarbe (gut = gruen, schlecht = rot)
Werte = [ K['value'] for K in self.Knoten.values() ]
WMin, WMax = min(Werte), max(Werte)
if WMax == WMin:
WMax = WMin + 1
def Farbe(v):
t = (v - WMin) / (WMax - WMin) # 0 schlecht .. 1 gut
r = int(200 * (1 - t) + 40 * t)
gg = int(50 * (1 - t) + 140 * t)
b = int(50 * (1 - t) + 80 * t)
return '#%02x%02x%02x' % (r, gg, b)
Teile = []
Teile.append( '\n' )
Teile.append(
'\n' )
fout = open( OutputSVG, 'w' )
fout.writelines( Teile )
fout.close()
class Farm:
"""enthaelt und pflegt Loesungsobjekte, kreuzt, mutiert, selektiert sie"""
def __init__(self, Zyklus, Klasse, *Arguments, **keywords):
"""initialisiere die Farm
mit einem definerten Zyklus fuer die Zucht und dem gewunschten zu Zuchtobjekt
Beobachtung: self.Entwicklung zeichnet nach jeder Zyklus-Aktion
den Pool-Zustand auf und laesst sich per alsSVG() visualisieren.
"""
self.Position = 0
self.Pool = []
self.PoolNeu = []
self.Zyklus = Zyklus
self.Klasse = Klasse
self.Entwicklung = Entwicklung()
# Abstammung: jede Loesung bekommt eine fortlaufende _absid; die
# aktuelle Generationsnummer waechst mit jedem Erzeugen/Mutieren.
self.Abstammung = Abstammung()
self._AbsZaehler = 0 # vergibt fortlaufende Loesungs-Ids 1..N
self._Generation = 0
# schon gesehene Loesungs-Signaturen (getGruppenId), damit dieselbe
# Sitzordnung nicht mehrfach gehalten und bewertet wird.
self._GeseheneGruppenIds = set()
if getattr(Zyklus, 'Modus', 'statisch') == 'adaptiv':
self._LaufAdaptiv(Klasse, *Arguments, **keywords)
else:
self._LaufStatisch(Klasse, *Arguments, **keywords)
def _Aufzeichnen(self, Aktion):
"""nach einer Aktion den aktuellen Pool-Zustand fuer die
Entwicklungsgrafik erfassen. Der beobachtete Pool ist der, in dem
gerade die Loesungen liegen: PoolNeu, sobald dort etwas gesammelt
wurde, sonst Pool."""
self.Entwicklung.erfassen(
Aktion, self.PoolNeu if self.PoolNeu else self.Pool )
def _LaufStatisch(self, Klasse, *Arguments, **keywords):
"""klassischer Ablauf: die Abfolge/Anzahl aus zyklus.cfg abspielen."""
for Aktion, Zahl in zip(self.Zyklus.Abfolge, self.Zyklus.Anzahl):
self.Position = self.Position + 1
if Aktion == 'e':
self.LoesungenErzeugen(int(Zahl), Klasse, *Arguments, **keywords)
elif Aktion == 'm':
self.LoesungenMutieren(int(Zahl))
elif Aktion == 's':
self.LoesungenAuswaehlen(int(Zahl))
elif Aktion == 'S':
self.LoesungenZufaelligAuswaehlen(int(Zahl))
elif Aktion == 'j':
self.LoesungenBehalten()
elif Aktion == 'z':
self.NeueGeneration()
else:
pass
self._Aufzeichnen(Aktion)
def _LaufAdaptiv(self, Klasse, *Arguments, **keywords):
"""adaptiver Ablauf: viele Startloesungen, dann Generationen mit
schrumpfender Population; Abbruch, sobald sich die beste Bewertung
ueber 'Geduld' Generationen um weniger als 'Schwelle' verbessert."""
P = self.Zyklus.Adaptiv
Population = max( int(P['Start']), int(P['MinPopulation']) )
# Startpopulation erzeugen (der 'maximale Anfangswert')
self.LoesungenErzeugen(Population, Klasse, *Arguments, **keywords)
self._Aufzeichnen('e')
BesterWert = max(self.Pool).value
OhneVerbesserung = 0
for Gen in range(int(P['MaxGenerationen'])):
self.Position = self.Position + 1
# Mutationskinder erzeugen (in PoolNeu) und die Eltern (Pool)
# dazunehmen, so dass PoolNeu Eltern + Kinder als Kandidaten haelt.
self.LoesungenMutieren( int(P['MutationProLoesung']) )
self._Aufzeichnen('m')
self.LoesungenBehalten() # PoolNeu += Pool
self.NeueGeneration() # Pool = PoolNeu (Kandidaten), PoolNeu leer
self._Aufzeichnen('j')
# Population fuer die naechste Generation schrumpfen (nie unter
# MinPopulation) und den Kandidatenpool auf die besten K trunkieren.
Population = max( int(P['MinPopulation']),
int(Population * float(P['Schrumpfung'])) )
self.PopulationBegrenzen( Population )
self._Aufzeichnen('s')
# Konvergenz pruefen
AktuellBest = max(self.Pool).value
if AktuellBest - BesterWert >= float(P['Schwelle']):
OhneVerbesserung = 0
else:
OhneVerbesserung = OhneVerbesserung + 1
if AktuellBest > BesterWert:
BesterWert = AktuellBest
print("+ Generation %d: bester Wert %.1f, Population %d, "
"ohne Verbesserung %d" % (Gen + 1, AktuellBest, len(self.Pool),
OhneVerbesserung))
if OhneVerbesserung >= int(P['Geduld']):
print("+ Abbruch: %d Generationen ohne nennenswerte "
"Verbesserung" % OhneVerbesserung)
break
def _NeueId(self):
"""neue Id fuer eine Loesung: eine fortlaufende Nummer 1..N ueber
alle je erzeugten Loesungen (Wurzeln wie Mutationskinder gleich).
Die Abstammung ergibt sich nicht aus der Id, sondern aus dem
separat gemerkten Vorgaenger (siehe erfassen(..., Eltern))."""
self._AbsZaehler = self._AbsZaehler + 1
return self._AbsZaehler
def _LoesungIstNeu(self, L):
"""Kontrolle nach jeder Erzeugung: True, wenn die Loesung eine bisher
ungesehene Sitzordnung hat (Vergleich ueber getGruppenId, also auf
Gruppen-Ebene) - und merkt sie dann vor. Duplikate (gleiche Sitz-
ordnung) werden so nicht noch einmal gehalten und bewertet.
Loesungen ohne getGruppenId gelten immer als neu (kein Dedup)."""
Hole = getattr(L, 'getGruppenId', None)
if Hole is None:
return True
Gid = Hole()
if Gid in self._GeseheneGruppenIds:
return False
self._GeseheneGruppenIds.add(Gid)
return True
def LoesungenErzeugen(self, N, Klasse, *Arguments, **keywords):
""" Erzeuge Loesungen (bereits gesehene Sitzordnungen werden verworfen)"""
for i in range(N):
L = Klasse( *Arguments, **keywords )
if not self._LoesungIstNeu(L):
continue # gleiche Sitzordnung schon vorhanden -> nicht behalten
L._absid = self._NeueId() # Wurzel: kein Vorgaenger
self.Abstammung.erfassen(
L._absid, None, self._Generation, getattr(L, 'value', 0), 'e' )
self.Pool.append(L)
#print(repr(len(self.Pool)) + ' Loesungsobjekte erzeugt')
def LoesungenKreuzen(self,N):
"""mache aus zwei Loesungen eine neue"""
#print("+ Loesungen kreuzen")
for i in range(N):
self.Pool.append(self.Pool[i].kreuzen())
def LoesungenMutieren(self, N):
"""veraendere die Loesung, so dass sie hoffentlich besser wird"""
print("+ Loesungen mutieren: ", N)
self._Generation = self._Generation + 1
for n in range(N):
#print(" -"+repr(n+1)+" mal")
for L in chain(self.Pool):
O = deepcopy(L)
Kind = O.mutieren()
if not self._LoesungIstNeu(Kind):
continue # gleiche Sitzordnung schon gesehen -> verwerfen
# Kind bekommt eine neue fortlaufende Nummer; der Vorgaenger
# (Eltern-Id) wird separat in der Abstammung gemerkt.
ElternId = getattr(L, '_absid', None)
Kind._absid = self._NeueId()
self.Abstammung.erfassen(
Kind._absid, ElternId, self._Generation,
getattr(Kind, 'value', 0), 'm' )
self.PoolNeu.append(Kind)
#print(self)
def LoesungenBehalten(self):
"""Behalte alle Eltern aus dem alten Pool"""
print("+ alten Loesungspool behalten")
self.PoolNeu = self.PoolNeu + self.Pool
#print(self)
def NeueGeneration(self):
"""Loesche alle bisherigen Eltern und mache einen neuen Zyklus"""
print("+ Neue Generation erzeugen")
self.Pool = self.PoolNeu
self.PoolNeu = []
#print(self)
#print("---------------")
def LoesungenAuswaehlen(self, N):
"""Selektiere die besten Loesungen"""
print("+ die besten -", repr(N) +" - Loesungen auswaehlen -")
for i in range(N):
L = max(self.Pool)
self.PoolNeu.append(L)
self.Pool.remove(L)
#print(self)
def PopulationBegrenzen(self, K):
"""Trunkierungs-Selektion: behalte im Pool nur die K besten
Loesungen und verwirf den Rest. Anders als LoesungenAuswaehlen
(das additiv nach PoolNeu verschiebt) veraendert dies den Pool an
Ort und Stelle - fuer die adaptive Generationenschleife, damit die
Population wirklich schrumpft."""
K = max( 0, min( K, len(self.Pool) ) )
self.Pool = sorted( self.Pool, reverse=True )[:K]
print("+ Population auf %d beste begrenzt" % K)
def LoesungenZufaelligAuswaehlen(self, N):
"""Selektiere zufaellig Loesungen"""
print("+ zufaellig auswaehlen -"+ repr(N) + "-")
for i in range(N):
self.PoolNeu.append(choice(self.Pool))
#print(self)
def __repr__(self):
"""Drucke die Farm am Bildschirm aus"""
return( " gewaehlte Objekte (%d): %s\n"
" Elternobjekte (%d) : %s\n"
% ( len(self.PoolNeu), self.PoolNeu,
len(self.Pool), self.Pool))
def laden(self):
print("+ laden")
def speichern(self):
print("+ speichern")
def Bester(self):
return max(self.Pool)
def TopN(self, N):
"""die N besten Loesungen, absteigend nach Wertigkeit sortiert.
Erweiterung von Bester() (das nur die einzelne beste liefert): gibt
eine Liste der bis zu N besten Loesungen zurueck, damit sich mehrere
gute Alternativen betrachten lassen. N wird auf [0, len(Pool)]
begrenzt; bei N >= Poolgroesse ist es der ganze Pool (sortiert)."""
N = max( 0, min( N, len(self.Pool) ) )
return sorted( self.Pool, reverse=True )[:N]
if __name__ == '__main__':
print("\n-- Erzeuge Zyklus:")
#Z = Zyklus( Name='Easy',
#Abfolge='e,s,+,m,j,S,+',
#Anzahl='10,5,x,2,x,5,x'
#)
Z = Zyklus( Name='Easy',
Abfolge='e,'+'s,z,m,j,z,'*3+'s',
Anzahl='10,'+'5,x,2,x,x,'*3+'1'
)
print(Z)
print("\n-- Erzeuge Loesungen:")
L1 = Loesung()
L2 = Loesung()
L3 = Loesung()
print(L1, L2, L3)
print('L1 > L2 =', L1 > L2)
print('L2 > L3 =', L2 > L3)
print('L1 > L3 =', L1 > L3)
A=2
B=3
print("\n-- Erzeuge Farm:")
F1 = Farm( Z, Loesung, Arg1=A, Arg2=B, Arg3=L1 )
print(F1)
else:
pass