code: import glob
  import os
  import sys
  import random
  import time
  import numpy as np
  import cv2
  import math
  from collections import deque
  from keras.applications.xception import Xception
  from keras.layers import Dense, GlobalAveragePooling2D
  from keras.optimizers import Adam
  from keras.models import Model
  
  '''
  Carla 패키지가 사용하는 egg파일 탐색
  '''
  try:
      sys.path.append(glob.glob('../carla/dist/carla-*%d.%d-%s.egg' % (
          sys.version_info.major,
          sys.version_info.minor,
          'win-amd64' if os.name == 'nt' else 'linux-x86_64'))[0])
  except IndexError:
      pass
  import carla
  
  SHOW_PREVIEW = False
  IM_WIDTH = 640
  IM_HEIGHT = 480
  SECONDS_PER_EPISODE = 10
  REPLAY_MEMORY_SIZE = 5_000
  MIN_REPLAY_MEMORY_SIZE = 1_000
  MINIBATCH_SIZE = 16
  PREDICTION_BATCH_SIZE = 1
  TRAINING_BATCH_SIZE = MINIBATCH_SIZE // 4
  UPDATE_TARGET_EVERY = 5
  MODEL_NAME = "Xception"
  
  MEMORY_FRACTION = 0.8
  MIN_REWARD = -200
  
  EPISODES = 100
  
  DISCOUNT = 0.99
  epsilon = 1
  EPSILON_DECAY = 0.95
  MIN_EPSILON = 0.001
  
  AGGREGATE_STATS_EVERY = 10
  
  '''
  환경 class 세팅
  '''
  class CarEnv:
      SHOW_CAM = SHOW_PREVIEW # 미리보기 여부
      STEER_AMT = 1.0
      im_width = IM_WIDTH
      im_height = IM_HEIGHT
      front_camera = None
      actor_list = []
      collision_hist = []  # collision 목록
  
      def __init__(self):
          self.client = carla.Client("localhost", 2000)
          self.client.set_timeout(2.0)
  
          # client가 켜져 있다면, world 검색 가능.
          self.world = self.client.get_world()
  
          # world에는 우리가 시뮬레이션에 액터를 새로 추가할 때 사용할 수 있는 bp 목록이 있다.
          self.blueprint_library = self.world.get_blueprint_library()
  
          # 차량 모델 지정
          self.model_3 = self.blueprint_library.filter("model3")[0]
  
      def reset(self):
          self.collision_hist = []
          self.actor_list = []
  
          # 랜덤한 위치에 차량 생성 후 actor list에 추가
          self.transform = random.choice(self.world.get_map().get_spawn_points())
          self.vehicle = self.world.spawn_actor(self.model_3, self.transform)
          self.actor_list.append(self.vehicle)
  
          # rgb Camera 센서의 bp 가져오기
          self.rgb_cam = self.blueprint_library.find('sensor.camera.rgb')
  
          # rgb Camera 센서로 입력 받은 이미지의 크기 조절
          self.rgb_cam.set_attribute("image_size_x", f"{self.im_width}")
          self.rgb_cam.set_attribute("image_size_y", f"{self.im_height}")
          self.rgb_cam.set_attribute("fov", f"110")
  
          # sensor의 위치 조정
          transform = carla.Transform(carla.Location(x=2.5, z=0.7))
  
          # 센서의 생성 및 리스트 추가.
          self.sensor = self.world.spawn_actor(self.rgb_cam, transform, attach_to=self.vehicle)
          self.actor_list.append(self.sensor)
  
          # 센서로 입력 받은 데이터를 활용하기 위해 lambda 함수 사용
          self.sensor.listen(lambda data: self.process_img(data))
  
          self.vehicle.apply_control(carla.VehicleControl(throttle=0.0, brake=0.0))
  
          '''
          차량 생성 시 차가 지면에 부딪히면 충돌이 발생. 
          또는 센서들이 초기화되고 값을 반환하는 데 시간이 걸릴 수 있음. 
          따라서 4초 정도의 대기시간을 사용.
          '''
          time.sleep(4)
  
          # collision 센서의 bp 가져오기
          colsensor = self.blueprint_library.find("sensor.other.collision")
  
          # 센서의 생성 및 리스트 추가
          self.colsensor = self.world.spawn_actor(colsensor, transform, attach_to=self.vehicle)
          self.actor_list.append(self.colsensor)
  
          # 센서로 입력 받은 데이터를 활용하기 위해 lambda 함수 사용
          self.colsensor.listen(lambda event: self.collision_data(event))
  
          while self.front_camera is None:
              time.sleep(0.01)
  
          '''
          에피소드의 실제 확인 시간 기록.
          브레이크와 스로틀이 사용되지 않는지 확인 후
          첫 번째 관찰 결과 반환.        
          '''
          self.episode_start = time.time()
          self.vehicle.apply_control(carla.VehicleControl(throttle=0.0, brake=0.0))
  
          return self.front_camera
  
      # collision data 처리
      def collision_data(self, event):
          self.collision_hist.append(event)
  
      # image data 처리
      def process_img(self, image):
          i = np.array(image.raw_data)
          #print(i.shape)
          i2 = i.reshape((self.im_height, self.im_width, 4))
          i3 = i2[:, :, :3]
          if self.SHOW_CAM:
              cv2.imshow("", i3)
              cv2.waitKey(1)
          self.front_camera = i3
  
      # action, reward, done, any_extra_info 관리
      def step(self, action):
          if action == 0:
              self.vehicle.apply_control(carla.VehicleControl(throttle=1.0, steer=-1*self.STEER_AMT))
          elif action == 1:
              self.vehicle.apply_control(carla.VehicleControl(throttle=1.0, steer= 0))
          elif action == 2:
              self.vehicle.apply_control(carla.VehicleControl(throttle=1.0, steer=1*self.STEER_AMT))
  
          v = self.vehicle.get_velocity()
          kmh = int(3.6 * math.sqrt(v.x**2 + v.y**2 + v.z**2))
  
          if len(self.collision_hist) != 0:
              done = True
              reward = -200
          elif kmh < 50:
              done = False
              reward = -1
          else:
              done = False
              reward = 1
  
          if self.episode_start + SECONDS_PER_EPISODE < time.time():
              done = True
  
          return self.front_camera, reward, done, None
  
  # 강화 학습
  class DQNAgent:
      def __init__(self):
          self.model = self.create_model()
          self.target_model = self.create_model()
          self.target_model.set_weights(self.model.get_weights())
  
          self.replay_memory = deque(maxlen=REPLAY_MEMORY_SIZE)
  
          self.tensorboard = ModifiedTensorBoard(log_dir=f"logs/{MODEL_NAME}-{int(time.time())}")
          self.target_update_counter = 0
          self.graph = tf.get_default_graph()
  
          self.terminate = False
          self.last_logged_episode = 0
          self.training_initialized = False
  
      # 모델 생성
      def create_model(self):
          base_model = Xception(weights=None, include_top=False, input_shape=(IM_HEIGHT, IM_WIDTH,3))
  
          x = base_model.output
          x = GlobalAveragePooling2D()(x)
  
          predictions = Dense(3, activation="linear")(x)
          model = Model(inputs=base_model.input, outputs=predictions)
          model.compile(loss="mse", optimizer=Adam(lr=0.001), metrics=["accuracy"])
          return model
  
  
      def update_replay_memory(self, transition):
          # transition = (current_state, action, reward, new_state, done)
          self.replay_memory.append(transition)
  
      def train(self):
          if len(self.replay_memory) < MIN_REPLAY_MEMORY_SIZE:
              return
  
          minibatch = random.sample(self.replay_memory, MINIBATCH_SIZE)
  
          current_states = np.array([transition[0] for transition in minibatch])/255
          with self.graph.as_default():
              current_qs_list = self.model.predict(current_states, PREDICTION_BATCH_SIZE)
  
          new_current_states = np.array([transition[3] for transition in minibatch])/255
          with self.graph.as_default():
              future_qs_list = self.target_model.predict(new_current_states, PREDICTION_BATCH_SIZE)
  
          # x = input / y = output
          X = []
          y = []
  
          for index, (current_state, action, reward, new_state, done) in enumerate(minibatch):
              if not done:
                  max_future_q = np.max(future_qs_list[index])
                  new_q = reward + DISCOUNT * max_future_q
              else:
                  new_q = reward
  
              current_qs = current_qs_list[index]
              current_qs[action] = new_q
  
              X.append(current_state)
              y.append(current_qs)
  
          '''
          step 단위가 아니라 episode 단위로 log 기록
          log_this_step이 true일 때만 TensorBoard에 log 기록
          '''
          log_this_step = False
          if self.tensorboard.step > self.last_logged_episode:
              log_this_step = True
              self.last_log_episode = self.tensorboard.step
  
          with self.graph.as_default():
              self.model.fit(np.array(X)/255, np.array(y), batch_size=TRAINING_BATCH_SIZE, verbose=0, shuffle=False, callbacks=[self.tensorboard] if log_this_step else None)
  
  
          if log_this_step:
              self.target_update_counter += 1
  
          # target_model 업데이트 여부 확인
          if self.target_update_counter > UPDATE_TARGET_EVERY:
              self.target_model.set_weights(self.model.get_weights())
              self.target_update_counter = 0
  
      def get_qs(self, state):
          return self.model.predict(np.array(state).reshape(-1 *state.shape)/255)[0]
  
      # train 진행
      def train_in_loop(self):
          X = np.random.uniform(size=(1, IM_HEIGHT, IM_WIDTH, 3)).astype(np.float32)
          y = np.random.uniform(size=(1, 3)).astype(np.float32)
          with self.graph.as_default():
              self.model.fit(X,y, verbose=False, batch_size=1)
  
          self.training_initialized = True
  
          while True:
              if self.terminate:
                  return
              self.train()
              time.sleep(0.01)
apis: numpy.array numpy.random.uniform numpy.max