ML-Ready Elevator Domain Model: Event Logging, Business Rules, and Robust Test Coverage

.gitignore

@@ -0,0 +1,9 @@
+__pycache__/
+*.py[cod]
+*$py.class
+
+.pytest_cache/
+
+elevator.db
+
+.vscode/

app/crud.py

@@ -0,0 +1,60 @@
+from sqlalchemy.orm import Session
+from . import models, schemas
+from datetime import timedelta
+
+def create_elevator(db: Session, elevator: schemas.ElevatorCreate):
+    db_elevator = models.Elevator(name=elevator.name)
+    db.add(db_elevator)
+    db.commit()
+    db.refresh(db_elevator)
+    return db_elevator
+
+def create_demand(db: Session, demand: schemas.DemandCreate):
+    #Business Rule: Demand Surge Tag
+    five_min_ago = demand.timestamp - timedelta(minutes=5)
+    recent = db.query(models.DemandEvent).filter(
+        models.DemandEvent.floor == demand.floor,
+        models.DemandEvent.timestamp >= five_min_ago
+    ).count()
+    surge = recent >= 2 #Third demand triggers it
+    db_demand = models.DemandEvent(**demand.dict(), surge_tag=surge)
+    db.add(db_demand)
+    db.commit()
+    db.refresh(db_demand)
+    all_demands = db.query(models.DemandEvent).all()
+    #print(f"ALL DEMANDS IN DB: {[{'floor': d.floor, 'timestamp': d.timestamp} for d in all_demands]}")
+    #print(f"DEBUG: floor={demand.floor}, timestamp={demand.timestamp}, recent={recent}, surge={surge} ")
+    return db_demand
+
+def create_resting(db: Session, resting: schemas.RestingCreate):
+    #Business Rule: Peak Hours & non-optimal rest
+    hour = resting.start_time.hour
+    peak = (7 <= hour < 10) or (16 <= hour < 19)
+    non_optimal = peak and resting.floor != 1
+    db_resting = models.RestingPeriod(
+        **resting.dict(),
+        peak_hours_flag=peak,
+        non_optimal_rested=non_optimal
+    )
+    db.add(db_resting)
+    db.commit()
+    db.refresh(db_resting)
+    return db_resting
+
+def end_resting(db: Session, resting_id: int, end_time):
+    from datetime import timezone
+    rest = db.query(models.RestingPeriod).filter(models.RestingPeriod.id == resting_id).first()
+    if rest:
+        rest.end_time = end_time
+        rest.duration_sec = int((end_time - rest.start_time).total_seconds())
+        #Business Rule: Idle Relocation
+        rest.relocation_flag = rest.duration_sec > 600
+        db.commit()
+        db.refresh(rest)
+    return rest
+
+def get_demands(db: Session):
+    return db.query(models.DemandEvent).all()
+
+def get_restings(db: Session):
+    return db.query(models.RestingPeriod).all()

app/database.py

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

app/main.py

@@ -0,0 +1,51 @@
+from fastapi import FastAPI, Depends, HTTPException
+from sqlalchemy.orm import Session
+from . import models, schemas, crud
+from .database import Base, engine, SessionLocal
+
+Base.metadata.create_all(bind=engine)
+app = FastAPI(title="Elevator ML Data Logger")
+
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+
+@app.post("/elevators/", response_model=schemas.Elevator)
+def create_elevator(elevator: schemas.ElevatorCreate, db: Session = Depends(get_db)):
+    return crud.create_elevator(db, elevator)
+
+@app.post("/demand/", response_model=schemas.Demand)
+def create_demand(demand: schemas.DemandCreate, db: Session = Depends(get_db)):
+    return crud.create_demand(db, demand)
+
+@app.post("/resting/", response_model=schemas.RestingPeriod)
+def create_resting(resting: schemas.RestingCreate, db: Session = Depends(get_db)):
+    return crud.create_resting(db, resting)
+
+@app.patch("/resting/{resting_id}/end", response_model=schemas.RestingPeriod)
+def end_resting(resting_id: int, end: schemas.RestingEnd, db: Session = Depends(get_db)):
+    result = crud.end_resting(db, resting_id, end.end_time)
+    if not result:
+        raise HTTPException(status_code=404, detail="Resting not found")
+    return result
+
+@app.get("/demands/", response_model=list[schemas.Demand])
+def get_demands(db: Session = Depends(get_db)):
+    return crud.get_demands(db)
+
+@app.get("/restings/", response_model=list[schemas.RestingPeriod])
+def get_restings(db: Session = Depends(get_db)):
+    return crud.get_restings(db)
+
+@app.get("/export/")
+def export_all(db: Session = Depends(get_db)):
+    demands = crud.get_demands(db)
+    restings = crud.get_restings(db)
+    #Return as flat JSON for ML
+    return {
+        "demands": [d.__dict__ for d in demands],
+        "restings": [r.__dict__ for r in restings]
+    }

app/models.py

@@ -0,0 +1,34 @@
+from sqlalchemy import Column, Integer, String, DateTime, Boolean, ForeignKey
+from sqlalchemy.orm import relationship
+from .database import Base
+
+class Elevator(Base):
+    __tablename__ = "elevators"
+    id = Column(Integer, primary_key=True, index=True)
+    name = Column(String, unique=True, index=True)
+    demands = relationship("DemandEvent", back_populates="elevator")
+    restings = relationship("RestingPeriod", back_populates="elevator")
+
+class DemandEvent(Base):
+    __tablename__ = "demands"
+    id = Column(Integer, primary_key=True, index=True)
+    elevator_id = Column(Integer, ForeignKey("elevators.id"))
+    floor = Column(Integer)
+    timestamp = Column(DateTime)
+    direction = Column(String)
+    surge_tag = Column(Boolean, default=False)
+    elevator = relationship("Elevator", back_populates="demands")
+
+class RestingPeriod(Base):
+    __tablename__ = "restings"
+    id = Column(Integer, primary_key=True, index=True)
+    elevator_id = Column(Integer, ForeignKey("elevators.id"))
+    floor = Column(Integer)
+    start_time = Column(DateTime)
+    end_time = Column(DateTime, nullable=True)
+    duration_sec = Column(Integer, nullable=True)
+    peak_hours_flag = Column(Boolean, default=False)
+    relocation_flag = Column(Boolean, default=False)
+    non_optimal_rested = Column(Boolean, default=False)
+    elevator = relationship("Elevator", back_populates="restings")
+

app/schemas.py

@@ -0,0 +1,49 @@
+from pydantic import BaseModel
+from typing import Optional
+from datetime import datetime
+
+class ElevatorCreate(BaseModel):
+    name: str
+
+class Elevator(BaseModel):
+    id: int
+    name: str
+    class Config:
+        orm_mode = True
+
+class DemandCreate(BaseModel):
+    elevator_id: int
+    floor: int
+    timestamp: datetime
+    direction: str
+
+class Demand(BaseModel):
+    id: int
+    elevator_id: int
+    floor: int
+    timestamp: datetime
+    direction: str
+    surge_tag: bool
+    class Config:
+        orm_mode = True
+
+class RestingCreate(BaseModel):
+    elevator_id: int
+    floor: int
+    start_time: datetime
+
+class RestingEnd(BaseModel):
+    end_time: datetime
+
+class RestingPeriod(BaseModel):
+    id: int
+    elevator_id: int
+    floor: int
+    start_time: datetime
+    end_time: Optional[datetime]
+    duration_sec: Optional[int]
+    peak_hours_flag: bool
+    relocation_flag: bool
+    non_optimal_rested: bool
+    class Config:
+        orm_mode = True 

app/tests/test_app.py

@@ -0,0 +1,93 @@
+import os
+import pytest
+from fastapi.testclient import TestClient
+from app.main import app
+from app.database import Base, engine
+import uuid
+
+
+@pytest.fixture(autouse=True, scope="function")
+def clean_db():
+    Base.metadata.drop_all(bind=engine)
+    Base.metadata.create_all(bind=engine)
+    yield
+
+
+def create_elevator(client, name=None):
+    if name is None:
+        import uuid
+        name = f"TestElevator_{uuid.uuid4()}"
+    resp = client.post("/elevators/", json={"name": name})
+    assert resp.status_code == 200
+    return resp.json()["id"]
+
+def test_create_demand():
+    client = TestClient(app)
+    eid = create_elevator(client)
+    payload = {
+        "elevator_id": eid,
+        "floor": 3,
+        "timestamp": "2024-06-27T09:00:00",
+        "direction": "up"
+    }
+    resp = client.post("/demand/", json=payload)
+    assert resp.status_code == 200
+    assert resp.json()["floor"] == 3
+
+def test_create_resting():
+    client = TestClient(app)
+    eid = create_elevator(client)
+    payload = {
+        "elevator_id": eid,
+        "floor": 2,
+        "start_time": "2024-06-27T08:00:00"
+    }
+    resp = client.post("/resting/", json=payload)
+    assert resp.status_code == 200
+    assert resp.json()["floor"] == 2
+
+def test_idle_relocation():
+    client = TestClient(app)
+    eid = create_elevator(client)
+    resp = client.post("/resting/", json={
+        "elevator_id": eid,
+        "floor": 2,
+        "start_time": "2024-06-27T08:00:00" 
+    }).json()
+    rid = resp["id"]
+    end_resp = client.patch(f"/resting/{rid}/end", json={"end_time": "2024-06-27T08:12:00"}).json()
+    assert end_resp["relocation_flag"] is True
+
+def test_peak_hours_non_optimal():
+    client = TestClient(app)
+    eid = create_elevator(client)
+    resp = client.post("/resting/", json={
+        "elevator_id": eid,
+        "floor": 4,
+        "start_time": "2024-06-27T08:15:00"  
+    }).json()
+    assert resp ["non_optimal_rested"] is True
+
+def test_demand_surge():
+    client = TestClient(app)
+    eid = create_elevator(client)
+    for minute in [0, 1]:
+        client.post("/demand/", json={
+        "elevator_id": eid,
+        "floor": 6,
+        "timestamp": f"2024-06-27T09:0{minute}:00",
+        "direction": "up" 
+        })
+    resp = client.post("/demand/", json={
+        "elevator_id": eid,
+        "floor": 6,
+        "timestamp": "2024-06-27T09:02:00",
+        "direction": "up"   
+    }).json()
+    assert resp["surge_tag"] is True
+
+def test_export():
+    client = TestClient(app)
+    resp = client.get("/export/")
+    assert resp.status_code == 200
+    assert "demands" in resp.json()

requirements.txt

@@ -0,0 +1,6 @@
+fastapi
+uvicorn
+sqlalchemy
+pydantic
+pytest
+httpx