Devtest Melissa Fagundes

.env

@@ -0,0 +1 @@
+DATABASE_URL=sqlite:///./elevator.db

Dockerfile

@@ -0,0 +1,9 @@
+FROM python:3.10-slim
+
+WORKDIR /app
+
+COPY . /app
+
+RUN pip install --no-cache-dir -r requirements.txt
+
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

README copy.md

@@ -0,0 +1,14 @@
+# Elevator Resting Floor Data Collector
+
+## 🚀 Objective
+Record elevator usage events to later train a predictive model that suggests the ideal "resting floor" for an elevator.
+
+## 📦 How to Run
+
+### Local Environment
+
+```bash
+python -m venv venv
+source venv/bin/activate
+pip install -r requirements.txt
+uvicorn main:app --reload

app/crud.py

@@ -0,0 +1,46 @@
+from sqlalchemy.orm import Session
+from app import models
+from datetime import datetime
+
+def create_elevator(db: Session, current_floor: int):
+    elevator = models.Elevator(current_floor=current_floor)
+    db.add(elevator)
+    db.commit()
+    db.refresh(elevator)
+    return elevator
+
+def update_elevator_status(db: Session, elevator_id: int, floor: int, is_moving: bool, is_occupied: bool):
+    elevator = db.query(models.Elevator).get(elevator_id)
+    if elevator:
+        from_floor = elevator.current_floor
+        elevator.current_floor = floor
+        elevator.is_moving = is_moving
+        elevator.is_occupied = is_occupied
+        elevator.last_updated = datetime.utcnow()
+        db.commit()
+        db.refresh(elevator)
+
+        # Evento
+        db.add(models.ElevatorEvent(
+            elevator_id=elevator_id,
+            event_type="MOVE" if is_moving else "REST",
+            from_floor=from_floor,
+            to_floor=floor
+        ))
+        db.commit()
+        return elevator
+    return None
+
+def create_demand(db: Session, floor_called_from: int, elevator_id: int = 1):
+    demand = models.Demand(floor_called_from=floor_called_from)
+    db.add(demand)
+    db.add(models.ElevatorEvent(
+        elevator_id=elevator_id,
+        event_type="CALL",
+        from_floor=floor_called_from
+    ))
+    db.commit()
+    return demand
+
+def get_events(db: Session):
+    return db.query(models.ElevatorEvent).all()

app/database.py

@@ -0,0 +1,9 @@
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker, declarative_base
+import os
+
+DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./elevator.db")
+
+engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
+SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
+Base = declarative_base()

app/models.py

@@ -0,0 +1,28 @@
+from sqlalchemy import Column, Integer, String, Boolean, ForeignKey, DateTime
+from sqlalchemy.ext.declarative import declarative_base
+from datetime import datetime
+
+Base = declarative_base()
+
+class Elevator(Base):
+    __tablename__ = "elevators"
+    id = Column(Integer, primary_key=True, index=True)
+    current_floor = Column(Integer, nullable=False)
+    is_moving = Column(Boolean, default=False)
+    is_occupied = Column(Boolean, default=False)
+    last_updated = Column(DateTime, default=datetime.utcnow)
+
+class ElevatorEvent(Base):
+    __tablename__ = "events"
+    id = Column(Integer, primary_key=True, index=True)
+    elevator_id = Column(Integer, ForeignKey("elevators.id"))
+    event_type = Column(String, nullable=False)  # CALL, MOVE, REST
+    from_floor = Column(Integer, nullable=True)
+    to_floor = Column(Integer, nullable=True)
+    timestamp = Column(DateTime, default=datetime.utcnow)
+
+class Demand(Base):
+    __tablename__ = "demands"
+    id = Column(Integer, primary_key=True, index=True)
+    floor_called_from = Column(Integer, nullable=False)
+    timestamp = Column(DateTime, default=datetime.utcnow)

app/routes.py

@@ -0,0 +1,31 @@
+from fastapi import APIRouter, Depends
+from sqlalchemy.orm import Session
+from app.database import SessionLocal
+from app import crud, schemas
+
+router = APIRouter()
+
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+
+@router.post("/elevator/")
+def create_elevator(payload: schemas.ElevatorCreate, db: Session = Depends(get_db)):
+    return crud.create_elevator(db, current_floor=payload.current_floor)
+
+@router.patch("/elevator/{elevator_id}/status")
+def update_status(elevator_id: int, payload: schemas.ElevatorStatusUpdate, db: Session = Depends(get_db)):
+    result = crud.update_elevator_status(db, elevator_id, payload.current_floor, payload.is_moving, payload.is_occupied)
+    return {"status": "updated"} if result else {"status": "elevator not found"}
+
+@router.post("/demand/")
+def create_demand(payload: schemas.DemandCreate, db: Session = Depends(get_db)):
+    crud.create_demand(db, floor_called_from=payload.floor_called_from)
+    return {"status": "demand recorded"}
+
+@router.get("/events/", response_model=list[schemas.Event])
+def get_all_events(db: Session = Depends(get_db)):
+    return crud.get_events(db)

app/schemas.py

@@ -0,0 +1,25 @@
+from pydantic import BaseModel
+from typing import Optional
+from datetime import datetime
+
+class ElevatorCreate(BaseModel):
+    current_floor: int
+
+class ElevatorStatusUpdate(BaseModel):
+    current_floor: int
+    is_moving: bool
+    is_occupied: bool
+
+class DemandCreate(BaseModel):
+    floor_called_from: int
+
+class Event(BaseModel):
+    id: int
+    elevator_id: int
+    event_type: str
+    from_floor: Optional[int]
+    to_floor: Optional[int]
+    timestamp: datetime
+
+    class Config:
+        orm_mode = True

export_events.py

@@ -0,0 +1,27 @@
+import csv
+from sqlalchemy.orm import Session
+from app.models import ElevatorEvent
+from app.database import SessionLocal
+
+def export_events_to_csv(filename="events_export.csv"):
+    db: Session = SessionLocal()
+    events = db.query(ElevatorEvent).all()
+    db.close()
+
+    with open(filename, mode='w', newline='') as csv_file:
+        fieldnames = ['id', 'elevator_id', 'event_type', 'from_floor', 'to_floor', 'timestamp']
+        writer = csv.DictWriter(csv_file, fieldnames=fieldnames)
+
+        writer.writeheader()
+        for event in events:
+            writer.writerow({
+                'id': event.id,
+                'elevator_id': event.elevator_id,
+                'event_type': event.event_type,
+                'from_floor': event.from_floor,
+                'to_floor': event.to_floor,
+                'timestamp': event.timestamp
+            })
+
+if __name__ == "__main__":
+    export_events_to_csv()

main.py

@@ -0,0 +1,12 @@
+from fastapi import FastAPI
+from app.routes import router
+from app.database import Base, engine
+
+Base.metadata.create_all(bind=engine)
+
+app = FastAPI()
+app.include_router(router)
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

requirements.txt

@@ -0,0 +1,5 @@
+fastapi
+uvicorn
+sqlalchemy
+pydantic
+pytest

tests/test_main.py

@@ -0,0 +1,49 @@
+import pytest
+from fastapi.testclient import TestClient
+from main import app, Base, engine, SessionLocal
+
+client = TestClient(app)
+
+@pytest.fixture(autouse=True)
+def setup_and_teardown():
+    Base.metadata.drop_all(bind=engine)
+    Base.metadata.create_all(bind=engine)
+    yield
+    Base.metadata.drop_all(bind=engine)
+
+def test_create_elevator():
+    response = client.post("/elevator/", json={"current_floor": 0})
+    assert response.status_code == 200
+    assert "id" in response.json()
+
+def test_update_elevator_status():
+    elevator = client.post("/elevator/", json={"current_floor": 0}).json()
+    elevator_id = elevator["id"]
+    payload = {
+        "current_floor": 3,
+        "is_moving": True,
+        "is_occupied": False
+    }
+    response = client.patch(f"/elevator/{elevator_id}/status", json=payload)
+    assert response.status_code == 200
+    assert response.json()["status"] == "updated"
+
+def test_create_demand():
+    client.post("/elevator/", json={"current_floor": 0})  # Cria elevador 1
+    response = client.post("/demand/", json={"floor_called_from": 5})
+    assert response.status_code == 200
+    assert response.json()["status"] == "demand recorded"
+
+def test_events_logged():
+    client.post("/elevator/", json={"current_floor": 0})
+    client.post("/demand/", json={"floor_called_from": 2})
+    client.patch("/elevator/1/status", json={
+        "current_floor": 5,
+        "is_moving": True,
+        "is_occupied": False
+    })
+    response = client.get("/events/")
+    assert response.status_code == 200
+    events = response.json()
+    assert any(e["event_type"] == "CALL" for e in events)
+    assert any(e["event_type"] == "MOVE" for e in events)