Files
platz/libs/ga.py
T
Michael Stangl 644f61ef6b Adaptiver GA-Zyklus, Pool-Groessen-Panel, Konfig direkt aus cfg-Dateien
Zyklus/Farm:
- neuer adaptiver Modus: grosse Startpopulation, die je Generation
  schrumpft (Start/Schrumpfung/MinPopulation) und bei Konvergenz stoppt
  (Geduld Generationen ohne Verbesserung >= Schwelle, max MaxGenerationen)
- Zyklus.laden erkennt adaptives Profil am Feld 'Start' (Modus adaptiv),
  sonst klassisch (Abfolge/Anzahl); Farm._LaufStatisch/_LaufAdaptiv
- neue Trunkierungs-Selektion Farm.PopulationBegrenzen
- [Adaptiv]-Profil in zyklus.cfg (dokumentiert)

Entwicklungsgrafik:
- zweites Panel unter dem Verlauf zeigt die Anzahl erzeugter
  Verteilungen im Pool je Schritt (gemeinsame Zeitachse)

Konfiguration:
- platz.cfg entfernt; zyklus.cfg/strafen.cfg werden direkt aus
  $PLATZ_CFG geladen
- neuer Schalter --zyklus-art (Default Adaptiv) waehlt das GA-Profil
- web/main.py analog: direktes Laden, Profil via PLATZ_ZYKLUSART

Doku (README.md, CLAUDE.md) und Tests entsprechend aktualisiert;
Beispiele mit dem Adaptiv-Default neu berechnet.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-11 14:31:59 +02:00

523 lines
18 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 __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 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()
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 LoesungenErzeugen(self, N, Klasse, *Arguments, **keywords):
""" Erzeuge Loesungen"""
for i in range(N):
L = Klasse( *Arguments, **keywords )
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)
for n in range(N):
#print(" -"+repr(n+1)+" mal")
for L in chain(self.Pool):
O = deepcopy(L)
self.PoolNeu.append(O.mutieren())
#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)
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