Michael Stangl
d4844ded67
- getGruppenId (Gruppen-Ebene, personen-unabhaengig: welche Gruppen mit welcher Anzahl je Tisch) und getPersonenId (Personen-Ebene: alphabetische Namen je Tisch) an Sitzplatzverteilung; Basis-Methoden an Loesung. Gleiche Gruppen-Verteilung -> gleiche getGruppenId und damit gleiche Wertigkeit, egal welche Personen die geteilten Plaetze einnehmen. - Farm verwirft nach jeder Erzeugung/Mutation Loesungen mit schon gesehener getGruppenId, so dass dieselbe Sitzordnung nicht mehrfach gehalten und bewertet wird (defensiv: Klassen ohne getGruppenId bleiben unberuehrt, Tests unveraendert). - Web: Varianten-Vergleich ueber getGruppenId statt der person-basierten _Signatur; gruppenId und personenId zusaetzlich im Ergebnis-JSON. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
734 lines
26 KiB
Python
Executable File
734 lines
26 KiB
Python
Executable File
"""
|
|
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 ('<polyline points="%s" fill="none" stroke="%s" '
|
|
'stroke-width="2"/>\n' % (punkte, Farbe), Beschriftung, Farbe)
|
|
|
|
Teile = []
|
|
Teile.append( '<?xml version="1.0" encoding="UTF-8"?>\n' )
|
|
Teile.append(
|
|
'<svg xmlns="http://www.w3.org/2000/svg" width="%d" height="%d" '
|
|
'viewBox="0 0 %d %d" font-family="sans-serif">\n'
|
|
% (Breite, Hoehe, Breite, Hoehe) )
|
|
Teile.append( '<rect width="100%" height="100%" fill="white"/>\n' )
|
|
Teile.append(
|
|
'<text x="%d" y="30" text-anchor="middle" font-size="16" '
|
|
'fill="#333">Entwicklung der Loesungsfindung</text>\n'
|
|
% (Breite // 2) )
|
|
|
|
# --- Oberes Panel: Wertigkeit ---
|
|
# Achsen
|
|
Teile.append(
|
|
'<line x1="%d" y1="%d" x2="%d" y2="%.1f" stroke="#333" '
|
|
'stroke-width="1.5"/>\n' % (Rand, Rand, Rand, OberUnten) )
|
|
Teile.append(
|
|
'<line x1="%d" y1="%.1f" x2="%d" y2="%.1f" stroke="#333" '
|
|
'stroke-width="1.5"/>\n'
|
|
% (Rand, OberUnten, Breite - Rand, OberUnten) )
|
|
|
|
# Y-Gitter + Beschriftung (5 Linien)
|
|
for k in range(6):
|
|
wert = WMin + (WMax - WMin) * k / 5
|
|
y = Y(wert)
|
|
Teile.append(
|
|
'<line x1="%d" y1="%.1f" x2="%d" y2="%.1f" stroke="#eee" '
|
|
'stroke-width="1"/>\n' % (Rand, y, Breite - Rand, y) )
|
|
Teile.append(
|
|
'<text x="%d" y="%.1f" text-anchor="end" font-size="11" '
|
|
'fill="#666">%.0f</text>\n' % (Rand - 8, y + 4, wert) )
|
|
Teile.append(
|
|
'<text x="18" y="%.1f" text-anchor="middle" font-size="12" '
|
|
'fill="#333" transform="rotate(-90 18 %.1f)">Wertigkeit '
|
|
'(hoeher = besser)</text>\n'
|
|
% ((Rand + OberUnten) / 2, (Rand + OberUnten) / 2) )
|
|
|
|
# die drei Kurven
|
|
Kurven = [
|
|
Polyline(2, '#2f6f4f', 'beste Loesung'),
|
|
Polyline(3, '#c08a2d', 'Durchschnitt'),
|
|
Polyline(4, '#a03030', 'schlechteste Loesung') ]
|
|
for (svg, _besch, _farbe) in Kurven:
|
|
Teile.append( svg )
|
|
|
|
# Legende oben rechts
|
|
LegX = Breite - Rand - 160
|
|
LegY = Rand + 10
|
|
for k, (_svg, besch, farbe) in enumerate(Kurven):
|
|
y = LegY + k * 18
|
|
Teile.append(
|
|
'<line x1="%d" y1="%d" x2="%d" y2="%d" stroke="%s" '
|
|
'stroke-width="2"/>\n' % (LegX, y, LegX + 24, y, farbe) )
|
|
Teile.append(
|
|
'<text x="%d" y="%d" font-size="11" fill="#333">%s</text>\n'
|
|
% (LegX + 30, y + 4, besch) )
|
|
|
|
# --- Unteres Panel: Anzahl erzeugter Verteilungen (Pool-Groesse) ---
|
|
Teile.append(
|
|
'<text x="%d" y="%.1f" text-anchor="middle" font-size="13" '
|
|
'fill="#333">Anzahl erzeugter Verteilungen im Pool</text>\n'
|
|
% (Breite // 2, UnterOben - 12) )
|
|
# Achsen
|
|
Teile.append(
|
|
'<line x1="%d" y1="%.1f" x2="%d" y2="%.1f" stroke="#333" '
|
|
'stroke-width="1.5"/>\n' % (Rand, UnterOben, Rand, UnterUnten) )
|
|
Teile.append(
|
|
'<line x1="%d" y1="%.1f" x2="%d" y2="%.1f" stroke="#333" '
|
|
'stroke-width="1.5"/>\n'
|
|
% (Rand, UnterUnten, Breite - Rand, UnterUnten) )
|
|
# Y-Gitter + Beschriftung (0, Mitte, Max)
|
|
for n in (0, NMax // 2, NMax):
|
|
y = YPool(n)
|
|
Teile.append(
|
|
'<line x1="%d" y1="%.1f" x2="%d" y2="%.1f" stroke="#eee" '
|
|
'stroke-width="1"/>\n' % (Rand, y, Breite - Rand, y) )
|
|
Teile.append(
|
|
'<text x="%d" y="%.1f" text-anchor="end" font-size="11" '
|
|
'fill="#666">%d</text>\n' % (Rand - 8, y + 4, n) )
|
|
# Poolgroessen-Kurve
|
|
punkte = ' '.join( '%.1f,%.1f' % (X(i), YPool(p[5]))
|
|
for i, p in enumerate(self.Punkte) )
|
|
Teile.append(
|
|
'<polyline points="%s" fill="none" stroke="#4060a0" '
|
|
'stroke-width="2"/>\n' % punkte )
|
|
|
|
# X-Beschriftung: Aktionsbuchstaben unter dem unteren Panel (bei
|
|
# wenigen Punkten alle, sonst ausgeduennt, damit die Achse lesbar
|
|
# bleibt)
|
|
Schritt = max(1, N // 30)
|
|
for i, p in enumerate(self.Punkte):
|
|
if i % Schritt == 0 or i == N - 1:
|
|
Teile.append(
|
|
'<text x="%.1f" y="%.1f" text-anchor="middle" font-size="10" '
|
|
'fill="#999">%s</text>\n' % (X(i), UnterUnten + 16, p[1]) )
|
|
Teile.append(
|
|
'<text x="%d" y="%d" text-anchor="middle" font-size="12" '
|
|
'fill="#333">Zyklus-Aktion (Zeit)</text>\n'
|
|
% (Breite // 2, Hoehe - 12) )
|
|
|
|
Teile.append( '</svg>\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( '<?xml version="1.0" encoding="UTF-8"?>\n' )
|
|
Teile.append(
|
|
'<svg xmlns="http://www.w3.org/2000/svg" width="%d" height="%d" '
|
|
'viewBox="0 0 %d %d" font-family="sans-serif">\n'
|
|
% (Breite, Hoehe, Breite, Hoehe) )
|
|
Teile.append( '<rect width="100%" height="100%" fill="white"/>\n' )
|
|
Teile.append(
|
|
'<text x="%d" y="30" text-anchor="middle" font-size="16" '
|
|
'fill="#333">Abstammung der Loesungen (Mutationsbaum)</text>\n'
|
|
% (Breite // 2) )
|
|
|
|
# zuerst die Kanten Eltern -> Kind (darunter liegend). Kanten entlang
|
|
# des hervorgehobenen Astes kraeftig, der Rest blass.
|
|
for Id, K in self.Knoten.items():
|
|
Eltern = K['eltern']
|
|
if Eltern is None or Eltern not in Pos or Id not in Pos:
|
|
continue
|
|
x1, y1 = Pos[Eltern]
|
|
x2, y2 = Pos[Id]
|
|
ImAst = Id in Ahnen and Eltern in Ahnen
|
|
Stil = ('stroke="#c04040" stroke-width="2.5"' if ImAst
|
|
else 'stroke="#bbb" stroke-width="1"')
|
|
Teile.append(
|
|
'<line x1="%.1f" y1="%.1f" x2="%.1f" y2="%.1f" %s/>\n'
|
|
% (x1, y1, x2, y2, Stil) )
|
|
|
|
# dann die Knoten
|
|
for Id, K in self.Knoten.items():
|
|
if Id not in Pos:
|
|
continue
|
|
x, y = Pos[Id]
|
|
r = 7 if Id in Ahnen else 5
|
|
Rahmen = ('stroke="#c04040" stroke-width="2"' if Id in Ahnen
|
|
else 'stroke="#333" stroke-width="0.5"')
|
|
Teile.append(
|
|
'<circle cx="%.1f" cy="%.1f" r="%d" fill="%s" %s>'
|
|
'<title>Id %s, Wert %.1f</title></circle>\n'
|
|
% (x, y, r, Farbe(K['value']), Rahmen, Id, K['value']) )
|
|
|
|
# die hervorgehobene Loesung mit ihrer sprechenden Abstammungs-Id
|
|
# beschriften (aus ihr laesst sich die ganze Herkunft ablesen)
|
|
if HervorId in Pos:
|
|
x, y = Pos[HervorId]
|
|
Teile.append(
|
|
'<text x="%.1f" y="%.1f" text-anchor="middle" font-size="12" '
|
|
'font-weight="bold" fill="#c04040">%s</text>\n'
|
|
% (x, y + 22, HervorId) )
|
|
|
|
# Generationsbeschriftung links
|
|
for gi, g in enumerate(GenListe):
|
|
y = Rand + gi * ZeilenHoehe
|
|
Teile.append(
|
|
'<text x="%d" y="%.1f" text-anchor="end" font-size="11" '
|
|
'fill="#666">Gen %s</text>\n' % (Rand - 12, y + 4, g) )
|
|
|
|
Teile.append( '</svg>\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
|