From 56cf2023d05addb351180d0993c68ac9b425374e Mon Sep 17 00:00:00 2001 From: Marcos Tidball <47951223+zysymu@users.noreply.github.com> Date: Mon, 30 Jun 2025 22:00:34 -0300 Subject: [PATCH 1/2] finish test --- README.md | 131 ++++++++++++++ config.py | 20 +++ data/__init__.py | 0 data/collector.py | 172 +++++++++++++++++++ elevator_simulation.db | Bin 0 -> 12288 bytes elevator_simulation_data.csv | 35 ++++ main.py | 128 ++++++++++++++ models/__init__.py | 0 models/building.py | 87 ++++++++++ models/elevator.py | 322 +++++++++++++++++++++++++++++++++++ simulation/__init__.py | 0 simulation/simulator.py | 160 +++++++++++++++++ tests/__init__.py | 0 tests/test_elevator.py | 320 ++++++++++++++++++++++++++++++++++ 14 files changed, 1375 insertions(+) create mode 100644 README.md create mode 100644 config.py create mode 100644 data/__init__.py create mode 100644 data/collector.py create mode 100644 elevator_simulation.db create mode 100644 elevator_simulation_data.csv create mode 100644 main.py create mode 100644 models/__init__.py create mode 100644 models/building.py create mode 100644 models/elevator.py create mode 100644 simulation/__init__.py create mode 100644 simulation/simulator.py create mode 100644 tests/__init__.py create mode 100644 tests/test_elevator.py diff --git a/README.md b/README.md new file mode 100644 index 0000000..9a92d75 --- /dev/null +++ b/README.md @@ -0,0 +1,131 @@ +# Elevator Simulation + +## overview + +the code models an elevator inspired by event-driven entities in game development systems. + +the goal here is to simulate an elevator and to use this simulation to generate data to train an ML model. + +## how it works + +the code is split into the following: + +``` +devtest/ +├── models/ +│ ├── elevator.py # core elevator logic +│ └── building.py # manages the building and elevators +├── simulation/ +│ └── simulator.py # main simulation runner +├── data/ +│ └── collector.py # saves data to sqlite +├── tests/ +│ └── test_elevator.py # simple tests that check if system is okay +├── config.py # settings for the simulation +└── main.py # demos +``` + +the simulation runs in discrete time steps (called _ticks_, inspired by game engines). in each tick, the elevator can move one floor. + +after taking a quick look online, i found out that elevators generally use the SCAN algorithm: the elevator keeps going in one direction as long as there are requests in that direction. once it's done, it switches directions if needed. this approach is more efficient than jumping around for every new request. + +we also have a couple of "business rules": +- if an elevator is 80% full, it won't accept new hall calls. +- if an elevator has been moving for 5 minutes straight, it takes a 30-second maintenance break. + +both of these rules are parameterized, so their values can be changed. + +## how to use it + +### run the demo + +to see it in action, just run: +```bash +python main.py +``` + +here's the terminal output: +``` +=== ELEVATOR SIMULATION DEMO === +running test scenario: +- request floor 3 (elevator should go UP) +- request floor 2 while moving up (should serve 2 then 3) +- request floor 1 and 0 (should serve both going DOWN) +running scenario with 4 requests... +request for floor 3: success +all elevators idle, continuing to next request... +request for floor 2: success +all elevators idle, continuing to next request... +request for floor 1: success +all elevators idle, continuing to next request... +request for floor 0: success +all elevators idle, continuing to next request... +scenario completed in 9.0 seconds +scenario generated 10 events +final elevator state: +- current floor: 0 +- direction: IDLE +- pending requests: [] +- time moving: 2.0s +- time idle: 4.0s +- occupancy: 0/8 +- near capacity: False +- maintenance mode: False + +=== RANDOM SIMULATION FOR DATA COLLECTION === +starting simulation for 60 seconds... +added request for floor 3 +elevator elevator_1: arrived at floor 3 +added request for floor 5 +elevator elevator_1: arrived at floor 5 +added request for floor 2 +elevator elevator_1: arrived at floor 2 +added request for floor 5 +elevator elevator_1: arrived at floor 5 +added request for floor 3 +elevator elevator_1: arrived at floor 3 +added request for floor 5 +elevator elevator_1: arrived at floor 5 +added request for floor 3 +elevator elevator_1: arrived at floor 3 +simulation completed after 60 ticks +collected 34 events +data summary: +- total events recorded: 34 +- simulation duration: 60.3s +- storage type: sqlite +- df shape: (34, 12) +- event types: {'moving': 23, 'arrived': 11} +- data exported to: elevator_simulation_data.csv +simulation complete! +``` + +### run tests +to run the tests: +```bash +python tests/test_elevator.py +``` + +### configuration + +you can change things like the number of floors and elevators in `config.py`. + +## the data + +all the simulation events are saved in a sqlite database file called `elevator_simulation.db`. this makes it easy to use the data later. + +here's what we save for each event: +- `timestamp`: time of the event +- `elevator_id`: id of the elevator +- `event_type`: what happened (e.g., `moving`, `arrived`, `became_idle`) +- `current_floor`: where the elevator is +- `direction`: which way the elevator is going +- `pending_requests`: how many requests are left +- `occupancy`: how many people are inside +- `time_since_last_request`: time since the last button was pressed +- `hour_of_day`: hour of day when the event happened +- `day_of_week`: day of week when the event happened + +### exporting to csv + +after running a simulation, the data is exported a CSV file, which allows it to be easily fed into an ML model via `pandas`. diff --git a/config.py b/config.py new file mode 100644 index 0000000..7ff61ab --- /dev/null +++ b/config.py @@ -0,0 +1,20 @@ +# building setup +BUILDING_CONFIG = { + 'floors': [0, 1, 2, 3, 4, 5], # available floors (0 is ground level) + 'num_elevators': 1, # number of elevators + 'elevator_capacity': 8, # max people that fit inside +} + +# timing settings +SIMULATION_CONFIG = { + 'tick_duration': 1.0, # how long each tick lasts (seconds) + 'floor_travel_time': 3.0, # time to move between floors + 'door_time': 2.0, # time for doors to open/close + 'max_wait_time': 60.0, # how long elevator waits around doing nothing +} + +# where we save all the data for ML training +DATA_CONFIG = { + 'db_file': 'elevator_simulation.db', # sqlite database file + 'collect_events': True, +} \ No newline at end of file diff --git a/data/__init__.py b/data/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/data/collector.py b/data/collector.py new file mode 100644 index 0000000..c8cdf5c --- /dev/null +++ b/data/collector.py @@ -0,0 +1,172 @@ +import time +from datetime import datetime +from typing import List, Dict, Any +import sqlite3 +import pandas as pd + + +class SQLiteDataCollector: + """ + collects all the elevator events and saves them to sqlite + """ + + def __init__(self, db_file: str = "elevator_simulation.db"): + self.db_file = db_file + self.start_time = time.time() + self.last_request_time = self.start_time + + self._initialize_database() + + def _get_connection(self): + """ + connect to the sqlite db + """ + return sqlite3.connect(self.db_file) + + def _initialize_database(self): + """ + set up the db table + """ + create_table_sql = """ + CREATE TABLE IF NOT EXISTS elevator_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + timestamp REAL NOT NULL, + simulation_time REAL NOT NULL, + elevator_id TEXT NOT NULL, + event_type TEXT NOT NULL, + current_floor INTEGER NOT NULL, + direction INTEGER NOT NULL, + is_moving BOOLEAN NOT NULL, + occupancy INTEGER NOT NULL, + pending_requests INTEGER NOT NULL, + time_since_last_request REAL NOT NULL, + hour_of_day INTEGER NOT NULL, + day_of_week INTEGER NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ); + """ + + with self._get_connection() as conn: + cursor = conn.cursor() + cursor.execute(create_table_sql) + conn.commit() + + def record_events(self, events: List[Dict[str, Any]], building_status: Dict): + """ + save elevator events with extra info for ML training + """ + if not events: + return + + current_time = time.time() + simulation_time = current_time - self.start_time + + # get time features + sim_datetime = datetime.fromtimestamp(current_time) + hour_of_day = sim_datetime.hour + day_of_week = sim_datetime.weekday() + + rows = [] + + for event in events: + # get elevator info + elevator_id = event.get('elevator_id', 'unknown') + elevator_info = building_status.get('elevators', {}).get(elevator_id, {}) + + # count how many requests are still pending + pending_requests = ( + len(elevator_info.get('up_requests', [])) + + len(elevator_info.get('down_requests', [])) + ) + + time_since_last_request = current_time - self.last_request_time + + # write everything into a row for the db + row_data = ( + current_time, + simulation_time, + elevator_id, + event.get('type', 'unknown'), + event.get('floor', elevator_info.get('current_floor', 0)), + event.get('direction', 0), + elevator_info.get('is_moving', False), + elevator_info.get('occupancy', 0), + pending_requests, + time_since_last_request, + hour_of_day, + day_of_week, + ) + + rows.append(row_data) + + self._insert_rows(rows) + + def _insert_rows(self, rows: List[tuple]): + """ + insert rows into db + """ + insert_sql = """ + INSERT INTO elevator_events ( + timestamp, simulation_time, elevator_id, event_type, current_floor, + direction, is_moving, occupancy, pending_requests, time_since_last_request, + hour_of_day, day_of_week + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """ + + with self._get_connection() as conn: + cursor = conn.cursor() + cursor.executemany(insert_sql, rows) + conn.commit() + + def record_request(self, floor: int, elevator_id: str): + """ + get last request time + """ + self.last_request_time = time.time() + + def get_data_summary(self) -> Dict: + """ + get a quick summary of how much data was collected + """ + with self._get_connection() as conn: + cursor = conn.cursor() + cursor.execute("SELECT COUNT(*) FROM elevator_events") + event_count = cursor.fetchone()[0] + + return { + 'total_events': event_count, + 'storage_type': 'sqlite', + 'simulation_duration': time.time() - self.start_time + } + + def get_dataframe(self) -> pd.DataFrame: + """ + get all the data as a pandas dataframe + """ + query = """ + SELECT timestamp, simulation_time, elevator_id, event_type, current_floor, + direction, is_moving, occupancy, pending_requests, time_since_last_request, + hour_of_day, day_of_week + FROM elevator_events + ORDER BY timestamp + """ + + with self._get_connection() as conn: + return pd.read_sql_query(query, conn) + + def export_to_csv(self, filename: str = "elevator_data.csv") -> str: + """ + export all data to a csv file + """ + df = self.get_dataframe() + df.to_csv(filename, index=False) + return filename + + def clear_data(self): + """ + clear all data from the database (useful for testing) + """ + with self._get_connection() as conn: + cursor = conn.cursor() + cursor.execute("DELETE FROM elevator_events") + conn.commit() \ No newline at end of file diff --git a/elevator_simulation.db b/elevator_simulation.db new file mode 100644 index 0000000000000000000000000000000000000000..7fb2cbd40541b60bb524d9396ede816f15fda6fa GIT binary patch literal 12288 zcmeI2TTC2P9EWFif!&3LrI#&N+XF2Ygf_Fca(U>SEesTyS?G2N7n=^t?$B=B3*B8X zqz@1yNu!C08q-AWi&y$!Y@#triQ&NqA57H57mW{%srq0{HfmCh7d*qvEHmt3crYTL5CDpz>e1VZ-ddy47iHQ8`Ydnn z*XpU$>Gf*V%i{c%svbliNFxS_0b+m{AO?s5Vt^PR28aP-;6F4lW~r#_?4+mi(b=Sw zyOu-`Kw?h1wjiZryEm=OyCBtaj5j8ToYEHd9JUi_v^0 zD@uz}I-lFMX)!Q&9R%%_K-Us+FdT|-LtGe)g@=V;cnXYgQy@4V866HGj=+T?U4}FA ziIkMfM^p0vCFenCGy+27Ja6ojOQaT((R?D47UlkCU5oipln6J8XjtNQrDA?*UNY+y zTgYaS5pgb=$z+R_;fe7?R*K0H_vx3&iK)zDB7Fs%8y)4jU}%qonOJOLKAMg#?Zcaw z(s4u+v+|6{<#YRVm;H&k1e!WxGMdY4r~Y-7gF3A$NR2k{;hDMl35u z^HN+y5Al&wqq4vR6JSX zE5DB*q!9ze05L!e{5u0etBbPPTAFA(YBt*L3w~Yy>D4j*8OX0K-$(4?P1zI8W)q82 z+{zZZev!Ya+sk+Zr`i4d<{sia6J;Fyv3R zWDV&QT-g@u<>hd=bY9^Yrr0@Qkb&o4&7C&j)MCroLWdiV^9T5&YC4CmSUgTiXHI)a zqv>j$bG|tT&$swqGK5oWWxefkG)$>}ztD^+HU8+G1vqeP=mP_7XI$Gp+Qi~~-9nQZ zPCu@A+|Ia~;2JfYuGV;E<4riw-1Mg*U2$9we2(_xiuIil8ZgDq-WdM@2J4@H#embs z6;cns2JtwCuUFIQ#}$uL$JLZpr_pq^&JMlphVUl*%@9r-*TV6GXg_9Ly}kSq9L;v( z0fa}G4Ghi7GHWsnN9z&X7;!VJ*H>X*+ub_`nkvo$sgo~5JkBc|R>SFM2_C1EYS}Wy z9nx?*b29_q&%oZ}SDFp!6r6l9>G)}nbROX#4p%mC7xsXAH!wJKG3k+ESa*zY0ORg* z@VmF5r!L-uPAa)_qUGQ9+zJ-w@$t24I{l=>~FEOsSWa$A{pFMsD6f(=H~EYP$`vI3~bXYvFX)O*~E=S5ul(qv=Y$bavw@ z*y>)MG=x)V7PipVvVza%AE{Dk4u_8R*xy|jAgKLo6GLM?d);psA780>LuKyV4X6Hi ze*#5YH!gkE8|#zTaOo&``vL*pp+vLmR~Me96uM@itYPAsd#aWiz1(6-qW4R?IfJKHUdv-R}XUr(|haSpS!@x00(I3dacL zN;tz0C3u`VL$*wDWw>;`U)4dXssA3zP?F9HsfHPJ0F%z&!&}vK`h`D<#VP5`X%>y9 zo4*sk&#$qC=A;~9C!AU; ElevatorSimulator: + """ + set up the simulation with elevators and data collection + """ + # create elevators + elevators = [] + for i in range(BUILDING_CONFIG['num_elevators']): + elevator = Elevator( + elevator_id=f"elevator_{i+1}", + floors=BUILDING_CONFIG['floors'], + capacity=BUILDING_CONFIG['elevator_capacity'] + ) + elevators.append(elevator) + + # create building + building = Building( + floors=BUILDING_CONFIG['floors'], + elevators=elevators + ) + + # create data collector + data_collector = SQLiteDataCollector(DATA_CONFIG['db_file']) + + # create simulator + simulator = ElevatorSimulator( + building=building, + data_collector=data_collector, + tick_duration=SIMULATION_CONFIG['tick_duration'] + ) + + return simulator + + +def demo_scenario(): + """ + demonstrates the elevator in a test scenario + """ + print("=== ELEVATOR SIMULATION DEMO ===") + + simulator = create_simulation() + + # test scenario + test_requests = [ + (3, 0), # request floor 3 immediately + (2, 2), # request floor 2 after 2 seconds + (1, 3), # request floor 1 after 3 more seconds + (0, 2), # request floor 0 after 2 more seconds + ] + + print("running test scenario:") + print("- request floor 3 (elevator should go UP)") + print("- request floor 2 while moving up (should serve 2 then 3)") + print("- request floor 1 and 0 (should serve both going DOWN)") + + events = simulator.run_scenario(test_requests) + + print(f"scenario generated {len(events)} events") + + # show final elevator state + status = simulator.get_simulation_status() + elevator_status = status['building_status']['elevators']['elevator_1'] + + print("final elevator state:") + print(f"- current floor: {elevator_status['current_floor']}") + print(f"- direction: {elevator_status['direction']}") + print(f"- pending requests: {elevator_status['up_requests'] + elevator_status['down_requests']}") + print(f"- time moving: {elevator_status['time_moving']:.1f}s") + print(f"- time idle: {elevator_status['time_idle']:.1f}s") + print(f"- occupancy: {elevator_status['occupancy']}/{elevator_status['capacity']}") + print(f"- near capacity: {elevator_status['is_near_capacity']}") + print(f"- maintenance mode: {elevator_status['is_maintenance_mode']}") + + +def run_random_simulation(): + """ + run the simulation with random requests + """ + print("\n=== RANDOM SIMULATION FOR DATA COLLECTION ===") + + simulator = create_simulation() + + # run for 60 seconds with requests every 8 seconds + simulator.run_simulation( + duration_seconds=60, + request_frequency=8.0 + ) + + # show data results + summary = simulator.data_collector.get_data_summary() + print("data summary:") + print(f"- total events recorded: {summary['total_events']}") + print(f"- simulation duration: {summary['simulation_duration']:.1f}s") + print(f"- storage type: {summary['storage_type']}") + + # show DataFrame example + df = simulator.data_collector.get_dataframe() + if not df.empty: + print(f"- df shape: {df.shape}") + print(f"- event types: {df['event_type'].value_counts().to_dict()}") + + # export to CSV + csv_file = simulator.data_collector.export_to_csv("elevator_simulation_data.csv") + print(f"- data exported to: {csv_file}") + + +def main(): + """ + main function - run both demo and data collection + """ + + # run demo scenario first + demo_scenario() + + # run random simulation + run_random_simulation() + + print("simulation complete!") + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/models/__init__.py b/models/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/models/building.py b/models/building.py new file mode 100644 index 0000000..51342d9 --- /dev/null +++ b/models/building.py @@ -0,0 +1,87 @@ +from typing import List, Dict, Optional +from .elevator import Elevator, RequestType + + +class Building: + """ + building entity + + when someone calls for an elevator, we pick the best one to send + (closest idle one, or least busy if they're all working) + + floors: list of available floors (0 is ground level) + elevators: list of elevator objects + total_requests: number of requests made to the building + """ + + def __init__(self, floors: List[int], elevators: List[Elevator]): + self.floors = floors + self.elevators = {elevator.id: elevator for elevator in elevators} + self.total_requests = 0 + + def add_request(self, floor: int, request_type: RequestType = RequestType.HALL_CALL) -> bool: + """ + assign request to closest idle elevator or least busy one + """ + # check if the floor is valid + if floor not in self.floors: + return False + + best_elevator = self._find_best_elevator(floor) + + if best_elevator: + success = best_elevator.add_request(floor, request_type) + if success: + self.total_requests += 1 + + return success + + return False + + def _find_best_elevator(self, requested_floor: int) -> Optional[Elevator]: + """ + pick which elevator should handle this request + prefer idle elevators, then pick the least busy one + """ + # find idle and busy elevators + idle_elevators = [e for e in self.elevators.values() if not e.has_requests()] + busy_elevators = [e for e in self.elevators.values() if e.has_requests()] + + # if any elevators are idle, pick the closest one + if idle_elevators: + return min(idle_elevators, key=lambda e: abs(e.current_floor - requested_floor)) + + # if all elevators are busy, pick the one with least requests + if busy_elevators: + return min(busy_elevators, key=lambda e: len(e.up_requests) + len(e.down_requests)) + + return None + + def tick(self, delta_time: float) -> List[dict]: + """ + run one tick for all elevators and collect everything + """ + all_events = [] + + # run one tick for each elevator + for elevator in self.elevators.values(): + events = elevator.tick(delta_time) + + # tag each event with elevator ID + for event in events: + event['elevator_id'] = elevator.id + all_events.append(event) + + return all_events + + def get_building_status(self) -> dict: + """ + get a snapshot of the whole building + """ + return { + 'total_elevators': len(self.elevators), + 'total_requests': self.total_requests, + 'elevators': {eid: elevator.get_status() for eid, elevator in self.elevators.items()}, + 'floors': self.floors + } + diff --git a/models/elevator.py b/models/elevator.py new file mode 100644 index 0000000..10e5410 --- /dev/null +++ b/models/elevator.py @@ -0,0 +1,322 @@ +from enum import Enum +from typing import Set, List, Optional + + +class Direction(Enum): + """ + which way the elevator is moving + """ + DOWN = -1 + IDLE = 0 + UP = 1 + +class RequestType(Enum): + """ + different kinds of elevator requests + """ + HALL_CALL = "hall_call" # someone outside requested the elevator on a floor + CAR_CALL = "car_call" # someone inside pressed a destination + +class Elevator: + """ + elevator entity + + uses SCAN algorithm: keep going in one direction until no more requests, then turn around + example: at floor 1, requests for 3,2,5 -> goes 2,3,5 (not 2,3,back to 2,then 5) + """ + + def __init__(self, elevator_id: str, floors: List[int], capacity: int): + self.id = elevator_id + self.floors = floors + self.capacity = capacity + + # initial state + self.current_floor = 0 + self.direction = Direction.IDLE + self.occupancy = 0 # number of people inside + self.is_moving = False + + # separate queues for up/down requests + self.up_requests: Set[int] = set() + self.down_requests: Set[int] = set() + + # timing for stats + self.time_moving = 0.0 + self.time_idle = 0.0 + + # business rules vars + self.capacity_threshold = 0.8 # 80% full = stop accepting hall calls + self.maintenance_threshold = 300.0 # 5 minutes of movement = forced break + self.maintenance_rest_time = 30.0 # 30 seconds of forced rest + self.maintenance_timer = 0.0 + self.is_maintenance_mode = False + + def add_request(self, floor: int, request_type: RequestType) -> bool: + """ + adds a request to the relevant queue + + business rule: if elevator is full (80%+), ignore hall calls + (people already inside can still pick destinations though) + """ + # check if the floor is valid + if floor not in self.floors: + return False + + # check if the elevator is already at the floor + if floor == self.current_floor: + return False + + # business rule 1: capacity check + # if too full, don't let new people call the elevator + if request_type == RequestType.HALL_CALL: + capacity_ratio = self.occupancy / self.capacity + if capacity_ratio >= self.capacity_threshold: + return False + + # business rule 2: maintenance mode + # elevator is taking a break, no new requests allowed + if self.is_maintenance_mode: + return False + + # add to relevant direction queue + if floor > self.current_floor: + self.up_requests.add(floor) + else: + self.down_requests.add(floor) + + return True + + def has_requests(self) -> bool: + """ + check if there are any requests + """ + return len(self.up_requests) > 0 or len(self.down_requests) > 0 + + def get_next_floor(self) -> Optional[int]: + """ + uses SCAN algorithm to get next floor + + keep going in same direction until no more requests that way, then switch directions + """ + # if there are no requests, return None + if not self.has_requests(): + return None + + # if idle, pick closest request + if self.direction == Direction.IDLE: + up_closest = min(self.up_requests) if self.up_requests else float('inf') + down_closest = max(self.down_requests) if self.down_requests else float('-inf') + + up_distance = up_closest - self.current_floor + down_distance = self.current_floor - down_closest + + if up_distance <= down_distance and self.up_requests: + self.direction = Direction.UP + return min(self.up_requests) + elif self.down_requests: + self.direction = Direction.DOWN + return max(self.down_requests) + + # keep going in current direction if we have requests that way + if self.direction == Direction.UP and self.up_requests: + return min(self.up_requests) + elif self.direction == Direction.DOWN and self.down_requests: + return max(self.down_requests) + + # switch directions if no more requests in current direction + if self.direction == Direction.UP and self.down_requests: + self.direction = Direction.DOWN + return max(self.down_requests) + elif self.direction == Direction.DOWN and self.up_requests: + self.direction = Direction.UP + return min(self.up_requests) + + return None + + def move_to_floor(self, target_floor: int) -> bool: + """ + move one elevator one floor closer to the target (we only move one floor per tick) + """ + # if already at the target, return True + if target_floor == self.current_floor: + return True + + # go up/down one floor + if target_floor > self.current_floor: + self.current_floor += 1 + self.direction = Direction.UP + else: + self.current_floor -= 1 + self.direction = Direction.DOWN + + self.is_moving = True + return self.current_floor == target_floor + + def arrive_at_floor(self) -> List[str]: + """ + when we arrive at a floor, remove any requests for it + """ + events = [] + + # remove requests for current floor and add event + if self.current_floor in self.up_requests: + self.up_requests.remove(self.current_floor) + events.append("arrived_up") + + if self.current_floor in self.down_requests: + self.down_requests.remove(self.current_floor) + events.append("arrived_down") + + # if no more requests, idle + if not self.has_requests(): + self.is_moving = False + self.direction = Direction.IDLE + events.append("stopped") + + return events + + def tick(self, delta_time: float) -> List[dict]: + """ + update everything that happens each tick + + includes maintenance mode: if elevator works too hard it takes a short break + """ + events = [] + + # track time spent moving vs idle + if self.is_moving: + self.time_moving += delta_time + else: + self.time_idle += delta_time + + # maintenance mode logic + if self.is_maintenance_mode: + self.maintenance_timer += delta_time + + # after break, return to normal + if self.maintenance_timer >= self.maintenance_rest_time: + self.is_maintenance_mode = False + self.maintenance_timer = 0.0 + events.append({ + 'type': 'maintenance_complete', + 'floor': self.current_floor, + 'rest_time': self.maintenance_rest_time + }) + + # continue to rest + else: + events.append({ + 'type': 'maintenance_mode', + 'floor': self.current_floor, + 'remaining_time': self.maintenance_rest_time - self.maintenance_timer + }) + return events + + # check if it's time to break + if self.time_moving >= self.maintenance_threshold and not self.is_maintenance_mode: + self.is_maintenance_mode = True + self.maintenance_timer = 0.0 + self.time_moving = 0.0 # reset timer + events.append({ + 'type': 'maintenance_started', + 'floor': self.current_floor, + 'reason': 'excessive_movement' + }) + return events + + # get next floor to visit + next_floor = self.get_next_floor() + + # if no more requests, idle + if next_floor is None: + if self.is_moving: + self.is_moving = False + self.direction = Direction.IDLE + events.append({ + 'type': 'became_idle', + 'floor': self.current_floor, + 'time_moving': self.time_moving, + 'time_idle': self.time_idle + }) + + # move to next floor and add event + else: + reached = self.move_to_floor(next_floor) + events.append({ + 'type': 'moving', + 'from_floor': self.current_floor - self.direction.value, + 'to_floor': self.current_floor, + 'direction': self.direction.value, + 'target_floor': next_floor + }) + + if reached: + arrival_events = self.arrive_at_floor() + events.append({ + 'type': 'arrived', + 'floor': self.current_floor, + 'arrival_events': arrival_events, + 'occupancy': self.occupancy + }) + + return events + + def set_occupancy(self, occupancy: int) -> bool: + """ + set how many people are in the elevator (for testing) + """ + if occupancy < 0 or occupancy > self.capacity: + return False + + self.occupancy = occupancy + return True + + def add_passengers(self, count: int) -> bool: + """ + add people to the elevator + """ + if self.occupancy + count > self.capacity: + return False + + self.occupancy += count + return True + + def remove_passengers(self, count: int) -> bool: + """ + remove people from the elevator + """ + if self.occupancy - count < 0: + return False + + self.occupancy -= count + return True + + def is_near_capacity(self) -> bool: + """ + check if the elevator is getting full + (used for the capacity business rule) + """ + return (self.occupancy / self.capacity) >= self.capacity_threshold + + def get_status(self) -> dict: + """ + get a snapshot of everything + """ + return { + 'id': self.id, + 'current_floor': self.current_floor, + 'direction': self.direction.name, + 'is_moving': self.is_moving, + 'occupancy': self.occupancy, + 'capacity': self.capacity, + 'up_requests': list(self.up_requests), + 'down_requests': list(self.down_requests), + 'time_moving': self.time_moving, + 'time_idle': self.time_idle, + + # business rule stuff + 'is_maintenance_mode': self.is_maintenance_mode, + 'maintenance_timer': self.maintenance_timer, + 'is_near_capacity': self.is_near_capacity(), + 'capacity_ratio': self.occupancy / self.capacity if self.capacity > 0 else 0 + } diff --git a/simulation/__init__.py b/simulation/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/simulation/simulator.py b/simulation/simulator.py new file mode 100644 index 0000000..6d54edc --- /dev/null +++ b/simulation/simulator.py @@ -0,0 +1,160 @@ +import time +import random +from typing import List, Dict, Optional + +from models.elevator import Elevator, RequestType +from models.building import Building +from data.collector import SQLiteDataCollector + + +class ElevatorSimulator: + """ + the main simulation runner + + handles the loop, timing, random requests, and saving + """ + + def __init__(self, building: Building, data_collector: SQLiteDataCollector, tick_duration: float = 1.0): + self.building = building + self.data_collector = data_collector + self.tick_duration = tick_duration + + self.simulation_time = 0.0 + self.tick_count = 0 + self.is_running = False + + def add_request(self, floor: int, request_type: RequestType = RequestType.HALL_CALL) -> bool: + """ + add a request to the building + """ + success = self.building.add_request(floor, request_type) + + # find which elevator got the request + if success: + elevator_id = list(self.building.elevators.keys())[0] + self.data_collector.record_request(floor, elevator_id) + + return success + + def tick(self) -> List[dict]: + """ + run one step of the simulation + """ + # update building (which updates all elevators) + events = self.building.tick(self.tick_duration) + + # record events to db + if events: + building_status = self.building.get_building_status() + self.data_collector.record_events(events, building_status) + + # update simulation state + self.simulation_time += self.tick_duration + self.tick_count += 1 + + return events + + def run_simulation(self, duration_seconds: float, request_frequency: float = 5.0): + """ + run simulation for a certain duration + generates random requests at a certain frequency + """ + print(f"starting simulation for {duration_seconds} seconds...") + + self.is_running = True + start_time = time.time() + last_request_time = start_time + + # run simulation until duration is reached + while self.is_running and (time.time() - start_time) < duration_seconds: + # generate random requests + current_time = time.time() + + # generate random requests at a certain frequency + if current_time - last_request_time >= request_frequency: + floor = random.choice(self.building.floors) + if self.add_request(floor): + print(f"added request for floor {floor}") + + last_request_time = current_time + + # execute simulation tick + events = self.tick() + + # print events + for event in events: + if event['type'] in ['arrived', 'became_idle']: + print(f"elevator {event['elevator_id']}: {event['type']} at floor {event.get('floor', 'N/A')}") + + # make the timer stop + time.sleep(self.tick_duration) + + # stop the simulation + self.is_running = False + print(f"simulation completed after {self.tick_count} ticks") + + # print summary + summary = self.data_collector.get_data_summary() + print(f"collected {summary['total_events']} events") + + def run_scenario(self, requests: List[tuple]) -> List[dict]: + """ + run a scenario with predefined requests + each request is (floor, delay_seconds) + """ + print(f"running scenario with {len(requests)} requests...") + + all_events = [] + scenario_start = time.time() + + for floor, delay in requests: + # wait for delay + if delay > 0: + time.sleep(delay) + + # add request + success = self.add_request(floor) + print(f"request for floor {floor}: {'success' if success else 'failed'}") + + timeout = time.time() + 30 # 30 second timeout + + # run simulation until elevator becomes idle or timeout + while time.time() < timeout: + events = self.tick() + all_events.extend(events) + + # check if elevator is idle (no more requests) + building_status = self.building.get_building_status() + all_idle = all( + not elevator['up_requests'] and not elevator['down_requests'] + for elevator in building_status['elevators'].values() + ) + + if all_idle: + print(f"all elevators idle, continuing to next request...") + break + + time.sleep(self.tick_duration) + + scenario_duration = time.time() - scenario_start + print(f"scenario completed in {scenario_duration:.1f} seconds") + + return all_events + + def get_simulation_status(self) -> dict: + """ + get current simulation status + """ + return { + 'simulation_time': self.simulation_time, + 'tick_count': self.tick_count, + 'is_running': self.is_running, + 'building_status': self.building.get_building_status(), + 'data_summary': self.data_collector.get_data_summary() + } + + def stop_simulation(self): + """ + stop the running simulation + """ + self.is_running = False \ No newline at end of file diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_elevator.py b/tests/test_elevator.py new file mode 100644 index 0000000..2eb7e04 --- /dev/null +++ b/tests/test_elevator.py @@ -0,0 +1,320 @@ +import sys +import os + +# Add parent directory to path for imports +sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from models.elevator import Elevator, RequestType, Direction +from models.building import Building +from data.collector import SQLiteDataCollector +import time + + +def test_elevator_basic_movement(): + """ + make sure the elevator can move around and handle basic requests + """ + print("testing basic elevator movement...") + + elevator = Elevator("test_elevator", [0, 1, 2, 3, 4, 5], 8) + + # Test initial state + assert elevator.current_floor == 0 + assert elevator.direction == Direction.IDLE + assert not elevator.has_requests() + + # Test adding request + success = elevator.add_request(3, RequestType.HALL_CALL) + assert success, "Should successfully add valid request" + assert elevator.has_requests(), "Should have pending requests" + + # Test movement logic + next_floor = elevator.get_next_floor() + assert next_floor == 3, f"Next floor should be 3, got {next_floor}" + + # Test moving towards target + reached = elevator.move_to_floor(3) + assert not reached, "Should not reach floor 3 in one move" + assert elevator.current_floor == 1, "Should move to floor 1" + assert elevator.direction == Direction.UP, "Should be moving UP" + + print("✓ Basic elevator movement tests passed") + + +def test_business_rule_capacity(): + """ + test the capacity rule: when elevator gets too full, stop accepting hall calls + (but people already inside can still pick destinations) + """ + print("testing capacity business rule...") + + elevator = Elevator("test_elevator", [0, 1, 2, 3, 4, 5], 8) + + # pack the elevator full (above 80% threshold) + elevator.set_occupancy(7) # 7/8 = 87.5% > 80% threshold + + # hall calls should be rejected when too full + success = elevator.add_request(3, RequestType.HALL_CALL) + assert not success, "should reject hall call when near capacity" + + # but people already inside can still pick destinations + success = elevator.add_request(3, RequestType.CAR_CALL) + assert success, "should accept car call even when near capacity" + + # when there's more room, hall calls work again + elevator.set_occupancy(5) # 5/8 = 62.5% < 80% threshold + success = elevator.add_request(4, RequestType.HALL_CALL) + assert success, "should accept hall call when under capacity" + + print("✓ capacity business rule tests passed") + + +def test_business_rule_maintenance(): + """Test Business Rule 2: Maintenance mode after excessive movement""" + print("Testing maintenance business rule...") + + elevator = Elevator("test_elevator", [0, 1, 2, 3, 4, 5], 8) + + # Reduce maintenance threshold for testing (3 seconds instead of 5 minutes) + elevator.maintenance_threshold = 3.0 + elevator.maintenance_rest_time = 1.0 + + # Manually set time_moving to trigger maintenance (simpler test) + elevator.time_moving = 4.0 # Above threshold + + # Add a request to trigger the tick logic + elevator.add_request(2, RequestType.HALL_CALL) + + # Run one tick - should trigger maintenance + events = elevator.tick(1.0) + + # Check that maintenance started + maintenance_started = any(event.get('type') == 'maintenance_started' for event in events) + assert maintenance_started, "Should trigger maintenance when time_moving exceeds threshold" + assert elevator.is_maintenance_mode, "Should be in maintenance mode" + + # Test that requests are rejected during maintenance + success = elevator.add_request(4, RequestType.HALL_CALL) + assert not success, "Should reject requests during maintenance mode" + + # Simulate rest period (1 second + a bit more) + events = elevator.tick(1.2) + + # Should complete maintenance + maintenance_complete = any(event.get('type') == 'maintenance_complete' for event in events) + assert maintenance_complete, "Should complete maintenance after rest period" + assert not elevator.is_maintenance_mode, "Should exit maintenance mode" + + # Test that requests are accepted again + success = elevator.add_request(4, RequestType.HALL_CALL) + assert success, "Should accept requests after maintenance" + + print("✓ Maintenance business rule tests passed") + + +def test_scan_algorithm(): + """Test the SCAN algorithm with multiple requests""" + print("Testing SCAN algorithm...") + + elevator = Elevator("test_elevator", [0, 1, 2, 3, 4, 5], 8) + + # Start at floor 1 + elevator.current_floor = 1 + + # Add requests: 4, 2, 5 (should serve in order: 2, 4, 5) + elevator.add_request(4, RequestType.HALL_CALL) + elevator.add_request(2, RequestType.HALL_CALL) + elevator.add_request(5, RequestType.HALL_CALL) + + # Should go to floor 2 first (closest in UP direction) + next_floor = elevator.get_next_floor() + assert next_floor == 2, f"Should go to floor 2 first, got {next_floor}" + assert elevator.direction == Direction.UP, "Should be moving UP" + + # Simulate reaching floor 2 + elevator.current_floor = 2 + elevator.arrive_at_floor() + + # Next should be floor 4 + next_floor = elevator.get_next_floor() + assert next_floor == 4, f"Should go to floor 4 next, got {next_floor}" + + print("✓ SCAN algorithm tests passed") + + +def test_building_request_distribution(): + """Test building's ability to distribute requests to elevators""" + print("Testing building request distribution...") + + # Create building with one elevator + elevator = Elevator("elevator_1", [0, 1, 2, 3, 4, 5], 8) + building = Building([0, 1, 2, 3, 4, 5], [elevator]) + + # Test adding request + success = building.add_request(3) + assert success, "Building should accept valid request" + assert building.total_requests == 1, "Should track total requests" + + # Test elevator got the request + assert elevator.has_requests(), "Elevator should have received request" + + # Test invalid request + success = building.add_request(10) # Floor doesn't exist + assert not success, "Building should reject invalid floor" + + print("✓ Building request distribution tests passed") + + +def test_sqlite_data_collection(): + """Test SQLite data collection functionality""" + print("Testing SQLite data collection...") + + # Create data collector with test database + collector = SQLiteDataCollector("test_elevator.db") + + # Clear any existing data + collector.clear_data() + + # Test basic event recording + events = [ + { + 'elevator_id': 'test_elevator', + 'type': 'moving', + 'floor': 1, + 'direction': 1 + } + ] + + building_status = { + 'elevators': { + 'test_elevator': { + 'current_floor': 1, + 'is_moving': True, + 'occupancy': 0, + 'up_requests': [2, 3], + 'down_requests': [] + } + } + } + + collector.record_events(events, building_status) + + # Test data summary + time.sleep(0.1) # Small delay to ensure data is recorded + summary = collector.get_data_summary() + assert summary['total_events'] >= 1, f"Should have recorded at least one event, got {summary['total_events']}" + + # Test DataFrame creation + df = collector.get_dataframe() + assert not df.empty, "DataFrame should not be empty" + assert 'elevator_id' in df.columns, "DataFrame should have elevator_id column" + assert 'event_type' in df.columns, "DataFrame should have event_type column" + + # Clean up test database + collector.clear_data() + + print("✓ SQLite data collection tests passed") + + +def test_occupancy_management(): + """Test elevator occupancy management methods""" + print("Testing occupancy management...") + + elevator = Elevator("test_elevator", [0, 1, 2, 3, 4, 5], 8) + + # Test adding passengers + success = elevator.add_passengers(3) + assert success, "Should successfully add passengers" + assert elevator.occupancy == 3, "Occupancy should be 3" + + # Test adding too many passengers + success = elevator.add_passengers(10) + assert not success, "Should reject adding too many passengers" + assert elevator.occupancy == 3, "Occupancy should remain 3" + + # Test removing passengers + success = elevator.remove_passengers(2) + assert success, "Should successfully remove passengers" + assert elevator.occupancy == 1, "Occupancy should be 1" + + # Test removing too many passengers + success = elevator.remove_passengers(5) + assert not success, "Should reject removing too many passengers" + assert elevator.occupancy == 1, "Occupancy should remain 1" + + # Test capacity checking + elevator.set_occupancy(7) # Above 80% threshold + assert elevator.is_near_capacity(), "Should be near capacity" + + elevator.set_occupancy(5) # Below 80% threshold + assert not elevator.is_near_capacity(), "Should not be near capacity" + + print("✓ Occupancy management tests passed") + + +def test_integration_with_business_rules(): + """Integration test with business rules""" + print("Testing integration with business rules...") + + # Create full simulation setup + elevator = Elevator("elevator_1", [0, 1, 2, 3, 4, 5], 8) + building = Building([0, 1, 2, 3, 4, 5], [elevator]) + + # Set elevator near capacity + elevator.set_occupancy(7) # Above threshold + + # Try to add hall call - should be rejected + success = building.add_request(2, RequestType.HALL_CALL) + assert not success, "Hall call should be rejected when near capacity" + + # Reduce occupancy and try again + elevator.set_occupancy(4) # Below threshold + success = building.add_request(2, RequestType.HALL_CALL) + assert success, "Hall call should be accepted when below capacity" + + # Run a few ticks + for i in range(10): + events = building.tick(1.0) + if not any(e.has_requests() for e in building.elevators.values()): + break # All requests served + + # Verify elevator handled requests + assert not elevator.has_requests(), "All requests should be served" + + print("✓ Integration with business rules tests passed") + + +def run_all_tests(): + """run all the tests to make sure everything works""" + print("running elevator simulation tests (with business rules)") + print("=" * 55) + + try: + test_elevator_basic_movement() + test_business_rule_capacity() + test_business_rule_maintenance() + test_scan_algorithm() + test_building_request_distribution() + test_sqlite_data_collection() + test_occupancy_management() + test_integration_with_business_rules() + + print("\n🎉 all tests passed!") + print("✓ basic elevator logic") + print("✓ business rule 1: capacity management") + print("✓ business rule 2: maintenance mode") + print("✓ sqlite data collection") + print("✓ integration testing") + return True + + except AssertionError as e: + print(f"\n❌ Test failed: {e}") + return False + except Exception as e: + print(f"\n💥 Test error: {e}") + return False + + +if __name__ == "__main__": + success = run_all_tests() + exit(0 if success else 1) \ No newline at end of file From 56e9653d84f5a05cfc3d43dd49d72affbee4dd30 Mon Sep 17 00:00:00 2001 From: Marcos Tidball <47951223+zysymu@users.noreply.github.com> Date: Mon, 30 Jun 2025 22:00:55 -0300 Subject: [PATCH 2/2] Delete readme.md --- readme.md | 58 ------------------------------------------------------- 1 file changed, 58 deletions(-) delete mode 100644 readme.md diff --git a/readme.md b/readme.md deleted file mode 100644 index ea5e444..0000000 --- a/readme.md +++ /dev/null @@ -1,58 +0,0 @@ -# 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