Buckets:
tahamajs/Sysmem2_in_AI / ComputerAssignments /CA14_multi_agent_reasoning /src /utils /environment.py
| import random | |
| import json | |
| from typing import Dict, List, Any, Optional, Tuple | |
| from dataclasses import dataclass, asdict | |
| from datetime import datetime, timedelta | |
| import logging | |
| class Flight: | |
| flight_id: str | |
| destination: str | |
| date: str | |
| airline: str | |
| price: float | |
| available_seats: int | |
| departure_time: str | |
| arrival_time: str | |
| def to_dict(self) -> Dict[str, Any]: | |
| return asdict(self) | |
| class Hotel: | |
| hotel_id: str | |
| name: str | |
| destination: str | |
| area: str | |
| check_in_date: str | |
| check_out_date: str | |
| price_per_night: float | |
| available_rooms: int | |
| amenities: List[str] | |
| def to_dict(self) -> Dict[str, Any]: | |
| return asdict(self) | |
| class Activity: | |
| activity_id: str | |
| name: str | |
| destination: str | |
| day: str | |
| duration: str | |
| price: float | |
| description: str | |
| category: str | |
| def to_dict(self) -> Dict[str, Any]: | |
| return asdict(self) | |
| class TravelEnvironment: | |
| def __init__(self): | |
| self.bookings = {} | |
| self.booking_counter = 0 | |
| self.flight_database = self._initialize_flight_database() | |
| self.hotel_database = self._initialize_hotel_database() | |
| self.activity_database = self._initialize_activity_database() | |
| self.config = { | |
| "failure_rate": 0.05, | |
| "price_variation": 0.1, | |
| "availability_threshold": 0.1, | |
| "booking_timeout": 30, | |
| "max_concurrent_bookings": 5, | |
| } | |
| self.logger = logging.getLogger(__name__) | |
| def _initialize_flight_database(self) -> Dict[str, Dict[str, Flight]]: | |
| flights = {} | |
| destinations = { | |
| "Paris": { | |
| "airlines": ["Air France", "Lufthansa", "British Airways"], | |
| "base_price": 450, | |
| "flight_codes": ["AF", "LH", "BA"], | |
| }, | |
| "New York": { | |
| "airlines": ["United Airlines", "Delta", "American Airlines"], | |
| "base_price": 650, | |
| "flight_codes": ["UA", "DL", "AA"], | |
| }, | |
| "Tokyo": { | |
| "airlines": ["Japan Airlines", "ANA", "United Airlines"], | |
| "base_price": 850, | |
| "flight_codes": ["JL", "NH", "UA"], | |
| }, | |
| "London": { | |
| "airlines": ["British Airways", "Virgin Atlantic", "Lufthansa"], | |
| "base_price": 500, | |
| "flight_codes": ["BA", "VS", "LH"], | |
| }, | |
| } | |
| start_date = datetime.now() | |
| for destination, info in destinations.items(): | |
| flights[destination] = {} | |
| for i in range(90): | |
| date = start_date + timedelta(days=i) | |
| date_str = date.strftime("%Y-%m-%d") | |
| if random.random() < 0.7: | |
| airline = random.choice(info["airlines"]) | |
| flight_code = random.choice(info["flight_codes"]) | |
| flight_id = f"{flight_code}{random.randint(100, 999)}" | |
| price = info["base_price"] * (1 + random.uniform(-0.2, 0.3)) | |
| flight = Flight( | |
| flight_id=flight_id, | |
| destination=destination, | |
| date=date_str, | |
| airline=airline, | |
| price=round(price, 2), | |
| available_seats=random.randint(5, 50), | |
| departure_time=f"{random.randint(6, 22):02d}:{random.choice(['00', '30'])}", | |
| arrival_time=f"{random.randint(8, 23):02d}:{random.choice(['00', '30'])}", | |
| ) | |
| flights[destination][date_str] = flight | |
| return flights | |
| def _initialize_hotel_database(self) -> Dict[str, Dict[str, Dict[str, Hotel]]]: | |
| hotels = {} | |
| destinations = { | |
| "Paris": { | |
| "areas": [ | |
| "1st arrondissement", | |
| "7th arrondissement", | |
| "Montmartre", | |
| "Marais", | |
| ], | |
| "hotels": [ | |
| ("Hotel Regina", "luxury", 300), | |
| ("Hotel Eiffel", "mid-range", 180), | |
| ("Hotel Montmartre", "budget", 120), | |
| ("Hotel Marais", "boutique", 200), | |
| ], | |
| }, | |
| "New York": { | |
| "areas": ["Manhattan", "Brooklyn", "Queens"], | |
| "hotels": [ | |
| ("The Plaza Hotel", "luxury", 500), | |
| ("1 Hotel Brooklyn", "mid-range", 250), | |
| ("Pod Hotel", "budget", 150), | |
| ("The Standard", "boutique", 300), | |
| ], | |
| }, | |
| "Tokyo": { | |
| "areas": ["Shibuya", "Ginza", "Shinjuku", "Asakusa"], | |
| "hotels": [ | |
| ("Park Hyatt Tokyo", "luxury", 400), | |
| ("The Ritz-Carlton Tokyo", "luxury", 450), | |
| ("Shibuya Hotel", "mid-range", 200), | |
| ("Asakusa Hotel", "budget", 100), | |
| ], | |
| }, | |
| } | |
| for destination, info in destinations.items(): | |
| hotels[destination] = {} | |
| for area in info["areas"]: | |
| hotels[destination][area] = {} | |
| for hotel_name, category, base_price in info["hotels"]: | |
| hotel_id = ( | |
| f"{hotel_name.replace(' ', '_').lower()}_{destination.lower()}" | |
| ) | |
| start_date = datetime.now() | |
| for i in range(90): | |
| date = start_date + timedelta(days=i) | |
| date_str = date.strftime("%Y-%m-%d") | |
| if random.random() < 0.8: | |
| price = base_price * (1 + random.uniform(-0.1, 0.2)) | |
| amenities = [] | |
| if category == "luxury": | |
| amenities = [ | |
| "spa", | |
| "gym", | |
| "restaurant", | |
| "concierge", | |
| "room_service", | |
| ] | |
| elif category == "mid-range": | |
| amenities = ["gym", "restaurant", "wifi"] | |
| else: | |
| amenities = ["wifi", "breakfast"] | |
| hotel = Hotel( | |
| hotel_id=hotel_id, | |
| name=hotel_name, | |
| destination=destination, | |
| area=area, | |
| check_in_date=date_str, | |
| check_out_date=(date + timedelta(days=1)).strftime( | |
| "%Y-%m-%d" | |
| ), | |
| price_per_night=round(price, 2), | |
| available_rooms=random.randint(2, 20), | |
| amenities=amenities, | |
| ) | |
| hotels[destination][area][date_str] = hotel | |
| return hotels | |
| def _initialize_activity_database(self) -> Dict[str, Dict[str, List[Activity]]]: | |
| activities = {} | |
| destinations = { | |
| "Paris": { | |
| "Day 1": [ | |
| ( | |
| "Eiffel Tower Visit", | |
| "sightseeing", | |
| 25, | |
| "Visit the iconic Eiffel Tower", | |
| ), | |
| ( | |
| "Louvre Museum Tour", | |
| "culture", | |
| 20, | |
| "Explore the world's largest museum", | |
| ), | |
| ( | |
| "Seine River Cruise", | |
| "leisure", | |
| 35, | |
| "Scenic boat tour along the Seine", | |
| ), | |
| ], | |
| "Day 2": [ | |
| ( | |
| "Montmartre Walking Tour", | |
| "sightseeing", | |
| 15, | |
| "Explore the artistic Montmartre district", | |
| ), | |
| ( | |
| "Sacré-Cœur Basilica", | |
| "culture", | |
| 10, | |
| "Visit the beautiful basilica", | |
| ), | |
| ("Moulin Rouge Show", "entertainment", 120, "Famous cabaret show"), | |
| ], | |
| "Day 3": [ | |
| ( | |
| "Versailles Palace", | |
| "culture", | |
| 30, | |
| "Day trip to the Palace of Versailles", | |
| ), | |
| ( | |
| "Notre-Dame Cathedral", | |
| "culture", | |
| 0, | |
| "Visit the historic cathedral", | |
| ), | |
| ( | |
| "Champs-Élysées Shopping", | |
| "shopping", | |
| 0, | |
| "Shop along the famous avenue", | |
| ), | |
| ], | |
| }, | |
| "New York": { | |
| "Day 1": [ | |
| ( | |
| "Central Park Walk", | |
| "leisure", | |
| 0, | |
| "Stroll through the famous park", | |
| ), | |
| ( | |
| "Metropolitan Museum", | |
| "culture", | |
| 25, | |
| "Visit the world-class museum", | |
| ), | |
| ( | |
| "Times Square", | |
| "sightseeing", | |
| 0, | |
| "Experience the bustling square", | |
| ), | |
| ], | |
| "Day 2": [ | |
| ("Statue of Liberty", "sightseeing", 20, "Visit the iconic statue"), | |
| ("Ellis Island", "culture", 15, "Learn about immigration history"), | |
| ("Broadway Show", "entertainment", 100, "Watch a Broadway musical"), | |
| ], | |
| "Day 3": [ | |
| ( | |
| "Brooklyn Bridge", | |
| "sightseeing", | |
| 0, | |
| "Walk across the historic bridge", | |
| ), | |
| ("High Line Park", "leisure", 0, "Walk along the elevated park"), | |
| ("9/11 Memorial", "culture", 0, "Pay respects at the memorial"), | |
| ], | |
| }, | |
| } | |
| for destination, days in destinations.items(): | |
| activities[destination] = {} | |
| for day, day_activities in days.items(): | |
| activities[destination][day] = [] | |
| for name, category, price, description in day_activities: | |
| activity_id = ( | |
| f"{name.replace(' ', '_').lower()}_{destination.lower()}" | |
| ) | |
| activity = Activity( | |
| activity_id=activity_id, | |
| name=name, | |
| destination=destination, | |
| day=day, | |
| duration=f"{random.randint(1, 4)} hours", | |
| price=price, | |
| description=description, | |
| category=category, | |
| ) | |
| activities[destination][day].append(activity) | |
| return activities | |
| def search_flights(self, destination: str, date: str) -> Dict[str, Any]: | |
| self.logger.info(f"Searching flights to {destination} on {date}") | |
| if random.random() < self.config["failure_rate"]: | |
| return { | |
| "success": False, | |
| "error": "Flight search service temporarily unavailable", | |
| "flights": [], | |
| } | |
| if destination not in self.flight_database: | |
| return { | |
| "success": False, | |
| "error": f"No flights available to {destination}", | |
| "flights": [], | |
| } | |
| if date not in self.flight_database[destination]: | |
| return { | |
| "success": False, | |
| "error": f"No flights found for {date}. Please try a different date.", | |
| "flights": [], | |
| } | |
| flight = self.flight_database[destination][date] | |
| if random.random() < self.config["availability_threshold"]: | |
| return { | |
| "success": False, | |
| "error": f"No seats available on {flight.flight_id} for {date}", | |
| "flights": [], | |
| } | |
| return { | |
| "success": True, | |
| "flights": [flight.to_dict()], | |
| "message": f"Found flight {flight.flight_id} to {destination}", | |
| } | |
| def book_flight( | |
| self, flight_id: str, destination: str, date: str | |
| ) -> Dict[str, Any]: | |
| self.logger.info(f"Booking flight {flight_id} to {destination} on {date}") | |
| if random.random() < self.config["failure_rate"]: | |
| return { | |
| "success": False, | |
| "error": "Booking service temporarily unavailable", | |
| } | |
| if ( | |
| destination not in self.flight_database | |
| or date not in self.flight_database[destination] | |
| ): | |
| return {"success": False, "error": "Flight not found"} | |
| flight = self.flight_database[destination][date] | |
| if flight.flight_id != flight_id: | |
| return {"success": False, "error": "Flight ID does not match"} | |
| if flight.available_seats <= 0: | |
| return {"success": False, "error": "No seats available"} | |
| booking_id = f"flight_{self.booking_counter}" | |
| self.booking_counter += 1 | |
| booking = { | |
| "booking_id": booking_id, | |
| "type": "flight", | |
| "flight_id": flight_id, | |
| "destination": destination, | |
| "date": date, | |
| "airline": flight.airline, | |
| "price": flight.price, | |
| "departure_time": flight.departure_time, | |
| "arrival_time": flight.arrival_time, | |
| "booking_time": datetime.now().isoformat(), | |
| } | |
| self.bookings[booking_id] = booking | |
| flight.available_seats -= 1 | |
| return { | |
| "success": True, | |
| "booking_id": booking_id, | |
| "message": f"✅ Flight booked: {flight_id} to {destination} on {date}", | |
| "booking": booking, | |
| } | |
| def search_hotels( | |
| self, destination: str, check_in_date: str, area: str = None | |
| ) -> Dict[str, Any]: | |
| self.logger.info(f"Searching hotels in {destination} for {check_in_date}") | |
| if random.random() < self.config["failure_rate"]: | |
| return { | |
| "success": False, | |
| "error": "Hotel search service temporarily unavailable", | |
| "hotels": [], | |
| } | |
| if destination not in self.hotel_database: | |
| return { | |
| "success": False, | |
| "error": f"No hotels available in {destination}", | |
| "hotels": [], | |
| } | |
| available_hotels = [] | |
| areas_to_search = ( | |
| [area] | |
| if area and area in self.hotel_database[destination] | |
| else self.hotel_database[destination].keys() | |
| ) | |
| for search_area in areas_to_search: | |
| if check_in_date in self.hotel_database[destination][search_area]: | |
| hotel = self.hotel_database[destination][search_area][check_in_date] | |
| if ( | |
| hotel.available_rooms > 0 | |
| and random.random() > self.config["availability_threshold"] | |
| ): | |
| available_hotels.append(hotel.to_dict()) | |
| if not available_hotels: | |
| return { | |
| "success": False, | |
| "error": f"No hotels available in {destination} for {check_in_date}", | |
| "hotels": [], | |
| } | |
| return { | |
| "success": True, | |
| "hotels": available_hotels, | |
| "message": f"Found {len(available_hotels)} hotels in {destination}", | |
| } | |
| def book_hotel( | |
| self, hotel_id: str, destination: str, check_in_date: str, area: str | |
| ) -> Dict[str, Any]: | |
| self.logger.info(f"Booking hotel {hotel_id} in {destination}") | |
| if random.random() < self.config["failure_rate"]: | |
| return { | |
| "success": False, | |
| "error": "Booking service temporarily unavailable", | |
| } | |
| if ( | |
| destination not in self.hotel_database | |
| or area not in self.hotel_database[destination] | |
| or check_in_date not in self.hotel_database[destination][area] | |
| ): | |
| return {"success": False, "error": "Hotel not found"} | |
| hotel = self.hotel_database[destination][area][check_in_date] | |
| if hotel.hotel_id != hotel_id: | |
| return {"success": False, "error": "Hotel ID does not match"} | |
| if hotel.available_rooms <= 0: | |
| return {"success": False, "error": "No rooms available"} | |
| booking_id = f"hotel_{self.booking_counter}" | |
| self.booking_counter += 1 | |
| booking = { | |
| "booking_id": booking_id, | |
| "type": "hotel", | |
| "hotel_id": hotel_id, | |
| "hotel_name": hotel.name, | |
| "destination": destination, | |
| "area": area, | |
| "check_in_date": check_in_date, | |
| "check_out_date": hotel.check_out_date, | |
| "price_per_night": hotel.price_per_night, | |
| "amenities": hotel.amenities, | |
| "booking_time": datetime.now().isoformat(), | |
| } | |
| self.bookings[booking_id] = booking | |
| hotel.available_rooms -= 1 | |
| return { | |
| "success": True, | |
| "booking_id": booking_id, | |
| "message": f"✅ Hotel booked: {hotel.name} in {area}, {destination}", | |
| "booking": booking, | |
| } | |
| def search_activities(self, destination: str, day: str) -> Dict[str, Any]: | |
| self.logger.info(f"Searching activities in {destination} for {day}") | |
| if random.random() < self.config["failure_rate"]: | |
| return { | |
| "success": False, | |
| "error": "Activity search service temporarily unavailable", | |
| "activities": [], | |
| } | |
| if destination not in self.activity_database: | |
| return { | |
| "success": False, | |
| "error": f"No activities found for {destination}", | |
| "activities": [], | |
| } | |
| if day not in self.activity_database[destination]: | |
| return { | |
| "success": False, | |
| "error": f"No activities found for {day} in {destination}", | |
| "activities": [], | |
| } | |
| activities = self.activity_database[destination][day] | |
| return { | |
| "success": True, | |
| "activities": [activity.to_dict() for activity in activities], | |
| "message": f"Found {len(activities)} activities for {day} in {destination}", | |
| } | |
| def get_booking_summary(self) -> Dict[str, Any]: | |
| return { | |
| "total_bookings": len(self.bookings), | |
| "bookings": list(self.bookings.values()), | |
| "booking_types": { | |
| "flight": len( | |
| [b for b in self.bookings.values() if b["type"] == "flight"] | |
| ), | |
| "hotel": len( | |
| [b for b in self.bookings.values() if b["type"] == "hotel"] | |
| ), | |
| }, | |
| } | |
| def clear_bookings(self): | |
| self.bookings.clear() | |
| self.booking_counter = 0 | |
| self.logger.info("All bookings cleared") | |
| def get_environment_stats(self) -> Dict[str, Any]: | |
| total_flights = sum( | |
| len(dest_flights) for dest_flights in self.flight_database.values() | |
| ) | |
| total_hotels = sum( | |
| len(area_hotels) | |
| for dest_hotels in self.hotel_database.values() | |
| for area_hotels in dest_hotels.values() | |
| ) | |
| total_activities = sum( | |
| len(day_activities) | |
| for dest_activities in self.activity_database.values() | |
| for day_activities in dest_activities.values() | |
| ) | |
| return { | |
| "destinations": len(self.flight_database), | |
| "total_flights": total_flights, | |
| "total_hotels": total_hotels, | |
| "total_activities": total_activities, | |
| "current_bookings": len(self.bookings), | |
| "config": self.config, | |
| } | |
Xet Storage Details
- Size:
- 21 kB
- Xet hash:
- 29b530cb81454dac0d0876d202861ca8c2bc9ceda34d4db3f15c0138968b435a
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.