diff --git a/elevator_system.db b/elevator_system.db new file mode 100644 index 0000000..0ff63ae Binary files /dev/null and b/elevator_system.db differ diff --git a/init_db.py b/init_db.py new file mode 100644 index 0000000..6b531bc --- /dev/null +++ b/init_db.py @@ -0,0 +1,25 @@ +import os +import sqlite3 + +if os.path.exists("elevator_system.db"): + os.remove("elevator_system.db") + print("Previous database deleted.") + +conn = sqlite3.connect("elevator_system.db") +cursor = conn.cursor() + +with open("src/schema.sql", "r") as f: + schema = f.read() + +cursor.executescript(schema) +conn.commit() + +cursor.execute("SELECT name FROM sqlite_master WHERE type='table';") +tables = cursor.fetchall() +print("Created tables:") +for table in tables: + print(f"- {table[0]}") + +conn.close() + +print("Database initialized successfully.") \ No newline at end of file diff --git a/readme copy.md b/readme copy.md new file mode 100644 index 0000000..ea5e444 --- /dev/null +++ b/readme copy.md @@ -0,0 +1,58 @@ +# Dev Test + +## Elevators +When an elevator is empty and not moving this is known as it's resting floor. +The ideal resting floor to be positioned on depends on the likely next floor that the elevator will be called from. + +We can build a prediction engine to predict the likely next floor based on historical demand, if we have the data. + +The goal of this project is to model an elevator and save the data that could later be used to build a prediction engine for which floor is the best resting floor at any time +- When people call an elevator this is considered a demand +- When the elevator is vacant and not moving between floors, the current floor is considered its resting floor +- When the elevator is vacant, it can stay at the current position or move to a different floor +- The prediction model will determine what is the best floor to rest on + + +_The requirement isn't to complete this system but to start building a system that would feed into the training and prediction +of an ML system_ + +You will need to talk through your approach, how you modelled the data and why you thought that data was important, provide endpoints to collect the data and +a means to store the data. Testing is important and will be used verify your system + +## A note on AI generated code +This project isn't about writing code, AI can and will do that for you. +The next step in this process is to talk through your solution and the decisions you made to come to them. It makes for an awkward and rather boring interview reviewing chatgpt's solution. + +If you use a tool to help you write code, that's fine, but we want to see _your_ thought process. + +Provided under the chatgpt folder is the response you get back from chat4o. +If your intention isn't to complete the project but to get an AI to spec it for you please, feel free to submit this instead of wasting OpenAI's server resources. + + +## Problem statement recap +This is a domain modeling problem to build a fit for purpose data storage with a focus on ai data ingestion +- Model the problem into a storage schema (SQL DB schema or whatever you prefer) +- CRUD some data +- Add some flair with a business rule or two +- Have the data in a suitable format to feed to a prediction training algorithm + +--- + +#### To start +- Fork this repo and begin from there +- For your submission, PR into the main repo. We will review it, a offer any feedback and give you a pass / fail if it passes PR +- Don't spend more than 4 hours on this. Projects that pass PR are paid at the standard hourly rate + +#### Marking +- You will be marked on how well your tests cover the code and how useful they would be in a prod system +- You will need to provide storage of some sort. This could be as simple as a sqlite or as complicated as a docker container with a migrations file +- Solutions will be marked against the position you are applying for, a Snr Dev will be expected to have a nearly complete solution and to have thought out the domain and built a schema to fit any issues that could arise +A Jr. dev will be expected to provide a basic design and understand how ML systems like to ingest data + + +#### Trip-ups from the past +Below is a list of some things from previous submissions that haven't worked out +- Built a prediction engine +- Built a full website with bells and whistles +- Spent more than the time allowed (you won't get bonus points for creating an intricate solution, we want a fit for purpose solution) +- Overcomplicated the system mentally and failed to start diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..21bbf7c --- /dev/null +++ b/requirements.txt @@ -0,0 +1,7 @@ +Flask==2.0.1 +Werkzeug==2.0.1 +Flask-SQLAlchemy==2.5.1 +SQLAlchemy==1.4.23 +pytest==7.3.1 +pytest-flask==1.2.0 +python-dotenv==1.0.0 \ No newline at end of file diff --git a/src/db.py b/src/db.py new file mode 100644 index 0000000..1038404 --- /dev/null +++ b/src/db.py @@ -0,0 +1,42 @@ +import os +from contextlib import contextmanager +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker, scoped_session +from sqlalchemy.ext.declarative import declarative_base + +# DB setup - using SQLite by default, but can override with env var +db_url = os.environ.get('DATABASE_URL', 'sqlite:///elevator_system.db') + +# Set up SQLAlchemy stuff +engine = create_engine(db_url, echo=False) # Set echo=True for debugging SQL +factory = sessionmaker(bind=engine) +Session = scoped_session(factory) + +# This is used as the base class for all our models +Base = declarative_base() + +# Create all the tables +def init_db(): + # Import models here to avoid circular imports + from src.models import Building, Elevator, ElevatorCall, ElevatorTrip, ElevatorState, TimeStatistic, FloorStatistic + Base.metadata.create_all(engine) + print("DB initialized!") + +# Helper for managing DB sessions +@contextmanager +def get_db_session(): + # Get a session, use it, then clean up properly + s = Session() + try: + # Let the caller use the session + yield s + # Auto-commit if no exceptions + s.commit() + except Exception as e: + # Something went wrong, rollback + print(f"DB Error: {e}") + s.rollback() + raise + finally: + # Always close the session + s.close() \ No newline at end of file diff --git a/src/main.py b/src/main.py new file mode 100644 index 0000000..26541ed --- /dev/null +++ b/src/main.py @@ -0,0 +1,424 @@ +import os +from datetime import datetime, date +from flask import Flask, request, jsonify, abort +from sqlalchemy import func + +from src.db import init_db, get_db_session +from src.models import * + +app = Flask(__name__) + +# Init DB +@app.before_first_request +def setup(): + init_db() + +# Error stuff +@app.errorhandler(400) +def bad_request(error): + return jsonify({"error": "Bad request", "msg": str(error)}), 400 + +@app.errorhandler(404) +def not_found(error): + return jsonify({"error": "Not found", "msg": str(error)}), 404 + +# Buildings +@app.route('/buildings', methods=['GET']) +def get_buildings(): + with get_db_session() as s: + buildings = s.query(Building).all() + return jsonify([{ + 'id': b.id, + 'name': b.name, + 'floors': b.floors, + 'created_at': b.created_at.isoformat() + } for b in buildings]) + +@app.route('/buildings', methods=['POST']) +def create_building(): + data = request.get_json() + + # Check required fields + if not data or 'name' not in data or 'floors' not in data: + abort(400, "Missing required fields") + + with get_db_session() as s: + b = Building(name=data['name'], floors=data['floors']) + s.add(b) + s.commit() + + return jsonify({ + 'id': b.id, + 'name': b.name, + 'floors': b.floors, + 'created_at': b.created_at.isoformat() + }), 201 + +# Elevators +@app.route('/buildings//elevators', methods=['POST']) +def create_elevator(building_id): + data = request.get_json() + + if not data or 'name' not in data: + abort(400, "Need elevator name") + + with get_db_session() as s: + # Find building + building = s.query(Building).filter(Building.id == building_id).first() + + if not building: + abort(404, f"No building #{building_id}") + + # Create elevator + elev = Elevator(building_id=building_id, name=data['name']) + s.add(elev) + s.commit() + + return jsonify({ + 'id': elev.id, + 'name': elev.name, + 'building_id': elev.building_id, + 'created_at': elev.created_at.isoformat() + }), 201 + +# Elevator calls +@app.route('/elevators//calls', methods=['POST']) +def create_elevator_call(elevator_id): + # Record elevator call and update stats + data = request.get_json() + + if not data or 'floor' not in data or 'direction' not in data: + abort(400, "Need floor and direction") + + if data['direction'] not in ['up', 'down']: + abort(400, "Direction must be up/down") + + with get_db_session() as s: + # Find elevator + elev = s.query(Elevator).filter(Elevator.id == elevator_id).first() + if not elev: + abort(404, f"No elevator #{elevator_id}") + + # Check floor + building = s.query(Building).filter(Building.id == elev.building_id).first() + if data['floor'] < 1 or data['floor'] > building.floors: + abort(400, f"Invalid floor number") + + # Make the call + call = ElevatorCall( + elevator_id=elevator_id, + floor=data['floor'], + direction=data['direction'] + ) + s.add(call) + + # Update stats + today = date.today() + hour = datetime.now().hour + + # Time stats + time_stat = s.query(TimeStatistic).filter( + TimeStatistic.elevator_id == elevator_id, + TimeStatistic.date == today, + TimeStatistic.hour == hour + ).first() + + if time_stat: + time_stat.calls_count += 1 + else: + time_stat = TimeStatistic( + elevator_id=elevator_id, + date=today, + hour=hour, + calls_count=1 + ) + s.add(time_stat) + + # Floor stats + floor_stat = s.query(FloorStatistic).filter( + FloorStatistic.elevator_id == elevator_id, + FloorStatistic.floor == data['floor'], + FloorStatistic.date == today + ).first() + + if floor_stat: + floor_stat.calls_count += 1 + else: + floor_stat = FloorStatistic( + elevator_id=elevator_id, + floor=data['floor'], + date=today, + calls_count=1 + ) + s.add(floor_stat) + + s.commit() + + return jsonify({ + 'id': call.id, + 'elevator_id': call.elevator_id, + 'timestamp': call.timestamp.isoformat(), + 'floor': call.floor, + 'direction': call.direction + }), 201 + +# Elevator states +@app.route('/elevators//states', methods=['POST']) +def create_elevator_state(elevator_id): + # Save current elevator state + data = request.get_json() + + # Check data + if not data or not all(f in data for f in ['floor', 'is_vacant', 'is_moving']): + abort(400, "Missing required fields") + + with get_db_session() as s: + # Check elevator exists + elev = s.query(Elevator).filter(Elevator.id == elevator_id).first() + if not elev: + abort(404, f"No elevator #{elevator_id}") + + # Save state + state = ElevatorState( + elevator_id=elevator_id, + floor=data['floor'], + is_vacant=data['is_vacant'], + is_moving=data['is_moving'] + ) + s.add(state) + s.commit() + + # Return state info + return jsonify({ + 'id': state.id, + 'elevator_id': state.elevator_id, + 'timestamp': state.timestamp.isoformat(), + 'floor': state.floor, + 'is_vacant': state.is_vacant, + 'is_moving': state.is_moving, + 'is_resting': state.is_resting + }), 201 + +# Trips +@app.route('/elevators//trips', methods=['POST']) +def create_elevator_trip(elevator_id): + data = request.get_json() + + # Check data + needed = ['origin_floor', 'destination_floor', 'occupancy'] + if not data or not all(field in data for field in needed): + abort(400, "Missing trip data") + + with get_db_session() as s: + # Check elevator + elev = s.query(Elevator).filter(Elevator.id == elevator_id).first() + if not elev: + abort(404, f"No elevator #{elevator_id}") + + # Create trip + trip = ElevatorTrip( + elevator_id=elevator_id, + start_time=datetime.now(), + origin_floor=data['origin_floor'], + destination_floor=data['destination_floor'], + occupancy=data['occupancy'] + ) + s.add(trip) + s.commit() + + # Return trip info + return jsonify({ + 'id': trip.id, + 'elevator_id': trip.elevator_id, + 'start_time': trip.start_time.isoformat(), + 'origin_floor': trip.origin_floor, + 'destination_floor': trip.destination_floor, + 'occupancy': trip.occupancy + }), 201 + +@app.route('/elevators//trips//complete', methods=['POST']) +def complete_elevator_trip(elevator_id, trip_id): + # Mark trip as complete and update wait times + with get_db_session() as s: + # Find elevator + elev = s.query(Elevator).filter(Elevator.id == elevator_id).first() + if not elev: + abort(404, f"No elevator #{elevator_id}") + + # Find trip + trip = s.query(ElevatorTrip).filter( + ElevatorTrip.id == trip_id, + ElevatorTrip.elevator_id == elevator_id + ).first() + + if not trip: + abort(404, f"Trip not found") + + if trip.end_time: + abort(400, "Already completed") + + # End the trip + trip.end_time = datetime.now() + + # Update wait times if empty elevator + if not trip.occupancy: + # Find matching call + call = s.query(ElevatorCall).filter( + ElevatorCall.elevator_id == elevator_id, + ElevatorCall.floor == trip.destination_floor, + ElevatorCall.wait_time.is_(None) + ).order_by(ElevatorCall.timestamp.desc()).first() + + if call: + # Calculate wait time + call.wait_time = int((trip.end_time - call.timestamp).total_seconds()) + + # Update stats + today = date.today() + floor_stat = s.query(FloorStatistic).filter( + FloorStatistic.elevator_id == elevator_id, + FloorStatistic.floor == call.floor, + FloorStatistic.date == today + ).first() + + if floor_stat: + if floor_stat.avg_wait_time: + # Update average + prev = floor_stat.avg_wait_time * (floor_stat.calls_count - 1) + floor_stat.avg_wait_time = (prev + call.wait_time) / floor_stat.calls_count + else: + floor_stat.avg_wait_time = call.wait_time + + s.commit() + + # Return trip data + return jsonify({ + 'id': trip.id, + 'elevator_id': trip.elevator_id, + 'start_time': trip.start_time.isoformat(), + 'end_time': trip.end_time.isoformat(), + 'origin_floor': trip.origin_floor, + 'destination_floor': trip.destination_floor, + 'occupancy': trip.occupancy, + 'duration': trip.trip_time() + }) + +# Stats for ML +@app.route('/ml/time_statistics', methods=['GET']) +def get_time_statistics(): + # Get stats by time + elev_id = request.args.get('elevator_id', type=int) + date_str = request.args.get('date') + + with get_db_session() as s: + q = s.query(TimeStatistic) + + # Apply filters + if elev_id: + q = q.filter(TimeStatistic.elevator_id == elev_id) + + if date_str: + try: + d = datetime.strptime(date_str, '%Y-%m-%d').date() + q = q.filter(TimeStatistic.date == d) + except ValueError: + abort(400, "Bad date format") + + # Get results + stats = q.all() + + # Format response + return jsonify([{ + 'elevator_id': stat.elevator_id, + 'date': stat.date.isoformat(), + 'hour': stat.hour, + 'calls_count': stat.calls_count, + 'most_common_origin_floor': stat.most_common_origin_floor, + 'most_common_destination_floor': stat.most_common_destination_floor + } for stat in stats]) + +@app.route('/ml/floor_statistics', methods=['GET']) +def get_floor_statistics(): + # Get stats by floor + elev_id = request.args.get('elevator_id', type=int) + floor = request.args.get('floor', type=int) + date_str = request.args.get('date') + + with get_db_session() as s: + q = s.query(FloorStatistic) + + # Filters + if elev_id: + q = q.filter(FloorStatistic.elevator_id == elev_id) + + if floor: + q = q.filter(FloorStatistic.floor == floor) + + if date_str: + try: + d = datetime.strptime(date_str, '%Y-%m-%d').date() + q = q.filter(FloorStatistic.date == d) + except ValueError: + abort(400, "Bad date format") + + # Get data + stats = q.all() + + # Format response + return jsonify([{ + 'elevator_id': stat.elevator_id, + 'floor': stat.floor, + 'date': stat.date.isoformat(), + 'calls_count': stat.calls_count, + 'avg_wait_time': stat.avg_wait_time + } for stat in stats]) + +@app.route('/ml/optimal_resting_floor', methods=['GET']) +def get_optimal_resting_floor(): + # Find best floor for elevator to wait at + elev_id = request.args.get('elevator_id', type=int) + + if not elev_id: + abort(400, "Need elevator_id") + + with get_db_session() as s: + # Check elevator exists + elev = s.query(Elevator).filter(Elevator.id == elev_id).first() + if not elev: + abort(404, f"No elevator #{elev_id}") + + # Get current hour + hour = datetime.now().hour + + # Try to find best floor based on time stats + time_stat = s.query(TimeStatistic).filter( + TimeStatistic.elevator_id == elev_id, + TimeStatistic.hour == hour + ).order_by(TimeStatistic.date.desc()).first() + + if time_stat and time_stat.most_common_origin_floor: + best_floor = time_stat.most_common_origin_floor + else: + # Try floor with most calls today + today = date.today() + floor_stat = s.query(FloorStatistic).filter( + FloorStatistic.elevator_id == elev_id, + FloorStatistic.date == today + ).order_by(FloorStatistic.calls_count.desc()).first() + + if floor_stat: + best_floor = floor_stat.floor + else: + # Just use ground floor + best_floor = 1 + + return jsonify({ + 'elevator_id': elev_id, + 'optimal_resting_floor': best_floor, + 'timestamp': datetime.now().isoformat() + }) + +# Run the app +if __name__ == '__main__': + port = int(os.environ.get('PORT', 5000)) + app.run(host='0.0.0.0', port=port, debug=True) \ No newline at end of file diff --git a/src/models.py b/src/models.py new file mode 100644 index 0000000..50a6220 --- /dev/null +++ b/src/models.py @@ -0,0 +1,186 @@ +from datetime import datetime +from sqlalchemy import * +from sqlalchemy.ext.declarative import declarative_base +from sqlalchemy.orm import relationship +import logging + +# Base model class +Base = declarative_base() + +# This handles buildings - pretty straightforward +class Building(Base): + __tablename__ = 'buildings' + + id = Column(Integer, primary_key=True) + name = Column(String, nullable=False) # Building name + floors = Column(Integer, nullable=False) # How many floors + created_at = Column(DateTime, default=datetime.utcnow) + + # Link to elevators + elevators = relationship("Elevator", back_populates="building") + + def __repr__(self): + return f"Building #{self.id}: {self.name} ({self.floors} floors)" + + +# Elevator model - the main thing we're tracking +class Elevator(Base): + __tablename__ = 'elevators' + + # Basic info + id = Column(Integer, primary_key=True) + building_id = Column(Integer, ForeignKey('buildings.id'), nullable=False) + name = Column(String, nullable=False) + created_at = Column(DateTime, default=datetime.utcnow) + + # Relationships - lots of them! + building = relationship("Building", back_populates="elevators") + calls = relationship("ElevatorCall", back_populates="elevator") + trips = relationship("ElevatorTrip", back_populates="elevator") + states = relationship("ElevatorState", back_populates="elevator") + time_stats = relationship("TimeStatistic", back_populates="elevator") + floor_stats = relationship("FloorStatistic", back_populates="elevator") + + def __str__(self): + return f"Elevator {self.name} in {self.building_id}" + + def __repr__(self): + return f"" + + # Get the latest state + def get_state(self, s): + return s.query(ElevatorState).filter( + ElevatorState.elevator_id == self.id + ).order_by(ElevatorState.timestamp.desc()).first() + + # Is it sitting idle? + def is_resting(self, s): + state = self.get_state(s) + return state and state.is_vacant and not state.is_moving + + # Where is it resting? + def rest_floor(self, s): + state = self.get_state(s) + if state and state.is_vacant and not state.is_moving: + return state.floor + return None + + +# When someone calls the elevator +class ElevatorCall(Base): + __tablename__ = 'elevator_calls' + + # Basic info + id = Column(Integer, primary_key=True) + elevator_id = Column(Integer, ForeignKey('elevators.id'), nullable=False) + timestamp = Column(DateTime, default=datetime.utcnow) + floor = Column(Integer, nullable=False) # Which floor called + direction = Column(String, CheckConstraint("direction IN ('up', 'down')"), nullable=False) + wait_time = Column(Integer) # How long they waited (seconds) + + # Link back to elevator + elevator = relationship("Elevator", back_populates="calls") + + def __repr__(self): + return f"Call #{self.id} - Floor {self.floor} going {self.direction}" + + +# A trip the elevator makes +class ElevatorTrip(Base): + __tablename__ = 'elevator_trips' + + # Trip details + id = Column(Integer, primary_key=True) + elevator_id = Column(Integer, ForeignKey('elevators.id'), nullable=False) + start_time = Column(DateTime, nullable=False) + end_time = Column(DateTime) # Null until trip completes + origin_floor = Column(Integer, nullable=False) + destination_floor = Column(Integer, nullable=False) + occupancy = Column(Boolean, nullable=False) # Had passengers? + + # Link to elevator + elevator = relationship("Elevator", back_populates="trips") + + def __repr__(self): + status = "Completed" if self.end_time else "In progress" + return f"Trip: {self.origin_floor}→{self.destination_floor} ({status})" + + # How long the trip took + def trip_time(self): + if not self.end_time: + return None + return (self.end_time - self.start_time).total_seconds() + + +# Current state of an elevator +class ElevatorState(Base): + __tablename__ = 'elevator_states' + + # State info + id = Column(Integer, primary_key=True) + elevator_id = Column(Integer, ForeignKey('elevators.id'), nullable=False) + timestamp = Column(DateTime, default=datetime.utcnow) + floor = Column(Integer, nullable=False) # Current floor + is_vacant = Column(Boolean, nullable=False) # Empty? + is_moving = Column(Boolean, nullable=False) # Moving? + + # Link to elevator + elevator = relationship("Elevator", back_populates="states") + + # Helper to check if it's just sitting there + @property + def is_resting(self): + return self.is_vacant and not self.is_moving + + def __repr__(self): + if self.is_vacant and not self.is_moving: + status = "resting" + elif not self.is_vacant: + status = "occupied" + else: + status = "moving" + return f"State: Floor {self.floor}, {status}" + + +# Stats by time of day - for ML predictions +class TimeStatistic(Base): + __tablename__ = 'time_statistics' + + # Basic info + id = Column(Integer, primary_key=True) + elevator_id = Column(Integer, ForeignKey('elevators.id'), nullable=False) + date = Column(Date, nullable=False) + hour = Column(Integer, CheckConstraint("hour BETWEEN 0 AND 23"), nullable=False) + + # Stats + calls_count = Column(Integer, default=0) + most_common_origin_floor = Column(Integer) # Where people call from most + most_common_destination_floor = Column(Integer) # Where they go most + + # Link to elevator + elevator = relationship("Elevator", back_populates="time_stats") + + def __str__(self): + return f"Stats for {self.date} at {self.hour}:00 - {self.calls_count} calls" + + +# Stats by floor - for ML predictions +class FloorStatistic(Base): + __tablename__ = 'floor_statistics' + + # Basic info + id = Column(Integer, primary_key=True) + elevator_id = Column(Integer, ForeignKey('elevators.id'), nullable=False) + floor = Column(Integer, nullable=False) + date = Column(Date, nullable=False) + + # Stats + calls_count = Column(Integer, default=0) # How many calls from this floor + avg_wait_time = Column(Float) # Average wait time in seconds + + # Link to elevator + elevator = relationship("Elevator", back_populates="floor_stats") + + def __str__(self): + wait = f"{self.avg_wait_time:.1f}s" if self.avg_wait_time else "unknown" + return f"Floor {self.floor} on {self.date}: {self.calls_count} calls, avg wait {wait}" \ No newline at end of file diff --git a/src/schema.sql b/src/schema.sql new file mode 100644 index 0000000..7e8938e --- /dev/null +++ b/src/schema.sql @@ -0,0 +1,85 @@ +-- Elevator system database schema + +-- Building information +CREATE TABLE buildings ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT NOT NULL, + floors INTEGER NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP +); + +-- Elevator information +CREATE TABLE elevators ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + building_id INTEGER NOT NULL, + name TEXT NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + FOREIGN KEY (building_id) REFERENCES buildings(id) +); + +-- Elevator calls (demands) +CREATE TABLE elevator_calls ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + elevator_id INTEGER NOT NULL, + timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + floor INTEGER NOT NULL, + direction TEXT CHECK(direction IN ('up', 'down')) NOT NULL, + wait_time INTEGER, -- Time in seconds until elevator arrived + FOREIGN KEY (elevator_id) REFERENCES elevators(id) +); + +-- Elevator trips +CREATE TABLE elevator_trips ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + elevator_id INTEGER NOT NULL, + start_time TIMESTAMP NOT NULL, + end_time TIMESTAMP, + origin_floor INTEGER NOT NULL, + destination_floor INTEGER NOT NULL, + occupancy BOOLEAN NOT NULL, -- TRUE if elevator had passengers + FOREIGN KEY (elevator_id) REFERENCES elevators(id) +); + +-- Elevator states (captures point-in-time state) +CREATE TABLE elevator_states ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + elevator_id INTEGER NOT NULL, + timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + floor INTEGER NOT NULL, + is_vacant BOOLEAN NOT NULL, + is_moving BOOLEAN NOT NULL, + is_resting BOOLEAN GENERATED ALWAYS AS (is_vacant AND NOT is_moving) STORED, + FOREIGN KEY (elevator_id) REFERENCES elevators(id) +); + +-- Time-based statistics (for ML features) +CREATE TABLE time_statistics ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + elevator_id INTEGER NOT NULL, + date DATE NOT NULL, + hour INTEGER NOT NULL CHECK(hour BETWEEN 0 AND 23), + calls_count INTEGER DEFAULT 0, + most_common_origin_floor INTEGER, + most_common_destination_floor INTEGER, + FOREIGN KEY (elevator_id) REFERENCES elevators(id), + UNIQUE(elevator_id, date, hour) +); + +-- Floor-based statistics (for ML features) +CREATE TABLE floor_statistics ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + elevator_id INTEGER NOT NULL, + floor INTEGER NOT NULL, + date DATE NOT NULL, + calls_count INTEGER DEFAULT 0, + avg_wait_time REAL, + FOREIGN KEY (elevator_id) REFERENCES elevators(id), + UNIQUE(elevator_id, floor, date) +); + +-- Indexes for performance +CREATE INDEX idx_elevator_calls_timestamp ON elevator_calls(timestamp); +CREATE INDEX idx_elevator_states_timestamp ON elevator_states(timestamp); +CREATE INDEX idx_elevator_trips_start_time ON elevator_trips(start_time); +CREATE INDEX idx_time_statistics_date_hour ON time_statistics(date, hour); +CREATE INDEX idx_floor_statistics_floor_date ON floor_statistics(floor, date); \ No newline at end of file diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..4fe2380 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,64 @@ +import os +import tempfile +import pytest +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker, scoped_session + +from src.db import Base +from src.main import app +from src.models import Building, Elevator, ElevatorCall, ElevatorTrip, ElevatorState, TimeStatistic, FloorStatistic + +@pytest.fixture +def client(): + # Create a temporary database file + db_fd, db_path = tempfile.mkstemp() + app.config['TESTING'] = True + + # Override the database URL + test_db_url = f'sqlite:///{db_path}' + + # Create test engine and session + engine = create_engine(test_db_url) + session_factory = sessionmaker(bind=engine) + Session = scoped_session(session_factory) + + # Create tables + Base.metadata.create_all(engine) + + # Patch the app to use the test database + with app.app_context(): + app.config['DATABASE_URL'] = test_db_url + + # Create a test client + with app.test_client() as client: + yield client + + # Clean up + os.close(db_fd) + os.unlink(db_path) + +@pytest.fixture +def db_session(client): + # Create a new session for each test + engine = create_engine(app.config['DATABASE_URL']) + session_factory = sessionmaker(bind=engine) + session = session_factory() + + try: + yield session + finally: + session.close() + +@pytest.fixture +def sample_building(db_session): + building = Building(name="Test Building", floors=10) + db_session.add(building) + db_session.commit() + return building + +@pytest.fixture +def sample_elevator(db_session, sample_building): + elevator = Elevator(building_id=sample_building.id, name="Test Elevator") + db_session.add(elevator) + db_session.commit() + return elevator \ No newline at end of file diff --git a/tests/test_api.py b/tests/test_api.py new file mode 100644 index 0000000..8ec86f6 --- /dev/null +++ b/tests/test_api.py @@ -0,0 +1,287 @@ +import json +import pytest +from datetime import datetime, date + +class TestBuildingAPI: + def test_create_building(self, client): + response = client.post('/buildings', json={ + 'name': 'Test Building', + 'floors': 10 + }) + + assert response.status_code == 201 + data = json.loads(response.data) + assert data['name'] == 'Test Building' + assert data['floors'] == 10 + assert 'id' in data + assert 'created_at' in data + + def test_get_buildings(self, client, sample_building): + response = client.get('/buildings') + + assert response.status_code == 200 + data = json.loads(response.data) + assert isinstance(data, list) + assert len(data) >= 1 + assert any(b['id'] == sample_building.id for b in data) + + def test_get_building(self, client, sample_building): + response = client.get(f'/buildings/{sample_building.id}') + + assert response.status_code == 200 + data = json.loads(response.data) + assert data['id'] == sample_building.id + assert data['name'] == sample_building.name + assert data['floors'] == sample_building.floors + + def test_get_nonexistent_building(self, client): + response = client.get('/buildings/9999') + + assert response.status_code == 404 + data = json.loads(response.data) + assert 'error' in data + assert 'Not found' in data['error'] + +class TestElevatorAPI: + def test_create_elevator(self, client, sample_building): + response = client.post(f'/buildings/{sample_building.id}/elevators', json={ + 'name': 'Test Elevator' + }) + + assert response.status_code == 201 + data = json.loads(response.data) + assert data['name'] == 'Test Elevator' + assert data['building_id'] == sample_building.id + assert 'id' in data + assert 'created_at' in data + + def test_get_elevators(self, client, sample_building, sample_elevator): + response = client.get(f'/buildings/{sample_building.id}/elevators') + + assert response.status_code == 200 + data = json.loads(response.data) + assert isinstance(data, list) + assert len(data) >= 1 + assert any(e['id'] == sample_elevator.id for e in data) + + def test_get_elevator(self, client, sample_elevator): + response = client.get(f'/elevators/{sample_elevator.id}') + + assert response.status_code == 200 + data = json.loads(response.data) + assert data['id'] == sample_elevator.id + assert data['name'] == sample_elevator.name + assert data['building_id'] == sample_elevator.building_id + + def test_get_nonexistent_elevator(self, client): + response = client.get('/elevators/9999') + + assert response.status_code == 404 + data = json.loads(response.data) + assert 'error' in data + assert 'Not found' in data['error'] + +class TestElevatorCallAPI: + def test_create_call(self, client, sample_elevator, db_session): + response = client.post(f'/elevators/{sample_elevator.id}/calls', json={ + 'floor': 3, + 'direction': 'up' + }) + + assert response.status_code == 201 + data = json.loads(response.data) + assert data['floor'] == 3 + assert data['direction'] == 'up' + assert data['elevator_id'] == sample_elevator.id + assert 'id' in data + assert 'timestamp' in data + + def test_create_call_invalid_direction(self, client, sample_elevator): + response = client.post(f'/elevators/{sample_elevator.id}/calls', json={ + 'floor': 3, + 'direction': 'sideways' # Invalid direction + }) + + assert response.status_code == 400 + data = json.loads(response.data) + assert 'error' in data + assert 'Bad request' in data['error'] + + def test_get_calls(self, client, sample_elevator, db_session): + # Create a call first + client.post(f'/elevators/{sample_elevator.id}/calls', json={ + 'floor': 3, + 'direction': 'up' + }) + + response = client.get(f'/elevators/{sample_elevator.id}/calls') + + assert response.status_code == 200 + data = json.loads(response.data) + assert isinstance(data, list) + assert len(data) >= 1 + assert data[0]['floor'] == 3 + assert data[0]['direction'] == 'up' + +class TestElevatorTripAPI: + def test_create_trip(self, client, sample_elevator): + response = client.post(f'/elevators/{sample_elevator.id}/trips', json={ + 'origin_floor': 2, + 'destination_floor': 8, + 'occupancy': True + }) + + assert response.status_code == 201 + data = json.loads(response.data) + assert data['origin_floor'] == 2 + assert data['destination_floor'] == 8 + assert data['occupancy'] is True + assert data['elevator_id'] == sample_elevator.id + assert 'id' in data + assert 'start_time' in data + + def test_complete_trip(self, client, sample_elevator, db_session): + # Create a trip first + response = client.post(f'/elevators/{sample_elevator.id}/trips', json={ + 'origin_floor': 2, + 'destination_floor': 8, + 'occupancy': True + }) + trip_id = json.loads(response.data)['id'] + + # Complete the trip + response = client.post(f'/elevators/{sample_elevator.id}/trips/{trip_id}/complete') + + assert response.status_code == 200 + data = json.loads(response.data) + assert data['id'] == trip_id + assert data['origin_floor'] == 2 + assert data['destination_floor'] == 8 + assert data['occupancy'] is True + assert 'end_time' in data + assert 'duration' in data + + def test_get_trips(self, client, sample_elevator, db_session): + # Create a trip first + client.post(f'/elevators/{sample_elevator.id}/trips', json={ + 'origin_floor': 2, + 'destination_floor': 8, + 'occupancy': True + }) + + response = client.get(f'/elevators/{sample_elevator.id}/trips') + + assert response.status_code == 200 + data = json.loads(response.data) + assert isinstance(data, list) + assert len(data) >= 1 + assert data[0]['origin_floor'] == 2 + assert data[0]['destination_floor'] == 8 + +class TestElevatorStateAPI: + def test_create_state(self, client, sample_elevator): + response = client.post(f'/elevators/{sample_elevator.id}/states', json={ + 'floor': 5, + 'is_vacant': True, + 'is_moving': False + }) + + assert response.status_code == 201 + data = json.loads(response.data) + assert data['floor'] == 5 + assert data['is_vacant'] is True + assert data['is_moving'] is False + assert data['is_resting'] is True + assert data['elevator_id'] == sample_elevator.id + assert 'id' in data + assert 'timestamp' in data + + def test_get_states(self, client, sample_elevator, db_session): + # Create a state first + client.post(f'/elevators/{sample_elevator.id}/states', json={ + 'floor': 5, + 'is_vacant': True, + 'is_moving': False + }) + + response = client.get(f'/elevators/{sample_elevator.id}/states') + + assert response.status_code == 200 + data = json.loads(response.data) + assert isinstance(data, list) + assert len(data) >= 1 + assert data[0]['floor'] == 5 + assert data[0]['is_vacant'] is True + assert data[0]['is_moving'] is False + + def test_get_current_state(self, client, sample_elevator, db_session): + # Create a state first + client.post(f'/elevators/{sample_elevator.id}/states', json={ + 'floor': 5, + 'is_vacant': True, + 'is_moving': False + }) + + response = client.get(f'/elevators/{sample_elevator.id}/states/current') + + assert response.status_code == 200 + data = json.loads(response.data) + assert data['floor'] == 5 + assert data['is_vacant'] is True + assert data['is_moving'] is False + assert data['is_resting'] is True + +class TestMLDataAPI: + def test_get_time_statistics(self, client, sample_elevator, db_session): + # No statistics yet + response = client.get('/ml/time_statistics') + assert response.status_code == 200 + data = json.loads(response.data) + assert isinstance(data, list) + + # Create a call to generate statistics + client.post(f'/elevators/{sample_elevator.id}/calls', json={ + 'floor': 3, + 'direction': 'up' + }) + + # Check statistics with elevator_id filter + response = client.get(f'/ml/time_statistics?elevator_id={sample_elevator.id}') + assert response.status_code == 200 + data = json.loads(response.data) + assert isinstance(data, list) + assert len(data) >= 1 + assert data[0]['elevator_id'] == sample_elevator.id + assert data[0]['calls_count'] >= 1 + + def test_get_floor_statistics(self, client, sample_elevator, db_session): + # No statistics yet + response = client.get('/ml/floor_statistics') + assert response.status_code == 200 + data = json.loads(response.data) + assert isinstance(data, list) + + # Create a call to generate statistics + client.post(f'/elevators/{sample_elevator.id}/calls', json={ + 'floor': 3, + 'direction': 'up' + }) + + # Check statistics with elevator_id and floor filters + response = client.get(f'/ml/floor_statistics?elevator_id={sample_elevator.id}&floor=3') + assert response.status_code == 200 + data = json.loads(response.data) + assert isinstance(data, list) + assert len(data) >= 1 + assert data[0]['elevator_id'] == sample_elevator.id + assert data[0]['floor'] == 3 + assert data[0]['calls_count'] >= 1 + + def test_get_optimal_resting_floor(self, client, sample_elevator, db_session): + response = client.get(f'/ml/optimal_resting_floor?elevator_id={sample_elevator.id}') + + assert response.status_code == 200 + data = json.loads(response.data) + assert data['elevator_id'] == sample_elevator.id + assert 'optimal_resting_floor' in data + assert 'timestamp' in data \ No newline at end of file diff --git a/tests/test_business_rules.py b/tests/test_business_rules.py new file mode 100644 index 0000000..f6e8040 --- /dev/null +++ b/tests/test_business_rules.py @@ -0,0 +1,266 @@ +import json +import pytest +from datetime import datetime, timedelta +from src.models import ElevatorCall, ElevatorTrip, ElevatorState, TimeStatistic, FloorStatistic + +class TestElevatorBehavior: + def test_elevator_call_updates_statistics(self, client, sample_elevator, db_session): + # Create a call + response = client.post(f'/elevators/{sample_elevator.id}/calls', json={ + 'floor': 3, + 'direction': 'up' + }) + + assert response.status_code == 201 + + # Check time statistics + today = datetime.now().date() + current_hour = datetime.now().hour + + time_stat = db_session.query(TimeStatistic).filter( + TimeStatistic.elevator_id == sample_elevator.id, + TimeStatistic.date == today, + TimeStatistic.hour == current_hour + ).first() + + assert time_stat is not None + assert time_stat.calls_count == 1 + + # Check floor statistics + floor_stat = db_session.query(FloorStatistic).filter( + FloorStatistic.elevator_id == sample_elevator.id, + FloorStatistic.floor == 3, + FloorStatistic.date == today + ).first() + + assert floor_stat is not None + assert floor_stat.calls_count == 1 + + # Create another call from the same floor + response = client.post(f'/elevators/{sample_elevator.id}/calls', json={ + 'floor': 3, + 'direction': 'down' + }) + + assert response.status_code == 201 + + # Check statistics are updated + db_session.refresh(time_stat) + db_session.refresh(floor_stat) + + assert time_stat.calls_count == 2 + assert floor_stat.calls_count == 2 + + def test_trip_completion_updates_wait_time(self, client, sample_elevator, db_session): + # Create a call + call_response = client.post(f'/elevators/{sample_elevator.id}/calls', json={ + 'floor': 3, + 'direction': 'up' + }) + + assert call_response.status_code == 201 + + # Create a trip to respond to the call (elevator moving to the call floor) + trip_response = client.post(f'/elevators/{sample_elevator.id}/trips', json={ + 'origin_floor': 1, + 'destination_floor': 3, + 'occupancy': False # Empty elevator going to pick up + }) + + assert trip_response.status_code == 201 + trip_id = json.loads(trip_response.data)['id'] + + # Complete the trip + complete_response = client.post(f'/elevators/{sample_elevator.id}/trips/{trip_id}/complete') + assert complete_response.status_code == 200 + + # Check that the call's wait time was updated + call = db_session.query(ElevatorCall).filter( + ElevatorCall.elevator_id == sample_elevator.id, + ElevatorCall.floor == 3 + ).order_by(ElevatorCall.timestamp.desc()).first() + + assert call is not None + assert call.wait_time is not None + assert call.wait_time > 0 + + # Check that floor statistics were updated with wait time + today = datetime.now().date() + floor_stat = db_session.query(FloorStatistic).filter( + FloorStatistic.elevator_id == sample_elevator.id, + FloorStatistic.floor == 3, + FloorStatistic.date == today + ).first() + + assert floor_stat is not None + assert floor_stat.avg_wait_time is not None + assert floor_stat.avg_wait_time > 0 + + def test_elevator_state_tracking(self, client, sample_elevator, db_session): + # Create initial state + state_response = client.post(f'/elevators/{sample_elevator.id}/states', json={ + 'floor': 1, + 'is_vacant': True, + 'is_moving': False + }) + + assert state_response.status_code == 201 + + # Create a call + call_response = client.post(f'/elevators/{sample_elevator.id}/calls', json={ + 'floor': 5, + 'direction': 'down' + }) + + assert call_response.status_code == 201 + + # Create a trip (elevator moving to the call floor) + trip_response = client.post(f'/elevators/{sample_elevator.id}/trips', json={ + 'origin_floor': 1, + 'destination_floor': 5, + 'occupancy': False + }) + + assert trip_response.status_code == 201 + trip_id = json.loads(trip_response.data)['id'] + + # Check that elevator state was updated to moving + current_state_response = client.get(f'/elevators/{sample_elevator.id}/states/current') + assert current_state_response.status_code == 200 + current_state = json.loads(current_state_response.data) + + assert current_state['is_moving'] is True + assert current_state['is_vacant'] is True + assert current_state['is_resting'] is False + + # Complete the trip + complete_response = client.post(f'/elevators/{sample_elevator.id}/trips/{trip_id}/complete') + assert complete_response.status_code == 200 + + # Check that elevator state was updated to stationary at the destination floor + current_state_response = client.get(f'/elevators/{sample_elevator.id}/states/current') + assert current_state_response.status_code == 200 + current_state = json.loads(current_state_response.data) + + assert current_state['is_moving'] is False + assert current_state['is_vacant'] is True + assert current_state['is_resting'] is True + assert current_state['floor'] == 5 + + def test_optimal_resting_floor_calculation(self, client, sample_elevator, db_session): + # Create multiple calls from different floors to establish a pattern + for _ in range(3): + client.post(f'/elevators/{sample_elevator.id}/calls', json={ + 'floor': 3, + 'direction': 'up' + }) + + for _ in range(2): + client.post(f'/elevators/{sample_elevator.id}/calls', json={ + 'floor': 7, + 'direction': 'down' + }) + + # Get the optimal resting floor + response = client.get(f'/ml/optimal_resting_floor?elevator_id={sample_elevator.id}') + + assert response.status_code == 200 + data = json.loads(response.data) + + # The optimal floor should be the one with the most calls (floor 3) + assert data['optimal_resting_floor'] == 3 + + # Add more calls from floor 7 to change the optimal floor + for _ in range(4): + client.post(f'/elevators/{sample_elevator.id}/calls', json={ + 'floor': 7, + 'direction': 'down' + }) + + # Get the optimal resting floor again + response = client.get(f'/ml/optimal_resting_floor?elevator_id={sample_elevator.id}') + + assert response.status_code == 200 + data = json.loads(response.data) + + # The optimal floor should now be floor 7 + assert data['optimal_resting_floor'] == 7 + +class TestMLDataPreparation: + def test_time_based_statistics(self, client, sample_elevator, db_session): + # Create calls at different hours by manipulating the timestamp + current_time = datetime.now() + + # Create a call for the current hour + client.post(f'/elevators/{sample_elevator.id}/calls', json={ + 'floor': 3, + 'direction': 'up' + }) + + # Manually insert calls for a different hour + different_hour = (current_time.hour + 1) % 24 + different_hour_call = ElevatorCall( + elevator_id=sample_elevator.id, + floor=5, + direction='down', + timestamp=current_time.replace(hour=different_hour) + ) + db_session.add(different_hour_call) + + # Manually create time statistics for the different hour + different_hour_stat = TimeStatistic( + elevator_id=sample_elevator.id, + date=current_time.date(), + hour=different_hour, + calls_count=1, + most_common_origin_floor=5 + ) + db_session.add(different_hour_stat) + db_session.commit() + + # Get time statistics + response = client.get(f'/ml/time_statistics?elevator_id={sample_elevator.id}') + + assert response.status_code == 200 + data = json.loads(response.data) + + # Should have statistics for both hours + assert len(data) >= 2 + + # Check that we have stats for both hours + hours = [stat['hour'] for stat in data] + assert current_time.hour in hours + assert different_hour in hours + + def test_floor_based_statistics(self, client, sample_elevator, db_session): + # Create calls from different floors + client.post(f'/elevators/{sample_elevator.id}/calls', json={ + 'floor': 3, + 'direction': 'up' + }) + + client.post(f'/elevators/{sample_elevator.id}/calls', json={ + 'floor': 7, + 'direction': 'down' + }) + + # Get floor statistics + response = client.get(f'/ml/floor_statistics?elevator_id={sample_elevator.id}') + + assert response.status_code == 200 + data = json.loads(response.data) + + # Should have statistics for both floors + assert len(data) >= 2 + + # Check that we have stats for both floors + floors = [stat['floor'] for stat in data] + assert 3 in floors + assert 7 in floors + + # Check call counts + floor_3_stat = next(stat for stat in data if stat['floor'] == 3) + floor_7_stat = next(stat for stat in data if stat['floor'] == 7) + + assert floor_3_stat['calls_count'] >= 1 + assert floor_7_stat['calls_count'] >= 1 \ No newline at end of file diff --git a/tests/test_models.py b/tests/test_models.py new file mode 100644 index 0000000..349c6ca --- /dev/null +++ b/tests/test_models.py @@ -0,0 +1,282 @@ +from datetime import datetime, timedelta +import pytest +from src.models import Building, Elevator, ElevatorCall, ElevatorTrip, ElevatorState, TimeStatistic, FloorStatistic + +class TestBuilding: + def test_create_building(self, db_session): + building = Building(name="Test Building", floors=10) + db_session.add(building) + db_session.commit() + + assert building.id is not None + assert building.name == "Test Building" + assert building.floors == 10 + assert building.created_at is not None + + def test_building_elevator_relationship(self, db_session, sample_building): + elevator = Elevator(building_id=sample_building.id, name="Test Elevator") + db_session.add(elevator) + db_session.commit() + + db_session.refresh(sample_building) + assert len(sample_building.elevators) == 1 + assert sample_building.elevators[0].name == "Test Elevator" + +class TestElevator: + def test_create_elevator(self, db_session, sample_building): + elevator = Elevator(building_id=sample_building.id, name="Test Elevator") + db_session.add(elevator) + db_session.commit() + + assert elevator.id is not None + assert elevator.name == "Test Elevator" + assert elevator.building_id == sample_building.id + assert elevator.created_at is not None + + def test_elevator_building_relationship(self, db_session, sample_elevator): + assert sample_elevator.building is not None + assert sample_elevator.building.name == "Test Building" + + def test_current_state(self, db_session, sample_elevator): + # No state yet + assert sample_elevator.current_state(db_session) is None + + # Add a state + state = ElevatorState( + elevator_id=sample_elevator.id, + floor=5, + is_vacant=True, + is_moving=False + ) + db_session.add(state) + db_session.commit() + + current_state = sample_elevator.current_state(db_session) + assert current_state is not None + assert current_state.floor == 5 + assert current_state.is_vacant is True + assert current_state.is_moving is False + + def test_is_resting(self, db_session, sample_elevator): + # No state yet + assert sample_elevator.is_resting(db_session) is False + + # Add a non-resting state (moving) + state1 = ElevatorState( + elevator_id=sample_elevator.id, + floor=5, + is_vacant=True, + is_moving=True + ) + db_session.add(state1) + db_session.commit() + + assert sample_elevator.is_resting(db_session) is False + + # Add a resting state + state2 = ElevatorState( + elevator_id=sample_elevator.id, + floor=7, + is_vacant=True, + is_moving=False + ) + db_session.add(state2) + db_session.commit() + + assert sample_elevator.is_resting(db_session) is True + + # Add a non-resting state (occupied) + state3 = ElevatorState( + elevator_id=sample_elevator.id, + floor=3, + is_vacant=False, + is_moving=False + ) + db_session.add(state3) + db_session.commit() + + assert sample_elevator.is_resting(db_session) is False + + def test_resting_floor(self, db_session, sample_elevator): + # No state yet + assert sample_elevator.resting_floor(db_session) is None + + # Add a non-resting state + state1 = ElevatorState( + elevator_id=sample_elevator.id, + floor=5, + is_vacant=True, + is_moving=True + ) + db_session.add(state1) + db_session.commit() + + assert sample_elevator.resting_floor(db_session) is None + + # Add a resting state + state2 = ElevatorState( + elevator_id=sample_elevator.id, + floor=7, + is_vacant=True, + is_moving=False + ) + db_session.add(state2) + db_session.commit() + + assert sample_elevator.resting_floor(db_session) == 7 + +class TestElevatorCall: + def test_create_call(self, db_session, sample_elevator): + call = ElevatorCall( + elevator_id=sample_elevator.id, + floor=3, + direction="up" + ) + db_session.add(call) + db_session.commit() + + assert call.id is not None + assert call.elevator_id == sample_elevator.id + assert call.floor == 3 + assert call.direction == "up" + assert call.timestamp is not None + assert call.wait_time is None + + def test_elevator_call_relationship(self, db_session, sample_elevator): + call = ElevatorCall( + elevator_id=sample_elevator.id, + floor=3, + direction="up" + ) + db_session.add(call) + db_session.commit() + + db_session.refresh(sample_elevator) + assert len(sample_elevator.calls) == 1 + assert sample_elevator.calls[0].floor == 3 + +class TestElevatorTrip: + def test_create_trip(self, db_session, sample_elevator): + start_time = datetime.now() + trip = ElevatorTrip( + elevator_id=sample_elevator.id, + start_time=start_time, + origin_floor=2, + destination_floor=8, + occupancy=True + ) + db_session.add(trip) + db_session.commit() + + assert trip.id is not None + assert trip.elevator_id == sample_elevator.id + assert trip.start_time == start_time + assert trip.end_time is None + assert trip.origin_floor == 2 + assert trip.destination_floor == 8 + assert trip.occupancy is True + + def test_trip_duration(self, db_session, sample_elevator): + start_time = datetime.now() + trip = ElevatorTrip( + elevator_id=sample_elevator.id, + start_time=start_time, + origin_floor=2, + destination_floor=8, + occupancy=True + ) + db_session.add(trip) + db_session.commit() + + # No end time yet + assert trip.duration() is None + + # Add end time + end_time = start_time + timedelta(seconds=30) + trip.end_time = end_time + db_session.commit() + + assert trip.duration() == 30.0 + +class TestElevatorState: + def test_create_state(self, db_session, sample_elevator): + state = ElevatorState( + elevator_id=sample_elevator.id, + floor=5, + is_vacant=True, + is_moving=False + ) + db_session.add(state) + db_session.commit() + + assert state.id is not None + assert state.elevator_id == sample_elevator.id + assert state.floor == 5 + assert state.is_vacant is True + assert state.is_moving is False + assert state.timestamp is not None + + def test_is_resting_property(self, db_session, sample_elevator): + # Resting state + state1 = ElevatorState( + elevator_id=sample_elevator.id, + floor=5, + is_vacant=True, + is_moving=False + ) + assert state1.is_resting is True + + # Not resting - moving + state2 = ElevatorState( + elevator_id=sample_elevator.id, + floor=5, + is_vacant=True, + is_moving=True + ) + assert state2.is_resting is False + + # Not resting - occupied + state3 = ElevatorState( + elevator_id=sample_elevator.id, + floor=5, + is_vacant=False, + is_moving=False + ) + assert state3.is_resting is False + +class TestStatistics: + def test_create_time_statistic(self, db_session, sample_elevator): + stat = TimeStatistic( + elevator_id=sample_elevator.id, + date=datetime.now().date(), + hour=14, + calls_count=5, + most_common_origin_floor=3, + most_common_destination_floor=7 + ) + db_session.add(stat) + db_session.commit() + + assert stat.id is not None + assert stat.elevator_id == sample_elevator.id + assert stat.hour == 14 + assert stat.calls_count == 5 + assert stat.most_common_origin_floor == 3 + assert stat.most_common_destination_floor == 7 + + def test_create_floor_statistic(self, db_session, sample_elevator): + stat = FloorStatistic( + elevator_id=sample_elevator.id, + floor=3, + date=datetime.now().date(), + calls_count=7, + avg_wait_time=15.5 + ) + db_session.add(stat) + db_session.commit() + + assert stat.id is not None + assert stat.elevator_id == sample_elevator.id + assert stat.floor == 3 + assert stat.calls_count == 7 + assert stat.avg_wait_time == 15.5 \ No newline at end of file