0

Criei uma interface com Tkinter para usar como environment para meus estudos com aprendizagem por reforço. Meu objetivo é rodar a classe que cria o environment e com um loop externo, fazer o envio de ações para treinar meu modelo de aprendizagem.

Problema que não estou sabendo lidar bem: A rotina de treinamento do modelo ficava dentro da mesma classe do ambiente, então, eu não tinha problema algum, depois que decidi que a rotina de treinamento vai ficar fora da classe que começaram os problemas. Tentei resolver com Thread, mas gerou um problema de escopo.

Código da classe que cria o ambiente:

import os
import sys
import threading
import tkinter as tk
import numpy as np
import time
import copy
import cv2

from PIL import Image

class Maze(threading.Thread):
    def __init__(self, config, mode):
        threading.Thread.__init__(self)
        self.root   =  None
        self.frame  =  None
        self.config =  config
        self.mode   =  mode
        self.shape  =  self.config["environment"].shape

        self.init    = [0, 0, 0]
        self.current = [0, 0, 0]
        self.ids     = []
        self.lines   = 0
        self.columns = 0

        self.canvas = None
        self.colorAnimate  = 'mediumOrchid1'
        self.paletteInit   = ('white', 'white', 'white', 'white')
        self.paletteLocked = ('gray10', 'gray10', 'gray10', 'gray10',)
        self.paletteUnlocked = ('white', 'white', 'white', 'white')
        self.paletteCurrent  = ('gray80', 'gray80', 'gray80', 'gray80')
        self.palettePositiveReward = ('lime green', 'lime green', 'lime green', 'lime green')
        self.paletteNegativeReward = ('firebrick1', 'firebrick1', 'firebrick1', 'firebrick1')

        self.start()

    def run(self):
        self.root  = tk.Tk()
        self.frame = tk.Frame(self.root)
        self.frame.pack()

        # Draw canvas
        self.draw_canvas()

        # Start the Tk GUI.
        self.root.mainloop()

    def draw_canvas(self):
        canvas_height = self.config["height"]
        canvas_width  = self.config["width"]
        self.canvas   = tk.Canvas(self.frame, width=canvas_width, height=canvas_height, background='gray75')
        self.canvas.pack()

        y = 0
        for i in self.config["environment"]:
            x = 0
            self.columns = 0
            for j in i:
                if j[0] == ' ':
                    self.draw(x=x, y=y, color=self.paletteUnlocked)
                elif j[0] == '+':
                    self.draw(x=x, y=y, color=self.palettePositiveReward)
                elif j[0] == '-':
                    self.draw(x=x, y=y, color=self.paletteNegativeReward)
                elif j[0] == 'I':
                    self.draw(x=x, y=y, color=self.paletteInit)
                    self.init[0] = self.lines
                    self.init[1] = self.columns
                    self.current = copy.deepcopy(self.init)
                else:
                    self.draw(x=x, y=y, color=self.paletteLocked)
                x += self.config["widthSquares"]*2
                self.columns += 1
            y += self.config["widthSquares"]*2
            self.lines += 1

        self.ids    =  self.ids.astype(int)
        self.ids    =  self.ids.reshape(self.shape[0], self.shape[1], 4)

    def draw(self, x, y, color):

        c1, c2, c3, c4 = color

        x1 = x / 2
        y1 = y / 2
        x2 = (self.config["widthSquares"] + x) / 2
        y2 = (self.config["widthSquares"] + y) / 2
        x3 = self.config["widthSquares"] + x / 2
        y3 = y / 2
        points = [x1, y1, x2, y2, x3, y3]
        id0 = self.canvas.create_polygon(points, fill=c1)

        x1 = self.config["widthSquares"] + x / 2
        y1 = self.config["widthSquares"] + y / 2
        x2 = (self.config["widthSquares"] + x) / 2
        y2 = (self.config["widthSquares"] + y) / 2
        x3 = self.config["widthSquares"] + x / 2
        y3 = y / 2
        points = [x1, y1, x2, y2, x3, y3]
        id1 = self.canvas.create_polygon(points, fill=c2)

        x1 = self.config["widthSquares"] + x / 2
        y1 = self.config["widthSquares"] + y / 2
        x2 = (self.config["widthSquares"] + x) / 2
        y2 = (self.config["widthSquares"] + y) / 2
        x3 = x / 2
        y3 = self.config["widthSquares"] + y / 2
        points = [x1, y1, x2, y2, x3, y3]
        id2 = self.canvas.create_polygon(points, fill=c3)

        x1 = x / 2
        y1 = y / 2
        x2 = (self.config["widthSquares"] + x) / 2
        y2 = (self.config["widthSquares"] + y) / 2
        x3 = x / 2
        y3 = self.config["widthSquares"] + y / 2
        points = [x1, y1, x2, y2, x3, y3]
        id3 = self.canvas.create_polygon(points, fill=c4)

        self.ids = np.append(self.ids, [id0, id1, id2, id3])

Aqui o código da rotina de treinamento:

import numpy as np
from environment.Maze import Maze
from agents.QTable import QTable


def train(config):
    # Episodes to run
    episodes = config["episodes"]

    # Shape of array
    shape = config["environment"].shape

    # Length for Q-Table (All possible states for this environment)
    state_dim = shape[0] * shape[1]

    # Instance Q-Table
    q_table = QTable(state_dim=state_dim, action_dim=4, gamma=0.999, alpha=0.8)

    # Instance Game Maze
    appMaze = Maze(config, mode='train-qtable')

    # Initialize reward with zero
    reward = 0

    # Loop episodes
    for episode in range(1, episodes + 1):

        # Print current episode
        print("\nEpisode: {}/{}".format(episode, episodes))

        # Reset game
        appMaze.reset()

        # Done False, is init
        done = False

        # Get current state
        state = appMaze.get_observable()

        # Play
        while not done:
            # Get action by q_table or epsilon pseudo random
            action = q_table.select_action(state)

            # Execute action
            next_state, reward, done = appMaze.step(action)

            # Update Q-Table
            q_table.update_q_table(state, next_state, action, reward)

            # Set current state now
            state = next_state

        if reward > 0:
            print("Winner")
        else:
            print("Looser")

    # Finish
    print("Finished")


if __name__ == "__main__":

    # Config for environment
    config = {
        "height": 600,  # Height for canvas
        "width": 600,  # Width for canvas
        "widthSquares": 100,  # Width and height for square
        "episodes": 100000, # Run this number of episodes
        "animate": False, # Animate action
        "delayBetweenAction": 0,  # Delay in seconds
        "rewardPositive": 10,
        "rewardNegative": -10,
        "rewardEachStep": -0.01,
        "rewardInvalidStep": -1,
        "image_dim": (64, 64, 2),
        "environment": np.array([
            [[' '], [' '], [' '], [' '], [' '], [' ']],
            [[' '], ['X'], ['X'], [' '], ['X'], ['X']],
            [['I'], ['X'], [' '], [' '], [' '], [' ']],
            [[' '], ['X'], ['X'], ['X'], ['X'], ['+']],
            [[' '], [' '], [' '], [' '], [' '], [' ']],
            [['-'], ['-'], ['-'], ['-'], ['-'], ['-']]
        ])
    }

    # Run train
    train(config)

Quando executa esse código, tudo corre normal, até chegar na linha appMaze.reset(). Na classe Maze existe a variável self.ids = [], que é responsável por armazenar os ID´s dos retangulos que foram desenhados no canvas, somente com esse ID eu posso mudar a cor do preenchimento dos retangulos e animar a direção escolhida pelo agente, mas quando chega nessa linha do reset, essa variável está vazia, assim como na definição dela no método __init__, com isso gerando o erro: IndexError: list index out of range

Eu sei que deixei passar algo despercebido, mas como todo iniciante, não está claro pra mim.

Ao rodar o código, ele cria uma interface como essa: enter image description here

Desde já agradeço.

1 Resposta 1

0

Entendo que existam duas soluções possíveis no seu caso para compartilhar uma mesma variável (no caso self.ids) em duas classes:

  1. Usar herança entre classes com atributo de classe
  2. Usar Singleton

Herança (exemplo)

>>> class Mae:
...     compartilhado = "Definido na classe mãe"  # atributo de classe
...     def __init__(self):
...         self.mae = "Mãe"                      # atrubuto de instância
...

>>> class Filha(Mae):
...     def __init__(self):
...         self.filha = "Filha"    # atributo de instância
...

>>> f = Filha()

>>> f.filha
'Filha'

>>> f.mae
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
AttributeError: 'Filha' object has no attribute 'mae'

>>> f.compartilhado
'Definido na classe mãe'

Perceba que o atributo mae seria somente de uma instância da classe mãe, porém o atributo de classe compartilhado pertence aos dois

Singleton (exemplo)

>>> class Singleton:
...     _instance = None
...     def __init__(self):
...         self.sing = "Singleton"
...     @staticmethod
...     def get_instance():
...         if Singleton._instance is None:
...             Singleton._instance = Singleton()
...         return Singleton._instance
...

Instanciando duas variáveis.

>>> single1 = Singleton.get_instance()
>>> single1.sing
'Singleton'

>>> single2 = Singleton.get_instance()
>>> single2.sing
'Singleton'
>>>

Nada demais...

Mas perceba o que acontecerá se vc mudar o atributo em uma das instâncias.

>>> single2.sing = "Novo valor"
>>> single2.sing
'Novo valor'

>>> single1.sing
'Novo valor'

Veja que o valor mudou para a outra instância também

Nota: Nesta pequena implementação, não levei em consideração o race condition

Para exemplos mais "corretos" veja aqui

Espero que ajude

1
  • Olá Paulo, obrigado por sua resposta. Infelizmente ela não me ajudou, porque a classe Maze, quando eu removo o Tkinter, ela funciona como esperado. Parece que a Thread do Tkinter está influenciando esse comportamento que eu não esperava.
    – Duque
    3/02/2021 às 10:56

Você deve fazer log-in para responder a esta pergunta.

Esta não é a resposta que você está procurando? Pesquise outras perguntas com a tag .