<s> """<STR_LIT>""" <EOL> from . Domain import Domain <EOL> import numpy as np <EOL> from itertools import tee , izip <EOL> import itertools <EOL> from Tkinter import Tk , Canvas <EOL> import os <EOL> from rlpy . Tools import __rlpy_location__ <EOL> __copyright__ = "<STR_LIT>" <EOL> __credits__ = [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" ] <EOL> __license__ = "<STR_LIT>" <EOL> __author__ = [ "<STR_LIT>" , <EOL> "<STR_LIT>" ] <EOL> class Pinball ( Domain ) : <EOL> """<STR_LIT>""" <EOL> default_config_dir = os . path . join ( <EOL> __rlpy_location__ , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ) <EOL> def __init__ ( self , noise = <NUM_LIT> , episodeCap = <NUM_LIT:1000> , <EOL> configuration = os . path . join ( default_config_dir , "<STR_LIT>" ) ) : <EOL> """<STR_LIT>""" <EOL> self . NOISE = noise <EOL> self . configuration = configuration <EOL> self . screen = None <EOL> self . episodeCap = episodeCap <EOL> self . actions_num = <NUM_LIT:5> <EOL> self . actions = [ <EOL> PinballModel . ACC_X , <EOL> PinballModel . DEC_Y , <EOL> PinballModel . DEC_X , <EOL> PinballModel . ACC_Y , <EOL> PinballModel . ACC_NONE ] <EOL> self . statespace_limits = np . array ( <EOL> [ [ <NUM_LIT:0.0> , <NUM_LIT:1.0> ] , [ <NUM_LIT:0.0> , <NUM_LIT:1.0> ] , [ - <NUM_LIT> , <NUM_LIT> ] , [ - <NUM_LIT> , <NUM_LIT> ] ] ) <EOL> self . continuous_dims = [ <NUM_LIT:4> ] <EOL> super ( Pinball , self ) . __init__ ( ) <EOL> self . environment = PinballModel ( <EOL> self . configuration , <EOL> random_state = self . random_state ) <EOL> def showDomain ( self , a ) : <EOL> if self . screen is None : <EOL> master = Tk ( ) <EOL> master . title ( '<STR_LIT>' ) <EOL> self . screen = Canvas ( master , width = <NUM_LIT> , height = <NUM_LIT> ) <EOL> self . screen . configure ( background = '<STR_LIT>' ) <EOL> self . screen . pack ( ) <EOL> self . environment_view = PinballView ( <EOL> self . screen , <EOL> <NUM_LIT> , <EOL> <NUM_LIT> , <EOL> self . environment ) <EOL> self . environment_view . blit ( ) <EOL> self . screen . pack ( ) <EOL> self . screen . update ( ) <EOL> def step ( self , a ) : <EOL> s = self . state <EOL> [ self . environment . ball . position [ <NUM_LIT:0> ] , <EOL> self . environment . ball . position [ <NUM_LIT:1> ] , <EOL> self . environment . ball . xdot , <EOL> self . environment . ball . ydot ] = s <EOL> if self . random_state . random_sample ( ) < self . NOISE : <EOL> a = self . random_state . choice ( self . possibleActions ( ) ) <EOL> reward = self . environment . take_action ( a ) <EOL> self . environment . _check_bounds ( ) <EOL> state = np . array ( self . environment . get_state ( ) ) <EOL> self . state = state . copy ( ) <EOL> return reward , state , self . isTerminal ( ) , self . possibleActions ( ) <EOL> def s0 ( self ) : <EOL> self . environment . ball . position [ <NUM_LIT:0> ] , self . environment . ball . position [ <EOL> <NUM_LIT:1> ] = self . environment . start_pos <EOL> self . environment . ball . xdot , self . environment . ball . ydot = <NUM_LIT:0.0> , <NUM_LIT:0.0> <EOL> self . state = np . array ( <EOL> [ self . environment . ball . position [ <NUM_LIT:0> ] , self . environment . ball . position [ <NUM_LIT:1> ] , <EOL> self . environment . ball . xdot , self . environment . ball . ydot ] ) <EOL> return self . state , self . isTerminal ( ) , self . possibleActions ( ) <EOL> def possibleActions ( self , s = <NUM_LIT:0> ) : <EOL> return np . array ( self . actions ) <EOL> def isTerminal ( self ) : <EOL> return self . environment . episode_ended ( ) <EOL> class BallModel : <EOL> """<STR_LIT>""" <EOL> DRAG = <NUM_LIT> <EOL> def __init__ ( self , start_position , radius ) : <EOL> """<STR_LIT>""" <EOL> self . position = start_position <EOL> self . radius = radius <EOL> self . xdot = <NUM_LIT:0.0> <EOL> self . ydot = <NUM_LIT:0.0> <EOL> def add_impulse ( self , delta_xdot , delta_ydot ) : <EOL> """<STR_LIT>""" <EOL> self . xdot += delta_xdot / <NUM_LIT> <EOL> self . ydot += delta_ydot / <NUM_LIT> <EOL> self . xdot = self . _clip ( self . xdot ) <EOL> self . ydot = self . _clip ( self . ydot ) <EOL> def add_drag ( self ) : <EOL> """<STR_LIT>""" <EOL> self . xdot *= self . DRAG <EOL> self . ydot *= self . DRAG <EOL> def step ( self ) : <EOL> """<STR_LIT>""" <EOL> self . position [ <NUM_LIT:0> ] += self . xdot * self . radius / <NUM_LIT> <EOL> self . position [ <NUM_LIT:1> ] += self . ydot * self . radius / <NUM_LIT> <EOL> def _clip ( self , val , low = - <NUM_LIT:2> , high = <NUM_LIT:2> ) : <EOL> """<STR_LIT>""" <EOL> if val > high : <EOL> val = high <EOL> if val < low : <EOL> val = low <EOL> return val <EOL> class PinballObstacle : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , points ) : <EOL> """<STR_LIT>""" <EOL> self . points = points <EOL> self . min_x = min ( self . points , key = lambda pt : pt [ <NUM_LIT:0> ] ) [ <NUM_LIT:0> ] <EOL> self . max_x = max ( self . points , key = lambda pt : pt [ <NUM_LIT:0> ] ) [ <NUM_LIT:0> ] <EOL> self . min_y = min ( self . points , key = lambda pt : pt [ <NUM_LIT:1> ] ) [ <NUM_LIT:1> ] <EOL> self . max_y = max ( self . points , key = lambda pt : pt [ <NUM_LIT:1> ] ) [ <NUM_LIT:1> ] <EOL> self . _double_collision = False <EOL> self . _intercept = None <EOL> def collision ( self , ball ) : <EOL> """<STR_LIT>""" <EOL> self . _double_collision = False <EOL> if ball . position [ <NUM_LIT:0> ] - ball . radius > self . max_x : <EOL> return False <EOL> if ball . position [ <NUM_LIT:0> ] + ball . radius < self . min_x : <EOL> return False <EOL> if ball . position [ <NUM_LIT:1> ] - ball . radius > self . max_y : <EOL> return False <EOL> if ball . position [ <NUM_LIT:1> ] + ball . radius < self . min_y : <EOL> return False <EOL> a , b = tee ( np . vstack ( [ np . array ( self . points ) , self . points [ <NUM_LIT:0> ] ] ) ) <EOL> next ( b , None ) <EOL> intercept_found = False <EOL> for pt_pair in izip ( a , b ) : <EOL> if self . _intercept_edge ( pt_pair , ball ) : <EOL> if intercept_found : <EOL> self . _intercept = self . _select_edge ( <EOL> pt_pair , <EOL> self . _intercept , <EOL> ball ) <EOL> self . _double_collision = True <EOL> else : <EOL> self . _intercept = pt_pair <EOL> intercept_found = True <EOL> return intercept_found <EOL> def collision_effect ( self , ball ) : <EOL> """<STR_LIT>""" <EOL> if self . _double_collision : <EOL> return [ - ball . xdot , - ball . ydot ] <EOL> obstacle_vector = self . _intercept [ <NUM_LIT:1> ] - self . _intercept [ <NUM_LIT:0> ] <EOL> if obstacle_vector [ <NUM_LIT:0> ] < <NUM_LIT:0> : <EOL> obstacle_vector = self . _intercept [ <NUM_LIT:0> ] - self . _intercept [ <NUM_LIT:1> ] <EOL> velocity_vector = np . array ( [ ball . xdot , ball . ydot ] ) <EOL> theta = self . _angle ( velocity_vector , obstacle_vector ) - np . pi <EOL> if theta < <NUM_LIT:0> : <EOL> theta += <NUM_LIT:2> * np . pi <EOL> intercept_theta = self . _angle ( [ - <NUM_LIT:1> , <NUM_LIT:0> ] , obstacle_vector ) <EOL> theta += intercept_theta <EOL> if theta > <NUM_LIT:2> * np . pi : <EOL> theta -= <NUM_LIT:2> * np . pi <EOL> velocity = np . linalg . norm ( [ ball . xdot , ball . ydot ] ) <EOL> return [ velocity * np . cos ( theta ) , velocity * np . sin ( theta ) ] <EOL> def _select_edge ( self , intersect1 , intersect2 , ball ) : <EOL> """<STR_LIT>""" <EOL> velocity = np . array ( [ ball . xdot , ball . ydot ] ) <EOL> obstacle_vector1 = intersect1 [ <NUM_LIT:1> ] - intersect1 [ <NUM_LIT:0> ] <EOL> obstacle_vector2 = intersect2 [ <NUM_LIT:1> ] - intersect2 [ <NUM_LIT:0> ] <EOL> angle1 = self . _angle ( velocity , obstacle_vector1 ) <EOL> if angle1 > np . pi : <EOL> angle1 -= np . pi <EOL> angle2 = self . _angle ( velocity , obstacle_vector2 ) <EOL> if angle1 > np . pi : <EOL> angle2 -= np . pi <EOL> if np . abs ( angle1 - ( np . pi / <NUM_LIT> ) ) < np . abs ( angle2 - ( np . pi / <NUM_LIT> ) ) : <EOL> return intersect1 <EOL> return intersect2 <EOL> def _angle ( self , v1 , v2 ) : <EOL> """<STR_LIT>""" <EOL> angle_diff = np . arctan2 ( v1 [ <NUM_LIT:0> ] , v1 [ <NUM_LIT:1> ] ) - np . arctan2 ( v2 [ <NUM_LIT:0> ] , v2 [ <NUM_LIT:1> ] ) <EOL> if angle_diff < <NUM_LIT:0> : <EOL> angle_diff += <NUM_LIT:2> * np . pi <EOL> return angle_diff <EOL> def _intercept_edge ( self , pt_pair , ball ) : <EOL> """<STR_LIT>""" <EOL> obstacle_edge = pt_pair [ <NUM_LIT:1> ] - pt_pair [ <NUM_LIT:0> ] <EOL> difference = np . array ( ball . position ) - pt_pair [ <NUM_LIT:0> ] <EOL> scalar_proj = difference . dot ( <EOL> obstacle_edge ) / obstacle_edge . dot ( obstacle_edge ) <EOL> if scalar_proj > <NUM_LIT:1.0> : <EOL> scalar_proj = <NUM_LIT:1.0> <EOL> elif scalar_proj < <NUM_LIT:0.0> : <EOL> scalar_proj = <NUM_LIT:0.0> <EOL> closest_pt = pt_pair [ <NUM_LIT:0> ] + obstacle_edge * scalar_proj <EOL> obstacle_to_ball = ball . position - closest_pt <EOL> distance = obstacle_to_ball . dot ( obstacle_to_ball ) <EOL> if distance <= ball . radius * ball . radius : <EOL> velocity = np . array ( [ ball . xdot , ball . ydot ] ) <EOL> ball_to_obstacle = closest_pt - ball . position <EOL> angle = self . _angle ( ball_to_obstacle , velocity ) <EOL> if angle > np . pi : <EOL> angle = <NUM_LIT:2> * np . pi - angle <EOL> if angle > np . pi / <NUM_LIT> : <EOL> return False <EOL> return True <EOL> else : <EOL> return False <EOL> class PinballModel : <EOL> """<STR_LIT>""" <EOL> ACC_X = <NUM_LIT:0> <EOL> ACC_Y = <NUM_LIT:1> <EOL> DEC_X = <NUM_LIT:2> <EOL> DEC_Y = <NUM_LIT:3> <EOL> ACC_NONE = <NUM_LIT:4> <EOL> STEP_PENALTY = - <NUM_LIT:1> <EOL> THRUST_PENALTY = - <NUM_LIT:5> <EOL> END_EPISODE = <NUM_LIT> <EOL> def __init__ ( self , configuration , random_state = np . random . RandomState ( ) ) : <EOL> """<STR_LIT>""" <EOL> self . random_state = random_state <EOL> self . action_effects = { self . ACC_X : ( <NUM_LIT:1> , <NUM_LIT:0> ) , self . ACC_Y : ( <EOL> <NUM_LIT:0> , <NUM_LIT:1> ) , self . DEC_X : ( - <NUM_LIT:1> , <NUM_LIT:0> ) , self . DEC_Y : ( <NUM_LIT:0> , - <NUM_LIT:1> ) , self . ACC_NONE : ( <NUM_LIT:0> , <NUM_LIT:0> ) } <EOL> self . obstacles = [ ] <EOL> self . target_pos = [ ] <EOL> self . target_rad = <NUM_LIT> <EOL> ball_rad = <NUM_LIT> <EOL> start_pos = [ ] <EOL> with open ( configuration ) as fp : <EOL> for line in fp . readlines ( ) : <EOL> tokens = line . strip ( ) . split ( ) <EOL> if not len ( tokens ) : <EOL> continue <EOL> elif tokens [ <NUM_LIT:0> ] == '<STR_LIT>' : <EOL> self . obstacles . append ( <EOL> PinballObstacle ( zip ( * [ iter ( map ( float , tokens [ <NUM_LIT:1> : ] ) ) ] * <NUM_LIT:2> ) ) ) <EOL> elif tokens [ <NUM_LIT:0> ] == '<STR_LIT:target>' : <EOL> self . target_pos = [ float ( tokens [ <NUM_LIT:1> ] ) , float ( tokens [ <NUM_LIT:2> ] ) ] <EOL> self . target_rad = float ( tokens [ <NUM_LIT:3> ] ) <EOL> elif tokens [ <NUM_LIT:0> ] == '<STR_LIT:start>' : <EOL> start_pos = zip ( * [ iter ( map ( float , tokens [ <NUM_LIT:1> : ] ) ) ] * <NUM_LIT:2> ) <EOL> elif tokens [ <NUM_LIT:0> ] == '<STR_LIT>' : <EOL> ball_rad = float ( tokens [ <NUM_LIT:1> ] ) <EOL> self . start_pos = start_pos [ <NUM_LIT:0> ] <EOL> a = self . random_state . randint ( len ( start_pos ) ) <EOL> self . ball = BallModel ( list ( start_pos [ a ] ) , ball_rad ) <EOL> def get_state ( self ) : <EOL> """<STR_LIT>""" <EOL> return ( <EOL> [ self . ball . position [ <NUM_LIT:0> ] , <EOL> self . ball . position [ <NUM_LIT:1> ] , <EOL> self . ball . xdot , <EOL> self . ball . ydot ] <EOL> ) <EOL> def take_action ( self , action ) : <EOL> """<STR_LIT>""" <EOL> for i in xrange ( <NUM_LIT:20> ) : <EOL> if i == <NUM_LIT:0> : <EOL> self . ball . add_impulse ( * self . action_effects [ action ] ) <EOL> self . ball . step ( ) <EOL> ncollision = <NUM_LIT:0> <EOL> dxdy = np . array ( [ <NUM_LIT:0> , <NUM_LIT:0> ] ) <EOL> for obs in self . obstacles : <EOL> if obs . collision ( self . ball ) : <EOL> dxdy = dxdy + obs . collision_effect ( self . ball ) <EOL> ncollision += <NUM_LIT:1> <EOL> if ncollision == <NUM_LIT:1> : <EOL> self . ball . xdot = dxdy [ <NUM_LIT:0> ] <EOL> self . ball . ydot = dxdy [ <NUM_LIT:1> ] <EOL> if i == <NUM_LIT> : <EOL> self . ball . step ( ) <EOL> elif ncollision > <NUM_LIT:1> : <EOL> self . ball . xdot = - self . ball . xdot <EOL> self . ball . ydot = - self . ball . ydot <EOL> if self . episode_ended ( ) : <EOL> return self . END_EPISODE <EOL> self . ball . add_drag ( ) <EOL> self . _check_bounds ( ) <EOL> if action == self . ACC_NONE : <EOL> return self . STEP_PENALTY <EOL> return self . THRUST_PENALTY <EOL> def episode_ended ( self ) : <EOL> """<STR_LIT>""" <EOL> return ( <EOL> np . linalg . norm ( np . array ( self . ball . position ) <EOL> - np . array ( self . target_pos ) ) < self . target_rad <EOL> ) <EOL> def _check_bounds ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . ball . position [ <NUM_LIT:0> ] > <NUM_LIT:1.0> : <EOL> self . ball . position [ <NUM_LIT:0> ] = <NUM_LIT> <EOL> if self . ball . position [ <NUM_LIT:0> ] < <NUM_LIT:0.0> : <EOL> self . ball . position [ <NUM_LIT:0> ] = <NUM_LIT> <EOL> if self . ball . position [ <NUM_LIT:1> ] > <NUM_LIT:1.0> : <EOL> self . ball . position [ <NUM_LIT:1> ] = <NUM_LIT> <EOL> if self . ball . position [ <NUM_LIT:1> ] < <NUM_LIT:0.0> : <EOL> self . ball . position [ <NUM_LIT:1> ] = <NUM_LIT> <EOL> class PinballView : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , screen , width , height , model ) : <EOL> """<STR_LIT>""" <EOL> self . screen = screen <EOL> self . width = <NUM_LIT> <EOL> self . height = <NUM_LIT> <EOL> self . model = model <EOL> self . x , self . y = self . _to_pixels ( self . model . ball . position ) <EOL> self . rad = int ( self . model . ball . radius * self . width ) <EOL> self . DARK_GRAY = [ <NUM_LIT:64> , <NUM_LIT:64> , <NUM_LIT:64> ] <EOL> self . DARK_BLUE = [ <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT> ] <EOL> self . LIGHT_GRAY = [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] <EOL> self . BALL_COLOR = [ <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:255> ] <EOL> self . TARGET_COLOR = [ <NUM_LIT:255> , <NUM_LIT:0> , <NUM_LIT:0> ] <EOL> for obs in model . obstacles : <EOL> coords_list = map ( self . _to_pixels , obs . points ) <EOL> chain = itertools . chain ( * coords_list ) <EOL> coords = list ( chain ) <EOL> self . screen . create_polygon ( coords , fill = '<STR_LIT>' ) <EOL> self . screen . pack ( ) <EOL> self . target_x , self . target_y = self . _to_pixels ( self . model . target_pos ) <EOL> self . target_rad = int ( self . model . target_rad * self . width ) <EOL> target_id = self . drawcircle ( <EOL> self . screen , <EOL> self . target_x , <EOL> self . target_y , <EOL> self . target_rad , <EOL> '<STR_LIT>' ) <EOL> self . ball_id = self . drawcircle ( <EOL> self . screen , <EOL> self . x , <EOL> self . y , <EOL> self . rad , <EOL> '<STR_LIT>' ) <EOL> self . screen . pack ( ) <EOL> def drawcircle ( self , canv , x , y , rad , color ) : <EOL> return ( <EOL> canv . create_oval ( <EOL> x - rad , <EOL> y - rad , <EOL> x + rad , <EOL> y + rad , <EOL> width = <NUM_LIT:0> , <EOL> fill = color ) <EOL> ) <EOL> def _to_pixels ( self , pt ) : <EOL> """<STR_LIT>""" <EOL> return [ int ( pt [ <NUM_LIT:0> ] * self . width ) , int ( pt [ <NUM_LIT:1> ] * self . height ) ] <EOL> def blit ( self ) : <EOL> """<STR_LIT>""" <EOL> self . screen . coords ( <EOL> self . ball_id , <EOL> self . x - <EOL> self . rad , <EOL> self . y - <EOL> self . rad , <EOL> self . x + <EOL> self . rad , <EOL> self . y + <EOL> self . rad ) <EOL> self . x , self . y = self . _to_pixels ( self . model . ball . position ) <EOL> self . screen . pack ( ) <EOL> def run_pinballview ( width , height , configuration ) : <EOL> """<STR_LIT>""" <EOL> width , height = float ( width ) , float ( height ) <EOL> master = Tk ( ) <EOL> master . title ( '<STR_LIT>' ) <EOL> screen = Canvas ( master , width = <NUM_LIT> , height = <NUM_LIT> ) <EOL> screen . configure ( background = '<STR_LIT>' ) <EOL> screen . pack ( ) <EOL> environment = PinballModel ( configuration ) <EOL> environment_view = PinballView ( screen , width , height , environment ) <EOL> actions = [ <EOL> PinballModel . ACC_X , <EOL> PinballModel . DEC_Y , <EOL> PinballModel . DEC_X , <EOL> PinballModel . ACC_Y , <EOL> PinballModel . ACC_NONE ] <EOL> done = False <EOL> while not done : <EOL> user_action = np . random . choice ( actions ) <EOL> environment_view . blit ( ) <EOL> if environment . episode_ended ( ) : <EOL> done = True <EOL> if environment . take_action ( user_action ) == environment . END_EPISODE : <EOL> done = True <EOL> environment_view . blit ( ) <EOL> screen . update ( ) </s>
