Gym Donkey Car

https://gym-donkeycar.readthedocs.io/en/latest
Admin
Gym Donkey Car provides OpenAI Gym environments for training and simulating autonomous driving...
Tokens:10,247
Snippets:60
Trust Score:8.3
Update:10 months ago
Show doc for...
Tokens:
Raw
### Install Gym Donkey Car from Source using setup.py

Source: https://gym-donkeycar.readthedocs.io/en/latest/_sources/installation.rst

Installs the gym-donkeycar package after downloading the source code (either by cloning or downloading a tarball). This command executes the setup script for installation. Requires Python.

```shell
python setup.py install
```

--------------------------------

### Install Gym Donkey Car from Local Source

Source: https://gym-donkeycar.readthedocs.io/en/latest/installation

Installs the OpenAI Gym Environments for Donkey Car after downloading the source code (either via clone or tarball). This command should be run from the root of the source directory.

```bash
$ python setup.py install

```

--------------------------------

### Install Stable Release using Pip

Source: https://gym-donkeycar.readthedocs.io/en/latest/installation

Installs the most recent stable release of OpenAI Gym Environments for Donkey Car using pip. This is the recommended installation method. Ensure pip is installed; refer to the Python installation guide if needed.

```bash
$ pip install gym_donkeycar

```

--------------------------------

### Install Gym Donkey Car from Source using Tarball

Source: https://gym-donkeycar.readthedocs.io/en/latest/_sources/installation.rst

Installs the gym-donkeycar package from a tarball downloaded from the GitHub repository. This method is an alternative to cloning the repository. Requires curl.

```shell
curl  -OL https://github.com/leigh-johnson/gym_donkeycar/tarball/master
```

--------------------------------

### Install Gym Donkey Car from Source using Git Clone

Source: https://gym-donkeycar.readthedocs.io/en/latest/_sources/installation.rst

Installs the gym-donkeycar package from its source code by cloning the GitHub repository. This method is useful for developers or when the latest unreleased changes are needed. Requires Git.

```shell
git clone git://github.com/leigh-johnson/gym_donkeycar
```

--------------------------------

### Install Gym Donkey Car using pip

Source: https://gym-donkeycar.readthedocs.io/en/latest/_sources/installation.rst

Installs the latest stable release of the gym-donkeycar package using pip. This is the recommended installation method. Ensure pip is installed on your system.

```shell
pip install gym_donkeycar
```

--------------------------------

### DonkeyUnityProcess Start Method - Python

Source: https://gym-donkeycar.readthedocs.io/en/latest/gym_donkeycar.envs

Starts the Donkey Unity simulator process. It allows specifying the path to the simulator executable, whether to run in headless mode, and the communication port. This method is crucial for initializing and connecting to the Donkeycar simulation environment.

```python
class DonkeyUnityProcess(object):
    ...
    def start(self, sim_path, headless=False, port=9090):
        pass
```

--------------------------------

### Download Donkey Car Gym Tarball

Source: https://gym-donkeycar.readthedocs.io/en/latest/installation

Downloads the master tarball of the OpenAI Gym Environments for Donkey Car source code from GitHub. This is an alternative method for obtaining the source code before installation.

```bash
$ curl  -OL https://github.com/leigh-johnson/gym_donkeycar/tarball/master

```

--------------------------------

### Clone Donkey Car Gym from GitHub Source

Source: https://gym-donkeycar.readthedocs.io/en/latest/installation

Clones the public repository of OpenAI Gym Environments for Donkey Car from GitHub. This method allows installation directly from the source code.

```bash
$ git clone git://github.com/leigh-johnson/gym_donkeycar

```

--------------------------------

### Install DonkeyCar Gym

Source: https://gym-donkeycar.readthedocs.io/en/latest/readme

Installs the DonkeyCar OpenAI Gym package using pip. This is the primary method for setting up the library for use.

```shell
pip install gym-donkeycar
```

--------------------------------

### Launch and Manage Donkey Car Unity Environment (Python)

Source: https://gym-donkeycar.readthedocs.io/en/latest/_modules/gym_donkeycar/envs/donkey_proc

This Python code defines a class to manage the Donkey Car Unity simulation process. It uses the 'subprocess' module to launch the Unity executable and the 'os' module to check for file existence. The 'start' method takes the path to the simulation, optional headless mode, and port number as input, launching the process and printing a confirmation. The 'quit' method terminates the simulation process if it is running.

```python
'''
file: donkey_proc.py
author: Felix Yu
date: 2018-09-12
'''
import subprocess
import os



[docs] class DonkeyUnityProcess(object):

    def __init__(self):
        self.proc1 = None

    ## ------ Launch Unity Env ----------- ##


[docs]    def start(self, sim_path, headless=False, port=9090):

        if not os.path.exists(sim_path):
            print(sim_path, "does not exist. you must start sim manually.")
            return

        port_args = ["--port", str(port), '-logFile', 'unitylog.txt']

        # Launch Unity environment
        if headless:
            self.proc1 = subprocess.Popen(
                [sim_path, '-nographics', '-batchmode'] + port_args)
        else:
            self.proc1 = subprocess.Popen(
                [sim_path] + port_args)

        print("donkey subprocess started")



[docs]    def quit(self):
        """
        Shutdown unity environment
        """
        if self.proc1 is not None:
            print("closing donkey sim subprocess")
            self.proc1.kill()
            self.proc1 = None

```

--------------------------------

### Import OpenAI Gym Environment for Donkey Car (Python)

Source: https://gym-donkeycar.readthedocs.io/en/latest/usage

This code snippet demonstrates the necessary import statement to begin using OpenAI Gym environments tailored for Donkey Car. Ensure the 'gym-donkeycar' library is installed in your Python environment before executing this import.

```python
import gym_donkeycar

```

--------------------------------

### Determine Episode Over Condition (Python)

Source: https://gym-donkeycar.readthedocs.io/en/latest/_modules/gym_donkeycar/envs/donkey_sim

Evaluates conditions to determine if the current training episode should end. It checks for excessive cross-track error (CTE) and collision events ('hit'). Initial large CTE values shortly after starting are ignored. Dependencies include the 'math' module and a logger.

```python
# we have a few initial frames on start that are sometimes very large CTE when it's behind
# the path just slightly. We ignore those.
if math.fabs(self.cte) > 2 * self.max_cte:
    pass
elif math.fabs(self.cte) > self.max_cte:
    logger.debug(f"game over: cte {self.cte}")
    self.over = True
elif self.hit != "none":
    logger.debug(f"game over: hit {self.hit}")
    self.over = True
```

--------------------------------

### Create Donkey Car Gym Environment

Source: https://gym-donkeycar.readthedocs.io/en/latest/readme

Demonstrates how to create a Donkey Car environment using the gym library. It initializes a specific environment, 'donkey-generated-track-v0', for use in reinforcement learning tasks.

```python
import gym

env = gym.make("donkey-generated-track-v0)
```

--------------------------------

### Initialize DonkeyCar Gym Environment (Python)

Source: https://gym-donkeycar.readthedocs.io/en/latest/_modules/gym_donkeycar/envs/donkey_env

Initializes the DonkeyCar OpenAI Gym environment, setting up the connection to the Donkey Car simulator. It configures action and observation spaces, handles environment variables for simulator path and port, and waits for the simulator to load. Dependencies include gym, numpy, and specific modules from gym_donkeycar.

```python
'''
file: donkey_env.py
author: Tawn Kramer
date: 2018-08-31
'''
import os
import random
import time

import numpy as np
import gym
from gym import spaces
from gym.utils import seeding
from gym_donkeycar.envs.donkey_sim import DonkeyUnitySimContoller
from gym_donkeycar.envs.donkey_proc import DonkeyUnityProcess



class DonkeyEnv(gym.Env):
    """
    OpenAI Gym Environment for Donkey
    """

    metadata = {
        "render.modes": ["human", "rgb_array"],
    }

    ACTION_NAMES = ["steer", "throttle"]
    STEER_LIMIT_LEFT = -1.0
    STEER_LIMIT_RIGHT = 1.0
    THROTTLE_MIN = 0.0
    THROTTLE_MAX = 5.0
    VAL_PER_PIXEL = 255

    def __init__(self, level, time_step=0.05, frame_skip=2):

        print("starting DonkeyGym env")

        # start Unity simulation subprocess
        self.proc = DonkeyUnityProcess()

        try:
            exe_path = os.environ['DONKEY_SIM_PATH']
        except:
            print("Missing DONKEY_SIM_PATH environment var. you must start sim manually")
            exe_path = "self_start"

        try:
            port_offset = 0
            # if more than one sim running on same machine set DONKEY_SIM_MULTI = 1
            random_port = os.environ['DONKEY_SIM_MULTI'] == '1'
            if random_port:
                port_offset = random.randint(0, 1000)
        except:
            pass

        try:
            port = int(os.environ['DONKEY_SIM_PORT']) + port_offset
        except:
            port = 9091 + port_offset
            print("Missing DONKEY_SIM_PORT environment var. Using default:", port)

        try:
            headless = os.environ['DONKEY_SIM_HEADLESS'] == '1'
        except:
            print("Missing DONKEY_SIM_HEADLESS environment var. Using defaults")
            headless = False

        self.proc.start(exe_path, headless=headless, port=port)

        # start simulation com
        self.viewer = DonkeyUnitySimContoller(
            level=level, time_step=time_step, port=port)

        # steering and throttle
        self.action_space = spaces.Box(low=np.array([self.STEER_LIMIT_LEFT, self.THROTTLE_MIN]),
                                       high=np.array([self.STEER_LIMIT_RIGHT, self.THROTTLE_MAX]), dtype=np.float32)

        # camera sensor data
        self.observation_space = spaces.Box(
            0, self.VAL_PER_PIXEL, self.viewer.get_sensor_size(), dtype=np.uint8)

        # simulation related variables.
        self.seed()

        # Frame Skipping
        self.frame_skip = frame_skip

        # wait until loaded
        self.viewer.wait_until_loaded()

    def __del__(self):
        self.close()


    def close(self):
        self.proc.quit()


    def seed(self, seed=None):
        self.np_random, seed = seeding.np_random(seed)
        return [seed]


    def step(self, action):
        for i in range(self.frame_skip):
            self.viewer.take_action(action)
            observation, reward, done, info = self.viewer.observe()
        return observation, reward, done, info


    def reset(self):
        self.viewer.reset()
        observation, reward, done, info = self.viewer.observe()
        time.sleep(1)
        return observation


    def render(self, mode="human", close=False):
        if close:
            self.viewer.quit()

        return self.viewer.render(mode)


    def is_game_over(self):
        return self.viewer.is_game_over()

```

--------------------------------

### Handle Scene Selection Ready Event (Python)

Source: https://gym-donkeycar.readthedocs.io/en/latest/_modules/gym_donkeycar/envs/donkey_sim

Callback function executed when the scene selection is ready. It logs the event and triggers the retrieval of available scene names. No input data is directly used other than logging the event.

```python
logger.debug("SceneSelectionReady ")
self.send_get_scene_names()
```

--------------------------------

### gym_donkeycar.core.fps Module

Source: https://gym-donkeycar.readthedocs.io/en/latest/gym_donkeycar.core

Provides functionality for timing frames per second using the FPSTimer class.

```APIDOC
## FPSTimer Class

### Description
Manages frame rate timing for the simulation.

### Methods
- `on_frame()`: Records the occurrence of a frame.
- `reset()`: Resets the frame rate timer.
```

--------------------------------

### SimServer: TCP Server for Donkey Car Simulator

Source: https://gym-donkeycar.readthedocs.io/en/latest/_modules/gym_donkeycar/core/tcp_server

The SimServer class establishes a TCP socket server that listens for incoming connections from the Unity donkey simulator. It binds to a specified address, listens for connections, and creates a new SimHandler for each client that connects. It also manages the server's lifecycle and notifies the message handler on shutdown.

```python
class SimServer(asyncore.dispatcher):
    """
      Receives network connections and establishes handlers for each client.
      Each client connection is handled by a new instance of the SteeringHandler class.
    """

    def __init__(self, address, msg_handler):
        asyncore.dispatcher.__init__(self)

        # create a TCP socket to listen for connections
        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)

        # in case we have shutdown recently, allow the os to reuse this address.
        # helps when restarting
        self.set_reuse_addr()

        # let TCP stack know that we'd like to sit on this address and listen for connections
        self.bind(address)

        # confirm for users what address we are listening on
        self.address = self.socket.getsockname()
        print('binding to', self.address)

        # let tcp stack know we plan to process one outstanding request to connect request each loop
        self.listen(5)

        # keep a pointer to our IMesgHandler handler
        self.msg_handler = msg_handler


    def handle_accept(self):
        # Called when a client connects to our socket
        client_info = self.accept()

        print('got a new client', client_info[1])

        # make a new steering handler to communicate with the client
        SimHandler(sock=client_info[0], msg_handler=self.msg_handler)



    def handle_close(self):
        print("server shutdown")
        # Called then server is shutdown
        self.close()

        if self.msg_handler:
            self.msg_handler.on_close()

```

--------------------------------

### gym_donkeycar.core.tcp_server Module

Source: https://gym-donkeycar.readthedocs.io/en/latest/gym_donkeycar.core

Implements a TCP socket server for communication with the Donkey Car simulator in Unity.

```APIDOC
## IMesgHandler Interface

### Description
An interface for handling messages received over the TCP server.

### Methods
- `on_close()`: Called when a client disconnects.
- `on_connect(socketHandler)`: Called when a new client connects.
- `on_disconnect()`: Called when a client disconnects.
- `on_recv_message(message)`: Called when a message is received from a client.
```

```APIDOC
## SimHandler Class

### Description
Handles messages from a single TCP client connection.

### Inheritance
Bases: `asyncore.dispatcher`

### Methods
- `handle_close()`: Handles the closing of the client socket.
- `handle_json_message(chunk)`: Processes incoming JSON messages.
- `handle_read()`: Reads data from the client socket.
- `handle_write()`: Writes data to the client socket.
- `queue_message(msg)`: Adds a message to the outgoing queue.
- `writable()`: Returns True if the socket is ready for writing.
```

```APIDOC
## SimServer Class

### Description
Manages network connections and establishes handlers for each client.

### Inheritance
Bases: `asyncore.dispatcher`

### Parameters
- `address` (tuple): The address and port to listen on.
- `msg_handler` (IMesgHandler): An instance of a message handler.

### Methods
- `handle_accept()`: Accepts new incoming connections.
- `handle_close()`: Handles the closing of the server socket.
```

```APIDOC
## replace_float_notation Function

### Description
Replaces Unity's float notation (e.g., comma as decimal separator) with standard dot notation to ensure valid JSON parsing.

### Parameters
- `string` (str): The potentially incorrectly formatted JSON string from Unity.

### Returns
- (str): A valid JSON string with standard float notation.
```

--------------------------------

### Receive and Load Scene Names (Python)

Source: https://gym-donkeycar.readthedocs.io/en/latest/_modules/gym_donkeycar/envs/donkey_sim

Processes the received list of scene names from the simulation. If names are provided, it logs them and initiates the loading of the first scene in the list based on an internal index. Dependencies include a logger.

```python
if data:
    names = data['scene_names']
    logger.debug(f"SceneNames: {names}")
    self.send_load_scene(names[self.iSceneToLoad])
```

--------------------------------

### DonkeyUnitySimHandler Class

Source: https://gym-donkeycar.readthedocs.io/en/latest/gym_donkeycar.envs

Handles messages and communication with the Donkey Car Unity simulator.

```APIDOC
## DonkeyUnitySimHandler Class

### Description
This class extends `gym_donkeycar.core.tcp_server.IMesgHandler` to manage communication with the Donkey Car Unity simulator via TCP. It processes incoming messages, sends commands, and handles various simulation events.

### Methods
- `__init__(self, level, time_step=0.05, max_cte=5.0, cam_resolution=None)`: Initializes the message handler.
- `calc_reward(self, done)`: Calculates the reward for the current simulation step.
- `determine_episode_over(self)`: Determines if the current episode is over.
- `get_sensor_size(self)`: Returns the size of the sensor data.
- `is_game_over(self)`: Checks if the simulation has ended.
- `observe(self)`: Retrieves the current observation from the simulator.
- `on_car_loaded(self, data)`: Callback function when the car is loaded.
- `on_connect(self, socketHandler)`: Callback function upon connection.
- `on_disconnect(self)`: Callback function upon disconnection.
- `on_recv_message(self, message)`: Processes a received message.
- `on_recv_scene_names(self, data)`: Callback function to receive scene names.
- `on_scene_selection_ready(self, data)`: Callback function when scene selection is ready.
- `on_telemetry(self, data)`: Callback function to receive telemetry data.
- `queue_message(self, msg)`: Queues a message to be sent.
- `reset(self)`: Resets the simulation.
- `send_control(self, steer, throttle)`: Sends steering and throttle control commands.
- `send_get_scene_names(self)`: Sends a request to get available scene names.
- `send_load_scene(self, scene_name)`: Sends a command to load a specific scene.
- `send_reset_car(self)`: Sends a command to reset the car.
- `take_action(self, action)`: Takes a specified action in the simulation.
```

--------------------------------

### Create AvcSparkfun Environment (Python)

Source: https://gym-donkeycar.readthedocs.io/en/latest/_modules/gym_donkeycar/envs/donkey_env

Creates an instance of the DonkeyCar environment using the AvcSparkfun track. This class inherits from DonkeyEnv and initializes the simulation with level 2, specific to the AvcSparkfun track. It relies on the default time step and frame skip settings from the parent class.

```python
class AvcSparkfunEnv(DonkeyEnv):

    def __init__(self):
        super(AvcSparkfunEnv, self).__init__(level=2)

```

--------------------------------

### SimHandler: Client Message Handling for TCP Server

Source: https://gym-donkeycar.readthedocs.io/en/latest/_modules/gym_donkeycar/core/tcp_server

The SimHandler class manages communication with a single connected TCP client. It handles the reading of incoming data, queuing of outgoing messages, and notifies the message handler of connection events. It buffers data to read and data to write, processing them in chunks.

```python
class SimHandler(asyncore.dispatcher):
    """
      Handles messages from a single TCP client.
    """

    def __init__(self, sock, chunk_size=(16 * 1024), msg_handler=None):
        # we call our base class init
        asyncore.dispatcher.__init__(self, sock=sock)

        # msg_handler handles incoming messages
        self.msg_handler = msg_handler

        if msg_handler:
            msg_handler.on_connect(self)

        # chunk size is the max number of bytes to read per network packet
        self.chunk_size = chunk_size

        # we make an empty list of packets to send to the client here
        self.data_to_write = []

        # and image bytes is an empty list of partial bytes of the image as it comes in
        self.data_to_read = []


    def writable(self):
        """
          We want to write if we have received data.
        """
        response = bool(self.data_to_write)

        return response



    def queue_message(self, msg):
        json_msg = json.dumps(msg)
        self.data_to_write.append(json_msg)



    def handle_write(self):
        """
          Write as much as possible of the most recent message we have received.
          This is only called by async manager when the socket is in a writable state
          and when self.writable return true, that yes, we have data to send.
        """

        # pop the first element from the list. encode will make it into a byte stream
        data = self.data_to_write.pop(0).encode()

        # send a slice of that data, up to a max of the chunk_size
        sent = self.send(data[:self.chunk_size])

        # if we didn't send all the data..
        if sent < len(data):
            # then slice off the portion that remains to be sent
            remaining = data[sent:]

            # since we've popped it off the list, add it back to the list to send next
            # probably should change this to a deque...
            self.data_to_write.insert(0, remaining)



    def handle_read(self):
        """
          Read an incoming message from the client and put it into our outgoing queue.

```

--------------------------------

### Send Load Scene Command (Python)

Source: https://gym-donkeycar.readthedocs.io/en/latest/_modules/gym_donkeycar/envs/donkey_sim

Instructs the DonkeyCar simulator to load a specified scene. The scene name is provided as an argument and included in the message payload. Dependencies: A logger for debug messages.

```python
msg = {'msg_type': 'load_scene', 'scene_name': scene_name}
self.queue_message(msg)
```

--------------------------------

### DonkeyUnitySimContoller Class for Unity Simulation

Source: https://gym-donkeycar.readthedocs.io/en/latest/_modules/gym_donkeycar/envs/donkey_sim

Manages the connection and interaction with the Donkey Unity simulator. It initializes the server, handler, and a separate thread for the simulation loop. Key methods include waiting for the sim to load, resetting the environment, taking actions, observing the state, and checking game over conditions.

```python
class DonkeyUnitySimContoller():

    def __init__(self, level, time_step=0.05, hostname='0.0.0.0',
                 port=9090, max_cte=5.0, loglevel='INFO', cam_resolution=(120, 160, 3)):

        logger.setLevel(loglevel)

        self.address = (hostname, port)

        self.handler = DonkeyUnitySimHandler(
            level, time_step=time_step, max_cte=max_cte,
            cam_resolution=cam_resolution)

        try:
            self.server = SimServer(self.address, self.handler)
        except OSError:
            raise SimFailed("failed to listen on address %s" % self.address)

        self.thread = Thread(target=asyncore.loop)
        self.thread.daemon = True
        self.thread.start()


    def wait_until_loaded(self):
        while not self.handler.loaded:
            logger.warning("waiting for sim to start..")
            time.sleep(3.0)


    def reset(self):
        self.handler.reset()


    def get_sensor_size(self):
        return self.handler.get_sensor_size()


    def take_action(self, action):
        self.handler.take_action(action)


    def observe(self):
        return self.handler.observe()


    def quit(self):
        pass


    def render(self, mode):
        pass


    def is_game_over(self):
        return self.handler.is_game_over()


    def calc_reward(self, done):
        return self.handler.calc_reward(done)

```

--------------------------------

### Create Warehouse Environment (Python)

Source: https://gym-donkeycar.readthedocs.io/en/latest/_modules/gym_donkeycar/envs/donkey_env

Creates an instance of the DonkeyCar environment using a warehouse track. This class inherits from DonkeyEnv and initializes the environment with level 1, corresponding to the warehouse track. It utilizes the default simulation parameters from the DonkeyEnv base class.

```python
class WarehouseEnv(DonkeyEnv):

    def __init__(self):
        super(WarehouseEnv, self).__init__(level=1)

```

--------------------------------

### Handle Car Loaded Event (Python)

Source: https://gym-donkeycar.readthedocs.io/en/latest/_modules/gym_donkeycar/envs/donkey_sim

Callback function executed when the car model has been successfully loaded. It sets an internal 'loaded' flag to true, indicating the car is ready for operation. Logs the car loading event.

```python
logger.debug("car loaded")
self.loaded = True
```

--------------------------------

### DonkeyUnitySimContoller Class

Source: https://gym-donkeycar.readthedocs.io/en/latest/gym_donkeycar.envs

Controls the Donkey Car Unity simulator, allowing for actions, observation, and simulation resets.

```APIDOC
## DonkeyUnitySimContoller Class

### Description
This class provides an interface to control the Donkey Car Unity simulator. It allows users to send actions, receive observations, calculate rewards, and manage the simulation state.

### Methods
- `__init__(self, level, time_step=0.05, hostname='0.0.0.0', port=9090, max_cte=5.0, loglevel='INFO', cam_resolution=(120, 160, 3))`: Initializes the simulator controller.
- `calc_reward(self, done)`: Calculates the reward for the current simulation step.
- `get_sensor_size(self)`: Returns the size of the sensor data.
- `is_game_over(self)`: Checks if the simulation has ended.
- `observe(self)`: Retrieves the current observation from the simulator.
- `quit(self)`: Quits the simulator.
- `render(self, mode)`: Renders the simulation view.
- `reset(self)`: Resets the simulation to its initial state.
- `take_action(self, action)`: Takes a specified action in the simulation.
- `wait_until_loaded(self)`: Waits until the simulator environment is loaded.
```

--------------------------------

### Create Generated Track Environment (Python)

Source: https://gym-donkeycar.readthedocs.io/en/latest/_modules/gym_donkeycar/envs/donkey_env

Creates an instance of the DonkeyCar environment with a generated track configuration. This class inherits from DonkeyEnv and initializes with level 3, representing a generated track. It uses the default time step and frame skip parameters provided by the DonkeyEnv.

```python
class GeneratedTrackEnv(DonkeyEnv):

    def __init__(self):
        super(GeneratedTrackEnv, self).__init__(level=3)

```

--------------------------------

### Create Generated Roads Environment (Python)

Source: https://gym-donkeycar.readthedocs.io/en/latest/_modules/gym_donkeycar/envs/donkey_env

Creates an instance of the DonkeyCar environment using a generated road track. This class inherits from DonkeyEnv and initializes the environment with level 0, representing a procedurally generated track. It uses the default time step and frame skip values from the parent class.

```python
class GeneratedRoadsEnv(DonkeyEnv):

    def __init__(self):
        super(GeneratedRoadsEnv, self).__init__(level=0)

```

--------------------------------

### Handle Incoming Data from Client Socket

Source: https://gym-donkeycar.readthedocs.io/en/latest/_modules/gym_donkeycar/core/tcp_server

Processes incoming data from a client socket, decodes it, and handles JSON messages or appends raw data. It manages connection drops and ensures proper message parsing.

```python
def handle_read(self):
        """
        handle_read should only be called when the given socket has data ready to be
        processed.
        """

        # receive a chunk of data with the max size chunk_size from our client.
        data = self.recv(self.chunk_size)

        if len(data) == 0:
            # this only happens when the connection is dropped
            self.handle_close()
            return

        self.data_to_read.append(data.decode("utf-8"))

        messages = ''.join(self.data_to_read).split('\n')

        self.data_to_read = []

        for mesg in messages:
            if len(mesg) < 2:
                continue
            if mesg[0] == '{' and mesg[-1] == '}':
                self.handle_json_message(mesg)
            else:
                self.data_to_read.append(mesg)
```

--------------------------------

### DonkeyUnitySimHandler for Message Handling and State Management

Source: https://gym-donkeycar.readthedocs.io/en/latest/_modules/gym_donkeycar/envs/donkey_sim

Implements the IMesgHandler interface to process messages from the Unity simulator and manage the car's state. It handles telemetry data, scene loading, and car loading events. This class also defines methods for resetting the simulation, capturing observations, and calculating rewards based on game state and car performance.

```python
class DonkeyUnitySimHandler(IMesgHandler):

    def __init__(self, level, time_step=0.05, max_cte=5.0, cam_resolution=None):
        self.iSceneToLoad = level
        self.time_step = time_step
        self.wait_time_for_obs = 0.1
        self.sock = None
        self.loaded = False
        self.max_cte = max_cte
        self.timer = FPSTimer()

        self.camera_img_size = cam_resolution
        self.image_array = np.zeros(self.camera_img_size)
        self.last_obs = None
        self.hit = "none"
        self.cte = 0.0
        self.x = 0.0
        self.y = 0.0
        self.z = 0.0
        self.speed = 0.0
        self.over = False
        self.fns = {'telemetry': self.on_telemetry,
                    "scene_selection_ready": self.on_scene_selection_ready,
                    "scene_names": self.on_recv_scene_names,
                    "car_loaded": self.on_car_loaded}


    def on_connect(self, socketHandler):
        self.sock = socketHandler


    def on_disconnect(self):
        self.sock = None


    def on_recv_message(self, message):
        if 'msg_type' not in message:
            logger.error('expected msg_type field')
            return

        msg_type = message['msg_type']
        if msg_type in self.fns:
            self.fns[msg_type](message)
        else:
            logger.warning(f'unknown message type {msg_type}')


    def reset(self):
        logger.debug("reseting")
        self.image_array = np.zeros(self.camera_img_size)
        self.last_obs = self.image_array
        self.hit = "none"
        self.cte = 0.0
        self.x = 0.0
        self.y = 0.0
        self.z = 0.0
        self.speed = 0.0
        self.over = False
        self.send_reset_car()
        self.timer.reset()
        time.sleep(1)


    def get_sensor_size(self):
        return self.camera_img_size


    def take_action(self, action):
        self.send_control(action[0], action[1])


    def observe(self):
        while self.last_obs is self.image_array:
            time.sleep(1.0 / 120.0)

        self.last_obs = self.image_array
        observation = self.image_array
        done = self.is_game_over()
        reward = self.calc_reward(done)
        info = {'pos': (self.x, self.y, self.z), 'cte': self.cte,
                "speed": self.speed, "hit": self.hit}

        self.timer.on_frame()

        return observation, reward, done, info


    def is_game_over(self):
        return self.over


    def calc_reward(self, done):
        if done:
            return -1.0

        if self.cte > self.max_cte:
            return -1.0

        if self.hit != "none":
            return -2.0

        return 1.0 - (self.cte / self.max_cte) * self.speed


    def on_telemetry(self, data):


```

--------------------------------

### Request Scene Names from Simulator (Python)

Source: https://gym-donkeycar.readthedocs.io/en/latest/_modules/gym_donkeycar/envs/donkey_sim

Sends a request to the DonkeyCar simulator to retrieve a list of available scene names. This is a fundamental step before loading a specific scene for an episode. The message is formatted as a dictionary.

```python
msg = {'msg_type': 'get_scene_names'}
self.queue_message(msg)
```

--------------------------------

### FPSTimer Class for FPS Calculation (Python)

Source: https://gym-donkeycar.readthedocs.io/en/latest/_modules/gym_donkeycar/core/fps

The FPSTimer class calculates and prints the frames per second (FPS) every 100 iterations. It requires the 'time' module for its operations. The primary methods are __init__, reset, and on_frame.

```python
import time



[docs]class FPSTimer(object):
    def __init__(self):
        self.t = time.time()
        self.iter = 0


[docs]    def reset(self):
        self.t = time.time()
        self.iter = 0



[docs]    def on_frame(self):
        self.iter += 1
        if self.iter == 100:
            e = time.time()
            print('fps', 100.0 / (e - self.t))
            self.t = time.time()
            self.iter = 0
```

--------------------------------

### IMesgHandler: Interface for Message Handling

Source: https://gym-donkeycar.readthedocs.io/en/latest/_modules/gym_donkeycar/core/tcp_server

The IMesgHandler class defines an abstract interface for handling messages and connection events within the TCP server. Concrete implementations of this class will provide specific logic for processing received messages, managing client connections, and handling server closure.

```python
class IMesgHandler(object):


    def on_connect(self, socketHandler):
        pass



    def on_recv_message(self, message):
        pass



    def on_close(self):
        pass



    def on_disconnect(self):
        pass

```

--------------------------------

### Send Vehicle Control Command (Python)

Source: https://gym-donkeycar.readthedocs.io/en/latest/_modules/gym_donkeycar/envs/donkey_sim

Transmits a control command to the DonkeyCar simulation, including steering, throttle, and brake values. The command is only sent if the car has been loaded. Steering, throttle, and brake are converted to strings before being included in the message. Dependencies: `__str__` method for steer and throttle inputs.

```python
if not self.loaded:
    return
msg = {'msg_type': 'control', 'steering': steer.__str__(
          ), 'throttle': throttle.__str__(), 'brake': '0.0'}
self.queue_message(msg)
```

--------------------------------

### Queue Message for Sending (Python)

Source: https://gym-donkeycar.readthedocs.io/en/latest/_modules/gym_donkeycar/envs/donkey_sim

Manages the queuing of messages to be sent to the DonkeyCar simulator via a socket connection. If the socket is not available, the message is logged and skipped. Otherwise, it's logged and sent using the socket's queue_message method. Dependencies: A logger and a socket object with a 'queue_message' method.

```python
if self.sock is None:
    logger.debug(f'skiping: \n {msg}')
    return

logger.debug(f'sending \n {msg}')
self.sock.queue_message(msg)
```

--------------------------------

### DonkeyUnityProcess Quit Method - Python

Source: https://gym-donkeycar.readthedocs.io/en/latest/gym_donkeycar.envs

Shuts down the Donkey Unity simulator process. This method ensures that the simulator is properly terminated, releasing any resources it may be using. It's important to call this method to cleanly exit the simulation.

```python
class DonkeyUnityProcess(object):
    ...
    def quit(self):
        """Shutdown unity environment"""
        pass
```

--------------------------------

### Handle Client Disconnection

Source: https://gym-donkeycar.readthedocs.io/en/latest/_modules/gym_donkeycar/core/tcp_server

Manages the closing of a client connection. It notifies the message handler of the disconnection and cleans up resources.

```python
def handle_close(self):
        # when client drops or closes connection
        if self.msg_handler:
            self.msg_handler.on_disconnect()
            self.msg_handler = None
            print('connection dropped')

        self.close()
```

--------------------------------

### Send Car Reset Command (Python)

Source: https://gym-donkeycar.readthedocs.io/en/latest/_modules/gym_donkeycar/envs/donkey_sim

Sends a command to reset the DonkeyCar's state within the simulation. This action is typically used to restart an episode or reposition the car. The message format is a simple dictionary indicating the reset action.

```python
msg = {'msg_type': 'reset_car'}
self.queue_message(msg)
```

--------------------------------

### DonkeyEnv Class Definition - Python

Source: https://gym-donkeycar.readthedocs.io/en/latest/gym_donkeycar.envs

Defines the base DonkeyEnv class, inheriting from gym.core.Env. It sets up fundamental parameters for the Donkeycar environment, including action names, steering and throttle limits, and rendering modes. This class serves as the foundation for all specific Donkeycar environments.

```python
class DonkeyEnv(gym.core.Env):
    """OpenAI Gym Environment for Donkey"""
    ACTION_NAMES = ['steer', 'throttle']
    STEER_LIMIT_LEFT = -1.0
    STEER_LIMIT_RIGHT = 1.0
    THROTTLE_MAX = 5.0
    THROTTLE_MIN = 0.0
    VAL_PER_PIXEL = 255

    def close(self):
        """Override close in your subclass to perform any necessary cleanup. Environments will automatically close() themselves when garbage collected or when the program exits."""
        pass

    def is_game_over(self):
        pass

    metadata = {'render.modes': ['human', 'rgb_array']}

    def render(self, mode='human'):
        """Renders the environment.
        The set of supported modes varies per environment. (And some environments do not support rendering at all.) By convention, if mode is:
          * human: render to the current display or terminal and return nothing. Usually for human consumption.
          * rgb_array: Return an numpy.ndarray with shape (x, y, 3), representing RGB values for an x-by-y pixel image, suitable for turning into a video.
          * ansi: Return a string (str) or StringIO.StringIO containing a terminal-style text representation. The text can include newlines and ANSI escape sequences (e.g. for colors).

        Note:
            Make sure that your class’s metadata ‘render.modes’ key includes
            the list of supported modes. It’s recommended to call super() in implementations to use the functionality of this method.

        Args:
            mode (str): the mode to render with
        Example: 

        class MyEnv(Env):
            metadata = {‘render.modes’: [‘human’, ‘rgb_array’]}
            def render(self, mode=’human’):
                if mode == ‘rgb_array’:
                    return np.array(…) # return RGB frame suitable for video 
                elif mode == ‘human’:
                    … # pop up a window and render 
                else:
                    super(MyEnv, self).render(mode=mode) # just raise an exception 
        """
        pass

    def reset(self):
        """Resets the state of the environment and returns an initial observation. 

        Returns: 
            observation (object): the initial observation.
        """
        pass

    def seed(self, seed=None):
        """Sets the seed for this env’s random number generator(s).

        Note:
            Some environments use multiple pseudorandom number generators. We want to capture all such seeds used in order to ensure that there aren’t accidental correlations between multiple generators.

        Returns:
            list<bigint>: Returns the list of seeds used in this env’s random
            number generators. The first value in the list should be the “main” seed, or the value which a reproducer should pass to ‘seed’. Often, the main seed equals the provided ‘seed’, but this won’t be true if seed=None, for example.
        """
        pass

    def step(self, action):
        """Run one timestep of the environment’s dynamics. When end of episode is reached, you are responsible for calling reset() to reset this environment’s state.
        Accepts an action and returns a tuple (observation, reward, done, info).

        Args:
            action (object): an action provided by the agent 

        Returns:
            observation (object): agent’s observation of the current environment
            reward (float) : amount of reward returned after previous action
            done (bool): whether the episode has ended, in which case further step() calls will return undefined results
            info (dict): contains auxiliary diagnostic information (helpful for debugging, and sometimes learning)
        """
        pass
```

--------------------------------

### Handle JSON Messages

Source: https://gym-donkeycar.readthedocs.io/en/latest/_modules/gym_donkeycar/core/tcp_server

Parses and handles JSON messages received from the client. It includes error handling for malformed JSON and delegates message processing to a message handler.

```python
def handle_json_message(self, chunk):
        '''
        We are expecing a json object
        '''
        try:
            # Replace comma with dots for floats
            # useful when using unity in a language different from English
            chunk = replace_float_notation(chunk)
            # convert data into a string with decode, and then load it as a json object
            jsonObj = json.loads(chunk)
        except Exception as e:
            # something bad happened, usually malformed json packet. jump back to idle and hope things continue
            print(e, 'failed to read json ', chunk)
            return
        '''
        try:
            if self.msg_handler:
                self.msg_handler.on_recv_message(jsonObj)
        except Exception as e:
            print(e, '>>> failure during on_recv_message:', chunk)
        '''
        if self.msg_handler:
            self.msg_handler.on_recv_message(jsonObj)
```

--------------------------------

### Replace Non-Standard Float Notation in JSON String

Source: https://gym-donkeycar.readthedocs.io/en/latest/_modules/gym_donkeycar/core/tcp_server

This function takes a string that may contain JSON with European-style float notation (using commas instead of dots) and converts it into a standard JSON format. It uses regular expressions to find and replace these notations, making the JSON parsable by standard JSON libraries. It handles both inline numbers and numbers at the end of JSON objects.

```python
import json
import re
import asyncore
import socket



[docs]def replace_float_notation(string):
    """
    Replace unity float notation for languages like
    French or German that use comma instead of dot.
    This convert the json sent by Unity to a valid one.
    Ex: "test": 1,2, "key": 2 -> "test": 1.2, "key": 2

    :param string: (str) The incorrect json string
    :return: (str) Valid JSON string
    """
    regex_french_notation = r'"[a-zA-Z_]+":(?P<num>[0-9,E-]+),'
    regex_end = r'"[a-zA-Z_]+":(?P<num>[0-9,E-]+)}'

    for regex in [regex_french_notation, regex_end]:
        matches = re.finditer(regex, string, re.MULTILINE)

        for match in matches:
            num = match.group('num').replace(',', '.')
            string = string.replace(match.group('num'), num)
    return string

```

--------------------------------

### Process DonkeyCar Data and Update State (Python)

Source: https://gym-donkeycar.readthedocs.io/en/latest/_modules/gym_donkeycar/envs/donkey_sim

Parses incoming data, including base64 encoded images, position, speed, and cross-track error, to update the car's state. Handles image decoding and state updates, conditionally updating cross-track error and checking for episode termination conditions. Dependencies include Pillow (PIL) and NumPy.

```python
imgString = data["image"]
image = Image.open(BytesIO(base64.b64decode(imgString)))

# always update the image_array as the observation loop will hang if not changing.
self.image_array = np.asarray(image)

self.x = data["pos_x"]
self.y = data["pos_y"]
self.z = data["pos_z"]
self.speed = data["speed"]

# Cross track error not always present.
# Will be missing if path is not setup in the given scene.
# It should be setup in the 4 scenes available now.
if "cte" in data:
    self.cte = data["cte"]

# don't update hit once session over
if self.over:
    return

self.hit = data["hit"]

self.determine_episode_over()
```
Tokens:
Raw