ArtiCleaning Robot Control API

1. Introduction

This toy project is a RESTful application developed for ArtiCleaning (invented), designed to remotely control household cleaning robots. The system allows users to define the environment (maps) in which the robots operate, submit cleaning plans (movement sequences), and retrieve historical data of past cleaning sessions.

The solution addresses the following core requirements:

• Environment Management: Parsing and storing rectangular grid maps from different file formats (JSON, TXT).
• Robot Control: Executing movement plans while handling collision detection with obstacles and boundary checks.
• Model Variants: Supporting different robot capabilities, specifically Basic (cleans everywhere) and Premium (skips already clean tiles to save energy).
• Persistence: Storing session metrics (duration, tiles cleaned, status) in a relational database and exporting them via CSV.

Read the full assignment description here.

---

2. General Strategy & Design

To ensure the application is maintainable, scalable, and robust, the following software design principles and strategies were adopted:

2.1. Layered Architecture

The application is structured into distinct layers, each with clear responsibilities.

About the domain logic, the Agent and the Environment are distinct entities, and their interaction is managed by the CleaningSession class, that implements the action-perception.

The API layer is completely decoupled from the core logic, allowing for easy modifications or replacements of either layer without affecting the other.

More specifically, the codebase is organized as follows:

• Core Domain (src/core.py, src/env.py, src/agent.py): Contains the business logic (Agents, Environment, Actions) independent of the web framework.
• Service Layer (src/session.py): Orchestrates the interaction between the Agent and the Environment via the CleaningSession class.
• API Layer (src/api/): Handles HTTP requests, validation (Pydantic), and state management (FastAPI), acting as the entry point.
• Infrastructure (src/utils/): Handles implementation details like Database access, File I/O, and Logging.

2.2. Design Patterns

Key design patterns were utilized to solve specific structural problems:

• Strategy Pattern (Polymorphism):

  • Implementation: The Agent class is defined as an Abstract Base Class (ABC). Concrete implementations, like BasicAgent (blind cleaning) and PremiumAgent (sensor-based cleaning), implement different cleaning policies.
  • Benefit: This allows the CleaningSession to operate without knowing the specific type of robot it is controlling. New robot models can be added in the future without modifying the action-perception loop.

• Factory Method:

  • Implementation: The MapParser class utilizes class methods (load_from_stream) to instantiate the parser (JsonMapParser or TxtMapParser). Each parser knows how to read and convert its specific file format into a standard internal grid representation (numpy array with 0 representing walkable tiles and 1 representing non-walkable tiles).
  • Benefit: Encapsulates the object creation logic. Supporting a new file format (e.g., YAML) only requires adding a new parser class.

• Repository Pattern:

  • Implementation: The SessionRepository class abstracts the data access layer.
  • Benefit: The domain logic and API endpoints interact with high-level methods like save_session() rather than writing raw SQL queries. This makes the code cleaner and allows for easier unit testing or future migration to a different database engine (e.g., PostgreSQL).

2.3. State Management & Concurrency

Despite being a toy application, I tried to anticipate potential concurrency issues that could arise in a production environment with multiple simultaneous users. The solutions implemented are simple (not production-grade probably) yet effective for the scope of this project:

• Multi-User Map Isolation (In-Memory State):

  • Problem: In a multi-user environment, User A might upload a map while User B is uploading a different one. If there were only one "current map" in the global state (as it was at the beginning), User B would overwrite User A's environment before User A could start the cleaning session.
  • Solution: The AppState class maintains a dictionary of maps. When a file is uploaded via /set-map, a unique UUID is generated (map_id). This ID is returned to the client and must be provided in the subsequent /clean request. This ensures that every cleaning session operates on the specific map instance intended by the user, isolating concurrent workflows.

• Database Concurrency (SQLite Locking):

  • Problem: SQLite is a file-based database that locks the file during write operations. If two robots finish their sessions simultaneously and try to write to the DB, a "Database is locked" error could crash one of the requests.
  • Solution: The SessionRepository initializes the SQLite connection with a timeout=30 parameter. This instructs the connection to wait up to 30 seconds for the lock to be released by another thread before raising an exception. This effectively creates a queue for write operations, ensuring data integrity under load without immediate failures.

• Dependency Injection:

  • Implementation: FastAPI's dependency injection system (Depends(get_app_state)) is used to inject the AppState and SessionRepository into endpoints.
  • Benefit: This ensures a single point of truth for the application state and manages the lifecycle of database connections efficiently.

2.4. Robustness

• Collision Handling: The environment rigorously checks bounds and obstacles. Exceptions (CollisionError) are caught and translated into specific session statuses ("ERROR") rather than crashing the server.
• Input Validation: Pydantic schemas enforce strict typing on API inputs (e.g., ensuring steps > 0, valid cardinal directions).

---

3. Core Domain Logic

The simulation is built around a standard Agent-Environment interaction loop. This logic resides entirely within src/core.py, src/env.py, and src/agent.py.

3.1. The Environment (DirtyGridEnv)

The environment models the physical world. It maintains two distinct layers of state:

1. Obstacle Layer (Static): Defined by the input map. Contains non-walkable tiles (1) and walkable tiles (0).
2. Dirt Layer (Dynamic): Tracks the cleanliness of the floor. By default, every tile is initialized as Dirty.

The environment exposes a step(action) method that:

• Updates the robot's position (checking for collisions).
• Updates the dirt layer (if a cleaning action is performed).
• Returns a new Observation (Obs), which includes the current coordinates and sensor data (e.g., is_dirty flag).

3.2. The Composite Action Space

Unlike simple navigation tasks, this domain requires the robot to make two simultaneous decisions at every time step: the next move (given by the plan) and whether to clean the current tile. Therefore, an Action is a composite object consisting of:

• MoveAction: NORTH, SOUTH, EAST, WEST, or STAY.
• CleanAction:
  • CLEAN: removes dirt from the current tile.
  • SKIP: save resources by skipping cleaning.

3.3. Agent Policies

The Agent class is an abstract base class that holds the Motion Plan (the sequence of steps parsed from the user's input). However, the decision of whether to clean or not is delegated to the policy() method, which is implemented differently by each robot model:

Basic Agent

The Basic robot has no dirt sensors. It cleans every tile it traverses, even if it visits the same tile twice.

• Policy: Always returns CleanAction.CLEAN, regardless of the observation.

Premium Agent

The Premium robot is equipped with dirt sensors. It inspects the obs.info['is_dirty'] data provided by the environment. It optimizes resource usage: if a path leads the robot over a tile it has already cleaned, it will pass through without incrementing the "cleaned tiles" counter.

• Policy:
  • If is_dirty == True: Returns CleanAction.CLEAN.
  • If is_dirty == False: Returns CleanAction.SKIP.

3.4. The Simulation Loop

The CleaningSession class orchestrates the interaction loop. At each time step, the robot:

1. observes the current position (and detects eventual collisions)
2. decides whether to clean based on its policy
3. fetches the next move from its motion plan

The action (clean current tile + move to the next tile) of the robot is implemented by the envirnoment that:

1. updates the dirt state of the current tile
2. updates the robot's position

The same repeats at the next time step, till the end of the plan.

---

4. Project Structure

The project is structured to separate configuration, source code, tests, and deployment definitions.

.
├── configs/
│   └── app.json            # Application configuration (DB path, Logging levels, Formats)
├── src/
│   ├── api/
│   │   ├── main.py         # FastAPI entry point & Endpoint definitions
│   │   ├── schemas.py      # Pydantic models for Input/Output validation
│   │   ├── plan_parser.py  # Plan parsing and expansion logic
│   │   ├── dependencies.py # Dependency injection helpers
│   │   └── app_state.py    # Global state definition
│   ├── utils/
│   │   ├── persistence.py  # SQLite repository and SessionRecord handling
│   │   ├── map.py          # Map parsing logic (JSON/TXT Factories)
│   │   ├── log.py          # Logging setup
│   │   └── config.py       # Configuration loader (Pydantic settings)
│   ├── agent.py            # Robot logic (Basic vs Premium implementations)
│   ├── env.py              # Grid Environment & Movement logic
│   ├── session.py          # Action-Perception loop manager
│   └── core.py             # Abstract Base Classes (ABC) and Data Structures
├── tests/                  # Unit and Integration tests
├── Dockerfile              # Containerization instructions
├── requirements.txt        # Python dependencies
└── README.md               # Project documentation

Moreover, on startup, the application reads the configuration from configs/app.json, initializes the logging system (logs everything to logs/app.log for debug purposes), and sets up the SQLite database (in data/persistence/sessions.db) if it does not already exist.

Note on the logging system: while currently it logs to a file, in a production environment, I would consider integrating with a centralized logging system for better traceability and monitoring. For this reason, I did not invested much time in structured logging, but it is listed as a TODO for future improvements.

---

5. Technical Details & Usage

5.1. Prerequisites

• Docker Engine (for containerized execution)
• Python 3.10+ (only required for local development/testing outside Docker)

5.2. Deployment (Docker)

The easiest way to run the application is using the provided Docker container. This ensures all dependencies and environment configurations are handled automatically.

1. Build the Image:

   docker build -t articleaning-api .

2. Run the Container: Map port 8000 of the container to your host machine.

   docker run -p 8000:8000 --name cleaning-robot articleaning-api

   Optional: Persistence By default, the SQLite database is stored inside the container. To persist the database and logs across container restarts, mount a volume:

   docker run -p 8000:8000 -v $(pwd)/data:/app/data articleaning-api

The API will be accessible at http://127.0.0.1:8000/docs.

5.3. Local Development

If you wish to modify the code or run the application without Docker:

1. Install Dependencies: It is recommended to use a virtual environment.

   python -m venv .env
   source .env/bin/activate 
   pip install -r requirements.txt

2. Run the Server: Start the application using Uvicorn with reload enabled.

   uvicorn src.api.main:app --reload

5.4. API Documentation

FastAPI provides an interactive Swagger UI to explore and test the endpoints. Once the app is running, visit:

http://127.0.0.1:8000/docs

Workflow Example

1. Upload Map (POST /set-map):

   • Upload one of the sample files (e.g., map.txt).
   • Copy the map_id from the response.

2. Clean (POST /clean):

   • Use the map_id from the previous step.
   • Define a robot type (basic or premium) and a movement plan.
   • Example Payload:

     {
       "map_id": "paste-your-uuid-here",
       "start_pos": [0, 0],
       "robot_type": "premium",
       "plan": [
         {"direction": "S", "steps": 2},
         {"direction": "E", "steps": 3}
       ]
     }

3. Get History (GET /history):

   • Download a CSV containing metrics for all executed sessions.

5.5. Testing

The project includes a test suite located in the tests/ directory.

To run the tests locally:

pytest

To run tests inside the Docker container:

docker run --rm articleaning-api pytest