diff --git a/.github/workflows/build.yaml b/.github/workflows/build.yaml
index 077f45bc..b794ca3b 100644
--- a/.github/workflows/build.yaml
+++ b/.github/workflows/build.yaml
@@ -19,10 +19,10 @@ jobs:
test:
strategy:
matrix:
- # there is no 3.10.15 on mac
+ # there is no 3.10 on mac
# see: https://raw.githubusercontent.com/actions/python-versions/main/versions-manifest.json
os: [ubuntu-latest]
- python-version: ['3.10.15', '3.11']
+ python-version: ['3.10', '3.11']
include:
- os: macos-15-intel # updated intel-based
diff --git a/docs/assets/images/occupancy_comparison.png b/docs/assets/images/occupancy_comparison.png
new file mode 100644
index 00000000..d7a8ad19
Binary files /dev/null and b/docs/assets/images/occupancy_comparison.png differ
diff --git a/docs/assets/plots/occupancy_comparison.html b/docs/assets/plots/occupancy_comparison.html
new file mode 100644
index 00000000..813b0b93
--- /dev/null
+++ b/docs/assets/plots/occupancy_comparison.html
@@ -0,0 +1,14 @@
+
+
+
+
+
+
\ No newline at end of file
diff --git a/smart_control/simulator/occupancy/__init__.py b/smart_control/simulator/occupancy/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/smart_control/simulator/occupancy/enhanced_occupancy.py b/smart_control/simulator/occupancy/enhanced_occupancy.py
new file mode 100644
index 00000000..93150076
--- /dev/null
+++ b/smart_control/simulator/occupancy/enhanced_occupancy.py
@@ -0,0 +1,469 @@
+"""An enhanced stochastic occupancy model for building simulation
+with minute-level control and different worker types. This model is based on the
+LIGHTSWITCHOccupancy model from stochastic_occupancy.py to include minute-level
+control instead of hour-level control. It has the same arrival/departure/lunch
+logic but it provides more fine-grained control and realistic occupant
+behaviour, such as optional weekend work and daily changing work hours and lunch
+break times (instead of a constant set of parameters for each occupant leading
+to a static occupancy profile that repeats itself every weekday of the year).
+The model samples new work and lunch time parameters for each occupant every
+day, caches them for the day and clears the cache on the next day to ensure
+consistency. It is possible to model low occupancy levels on the weekends by
+using different worker types.
+"""
+
+import datetime
+import enum
+from typing import Dict, Union
+
+from absl import logging
+import gin
+import numpy as np
+import pandas as pd
+
+from smart_control.models.base_occupancy import BaseOccupancy
+from smart_control.utils import conversion_utils
+
+# Seeds for np.random.RandomState must be integers in [0, 2**32 - 1].
+# We use modulo SEED_MOD_32 to constrain hashes into this valid range.
+SEED_MOD_32 = 2**32
+
+
+class OccupancyStateEnum(enum.Enum):
+ AWAY = 1
+ WORK = 2
+
+
+class WorkerType(enum.Enum):
+ WEEKDAY_ONLY = 1
+ WEEKEND_REGULAR = 2 # worker types that works on weekends and weekdays
+ WEEKEND_OCCASIONAL = (
+ 3 # regular weekday workers who need to work on weekends occasionally
+ )
+
+
+class MinuteLevelZoneOccupant:
+ """MinuteLevelZoneOccupant with minute-level control and day-specific
+ parameters.
+ This class samples a full daily schedule (arrival, lunch, departure) at minute
+ resolution, caches it per occupant per day, and supports weekend worker types.
+ We intentionally do not inherit from the legacy occupant classes because their
+ semantics differ:
+ - stochastic_occupancy.ZoneOccupant: samples fixed hour‑level times once at
+ initialisation and repeats the same schedule every workday.
+ - randomized_arrival_departure_occupancy.ZoneOccupant: uses independent
+ per-step Bernoulli draws in hour-level arrival/departure windows.
+ Inheritance would require overriding most behaviours and would reduce clarity,
+ so we keep the implementations separate.
+ """
+
+ def __init__(
+ self,
+ earliest_expected_arrival_min: int,
+ latest_expected_arrival_min: int,
+ earliest_expected_departure_min: int,
+ latest_expected_departure_min: int,
+ lunch_start_min: int,
+ lunch_end_min: int,
+ step_size: pd.Timedelta,
+ random_state: np.random.RandomState,
+ time_zone: Union[datetime.tzinfo, str] = "UTC",
+ worker_type: WorkerType = WorkerType.WEEKDAY_ONLY,
+ weekend_work_prob: float = 0.10,
+ occupant_id: int = 0,
+ lunch_duration_min: int = 30,
+ lunch_duration_max: int = 90,
+ ):
+ if not (
+ earliest_expected_arrival_min
+ < latest_expected_arrival_min
+ < earliest_expected_departure_min
+ < latest_expected_departure_min
+ ):
+ raise ValueError(
+ "Expected arrival/departure minutes to satisfy:"
+ " earliest_expected_arrival_min < latest_expected_arrival_min <"
+ " earliest_expected_departure_min < latest_expected_departure_min"
+ f" (got {earliest_expected_arrival_min},"
+ f" {latest_expected_arrival_min}, {earliest_expected_departure_min},"
+ f" {latest_expected_departure_min})."
+ )
+ if not lunch_start_min < lunch_end_min:
+ raise ValueError(
+ f"Expected lunch_start_min < lunch_end_min (got {lunch_start_min} >="
+ f" {lunch_end_min})."
+ )
+
+ self._earliest_expected_arrival_min = earliest_expected_arrival_min
+ self._latest_expected_arrival_min = latest_expected_arrival_min
+ self._earliest_expected_departure_min = earliest_expected_departure_min
+ self._latest_expected_departure_min = latest_expected_departure_min
+ self._lunch_start_min = lunch_start_min
+ self._lunch_end_min = lunch_end_min
+ self._step_size = step_size
+ self._random_state = random_state
+
+ if time_zone is None:
+ raise ValueError(
+ "time_zone must be provided (e.g., 'UTC' or an IANA zone)."
+ )
+ try:
+ _ = pd.Timestamp("2000-01-01", tz=time_zone)
+ except Exception as e:
+ raise ValueError(f"Invalid time_zone: {time_zone!r}") from e
+
+ self._time_zone = time_zone
+ self.state = OccupancyStateEnum.AWAY
+ self.daily_cache = {}
+ self.worker_type = worker_type
+ self.weekend_work_prob = weekend_work_prob
+ self.occupant_id = occupant_id
+ self.daily_work_cache = {}
+ self._lunch_duration_min = lunch_duration_min
+ self._lunch_duration_max = lunch_duration_max
+
+ def _generate_cpf(self, start, end, random_state=None):
+ if random_state is None:
+ random_state = self._random_state
+ values = np.arange(start, end + 1)
+ probabilities = random_state.rand(len(values))
+ cumulative_probabilities = np.cumsum(probabilities / probabilities.sum())
+ return values, cumulative_probabilities
+
+ def _sample_event_time(self, start, end, random_state=None):
+ if random_state is None:
+ random_state = self._random_state
+ values, cumulative_probabilities = self._generate_cpf(
+ start, end, random_state
+ )
+ random_value = random_state.rand()
+ index = np.searchsorted(cumulative_probabilities, random_value)
+ logging.info(
+ "Sampled event time: start=%s, end=%s, value=%s",
+ start,
+ end,
+ values[index],
+ )
+ return values[index]
+
+ def _sample_lunch_duration(self, random_state=None):
+ if random_state is None:
+ random_state = self._random_state
+ values, cumulative_probabilities = self._generate_cpf(
+ self._lunch_duration_min, self._lunch_duration_max, random_state
+ )
+ random_value = random_state.rand()
+ index = np.searchsorted(cumulative_probabilities, random_value)
+ logging.info("Sampled lunch duration: %s minutes", values[index])
+ return values[index]
+
+ def _to_local_time(self, timestamp: pd.Timestamp) -> pd.Timestamp:
+ """Return timestamp localised/converted to this occupant's time zone."""
+ if timestamp.tz is None:
+ return timestamp.tz_localize(self._time_zone)
+ return timestamp.tz_convert(self._time_zone)
+
+ def _get_daily_params(self, timestamp: pd.Timestamp) -> Dict[str, int]:
+ local_timestamp = self._to_local_time(timestamp)
+ date_key = local_timestamp.date()
+
+ if self.daily_cache and list(self.daily_cache.keys())[0] != date_key:
+ self.daily_cache.clear()
+ logging.info(
+ "MinuteLevelZoneOccupant: cleared day cache for new date %s", date_key
+ )
+
+ if date_key in self.daily_cache:
+ return self.daily_cache[date_key]
+
+ day_seed = hash(str(date_key) + str(self.occupant_id) + "daily_params") % (
+ SEED_MOD_32
+ )
+ day_random_state = np.random.RandomState(day_seed)
+
+ arrival = self._sample_event_time(
+ self._earliest_expected_arrival_min,
+ self._latest_expected_arrival_min,
+ day_random_state,
+ )
+ departure = self._sample_event_time(
+ self._earliest_expected_departure_min,
+ self._latest_expected_departure_min,
+ day_random_state,
+ )
+ lunch_start = self._sample_event_time(
+ self._lunch_start_min, self._lunch_end_min, day_random_state
+ )
+ lunch_duration = self._sample_lunch_duration(day_random_state)
+
+ self.daily_cache[date_key] = {
+ "arrival_time": arrival,
+ "departure_time": departure,
+ "lunch_start_time": lunch_start,
+ "lunch_duration": lunch_duration,
+ }
+ return self.daily_cache[date_key]
+
+ def _minutes_since_midnight(self, local_timestamp: pd.Timestamp) -> int:
+ return local_timestamp.hour * 60 + local_timestamp.minute
+
+ def _should_work_today(self, timestamp: pd.Timestamp) -> bool:
+ local_timestamp = self._to_local_time(timestamp)
+
+ day = pd.Timestamp(
+ year=local_timestamp.year,
+ month=local_timestamp.month,
+ day=local_timestamp.day,
+ )
+ date_key = day.date()
+
+ if date_key in self.daily_work_cache:
+ return self.daily_work_cache[date_key]
+
+ if conversion_utils.is_work_day(day):
+ self.daily_work_cache[date_key] = True
+ return True
+
+ if self.worker_type == WorkerType.WEEKDAY_ONLY:
+ self.daily_work_cache[date_key] = False
+ return False
+
+ if self.worker_type == WorkerType.WEEKEND_REGULAR:
+ self.daily_work_cache[date_key] = True
+ return True
+
+ elif self.worker_type == WorkerType.WEEKEND_OCCASIONAL:
+ seed = hash(str(date_key) + str(self.occupant_id)) % SEED_MOD_32
+ random_state = np.random.RandomState(seed)
+ work_today = random_state.rand() < self.weekend_work_prob
+ self.daily_work_cache[date_key] = work_today
+ return work_today
+
+ self.daily_work_cache[date_key] = False
+ return False
+
+ def _occupant_arrived(self, timestamp: pd.Timestamp) -> bool:
+ local_timestamp = self._to_local_time(timestamp)
+
+ current_min = self._minutes_since_midnight(local_timestamp)
+ params = self._get_daily_params(timestamp)
+
+ arrived = current_min >= params["arrival_time"]
+ logging.info(
+ "Arrival check: hour=%s, arrival_time=%s, arrived=%s",
+ local_timestamp.hour,
+ params["arrival_time"],
+ arrived,
+ )
+ return arrived
+
+ def _occupant_departed(self, timestamp: pd.Timestamp) -> bool:
+ local_timestamp = self._to_local_time(timestamp)
+
+ current_min = self._minutes_since_midnight(local_timestamp)
+ params = self._get_daily_params(timestamp)
+
+ departed = current_min >= params["departure_time"]
+ logging.info(
+ "Departure check: hour=%s, departure_time=%s, departed=%s",
+ local_timestamp.hour,
+ params["departure_time"],
+ departed,
+ )
+ return departed
+
+ def peek(self, current_time: pd.Timestamp) -> OccupancyStateEnum:
+ """Checks the current occupancy state based on the provided timestamp.
+
+ This method determines the occupancy state (AWAY or WORK) based on
+ the current time, considering workdays, weekends, arrival/departure times,
+ and a lunch break.
+
+ Args:
+ current_time: The current timestamp to evaluate.
+
+ Returns:
+ The current `OccupancyStateEnum` (AWAY or WORK).
+ """
+ local_timestamp = self._to_local_time(current_time)
+
+ logging.info(
+ "Peek called: current_time=%s, local_time=%s, state_before=%s",
+ current_time,
+ local_timestamp,
+ self.state,
+ )
+
+ if not self._should_work_today(current_time):
+ self.state = OccupancyStateEnum.AWAY
+ return self.state
+
+ if self._occupant_arrived(current_time) and not self._occupant_departed(
+ current_time
+ ):
+ self.state = OccupancyStateEnum.WORK
+ else:
+ self.state = OccupancyStateEnum.AWAY
+
+ if self.state == OccupancyStateEnum.WORK:
+ current_min = self._minutes_since_midnight(local_timestamp)
+ params = self._get_daily_params(current_time)
+ lunch_start = params["lunch_start_time"]
+ lunch_end = lunch_start + params["lunch_duration"]
+ if lunch_start <= current_min < lunch_end:
+ self.state = OccupancyStateEnum.AWAY
+ return OccupancyStateEnum.AWAY
+
+ logging.info("Peek result state=%s", self.state)
+
+ return self.state
+
+
+@gin.configurable
+class EnhancedOccupancy(BaseOccupancy):
+ """Enhanced occupancy model with minute-level control and different
+ worker types.
+ """
+
+ def __init__(
+ self,
+ zone_assignment: int,
+ earliest_expected_arrival_hour: int,
+ latest_expected_arrival_hour: int,
+ earliest_expected_departure_hour: int,
+ latest_expected_departure_hour: int,
+ lunch_start_hour: int = 12,
+ lunch_end_hour: int = 14,
+ time_step: pd.Timedelta = pd.Timedelta(minutes=5),
+ time_zone: str = "UTC",
+ # 5% of the workforce are regular weekend workers
+ weekend_regular_pct: float = 0.05,
+ # 5% of the workforce are occasional weekend worker
+ weekend_occasional_pct: float = 0.05,
+ # 10% chance per weekend day that an occasional worker will work
+ occasional_daily_prob: float = 0.10,
+ ):
+ self._zone_assignment = zone_assignment
+ self._zone_occupants = {}
+ self._step_size = time_step
+ self._earliest_expected_arrival = earliest_expected_arrival_hour * 60
+ self._latest_expected_arrival = latest_expected_arrival_hour * 60
+ self._earliest_expected_departure = earliest_expected_departure_hour * 60
+ self._latest_expected_departure = latest_expected_departure_hour * 60
+ self._lunch_start = lunch_start_hour * 60
+ self._lunch_end = lunch_end_hour * 60
+ self._time_zone = time_zone
+ self._weekend_regular_pct = weekend_regular_pct
+ self._weekend_occasional_pct = weekend_occasional_pct
+ self._occasional_prob = occasional_daily_prob
+
+ total_pct = weekend_regular_pct + weekend_occasional_pct
+ if total_pct > 1.0:
+ raise ValueError(
+ "Total percentage of weekend workers must be less than or equal to 1"
+ )
+
+ def _initialize_zone(self, zone_id: str):
+ if zone_id not in self._zone_occupants:
+ self._zone_occupants[zone_id] = []
+ for i in range(self._zone_assignment):
+ worker_random_state = np.random.RandomState(
+ hash(f"{zone_id}_{i}") % SEED_MOD_32
+ )
+ u = worker_random_state.rand()
+ if u < self._weekend_regular_pct:
+ worker_type = WorkerType.WEEKEND_REGULAR
+ weekend_prob = 1.0
+ elif u < self._weekend_regular_pct + self._weekend_occasional_pct:
+ worker_type = WorkerType.WEEKEND_OCCASIONAL
+ weekend_prob = self._occasional_prob
+ else:
+ worker_type = WorkerType.WEEKDAY_ONLY
+ weekend_prob = 0.0
+
+ occupant_random_state = np.random.RandomState(
+ (hash(f"{zone_id}_{i}_behaviour") % SEED_MOD_32)
+ )
+
+ self._zone_occupants[zone_id].append(
+ MinuteLevelZoneOccupant(
+ self._earliest_expected_arrival,
+ self._latest_expected_arrival,
+ self._earliest_expected_departure,
+ self._latest_expected_departure,
+ self._lunch_start,
+ self._lunch_end,
+ self._step_size,
+ occupant_random_state,
+ self._time_zone,
+ worker_type=worker_type,
+ weekend_work_prob=weekend_prob,
+ occupant_id=i,
+ )
+ )
+
+ def average_zone_occupancy(
+ self, zone_id: str, start_time: pd.Timestamp, end_time: pd.Timestamp
+ ) -> float:
+ """Calculates the average occupancy within a time interval for a zone.
+
+ Args:
+ zone_id: specific zone identifier for the building.
+ start_time: **local time** with TZ for the beginning of the interval.
+ end_time: **local time** with TZ for the end of the interval.
+
+ Raises:
+ ValueError: If start_time or end_time is timezone-naive, or if end_time
+ is not after start_time.
+
+ Returns:
+ Average number of people in the zone for the interval.
+ """
+ self._initialize_zone(zone_id)
+
+ if start_time.tz is None or end_time.tz is None:
+ raise ValueError("start_time and end_time must be timezone-aware.")
+ if start_time >= end_time:
+ raise ValueError("end_time must be after start_time.")
+
+ current_time = start_time
+ total_occupancy = 0
+ steps = 0
+
+ while current_time < end_time:
+ num_occupants = 0
+ for occupant in self._zone_occupants[zone_id]:
+ state = occupant.peek(current_time)
+ if state == OccupancyStateEnum.WORK:
+ num_occupants += 1
+
+ total_occupancy += num_occupants
+ steps += 1
+ current_time += self._step_size
+
+ return total_occupancy / steps if steps > 0 else 0.0
+
+ def get_worker_distribution(self, zone_id: str) -> Dict[str, int]:
+ """Returns the distribution of worker types in the given zone.
+
+ Args:
+ zone_id: The specific zone identifier for the building.
+
+ Returns:
+ A dictionary with counts for each worker type:
+ {
+ "weekday_only": int,
+ "weekend_regular": int,
+ "weekend_occasional": int,
+ }.
+ """
+ self._initialize_zone(zone_id)
+ counts = {"weekday_only": 0, "weekend_regular": 0, "weekend_occasional": 0}
+ for occupant in self._zone_occupants[zone_id]:
+ if occupant.worker_type == WorkerType.WEEKDAY_ONLY:
+ counts["weekday_only"] += 1
+ elif occupant.worker_type == WorkerType.WEEKEND_REGULAR:
+ counts["weekend_regular"] += 1
+ elif occupant.worker_type == WorkerType.WEEKEND_OCCASIONAL:
+ counts["weekend_occasional"] += 1
+ return counts
diff --git a/smart_control/simulator/occupancy/enhanced_occupancy_test.py b/smart_control/simulator/occupancy/enhanced_occupancy_test.py
new file mode 100644
index 00000000..53379d99
--- /dev/null
+++ b/smart_control/simulator/occupancy/enhanced_occupancy_test.py
@@ -0,0 +1,282 @@
+"""Test the enhanced occupancy model."""
+
+from absl.testing import absltest
+from absl.testing import parameterized
+import numpy as np
+import pandas as pd
+
+from smart_control.simulator.occupancy.enhanced_occupancy import EnhancedOccupancy
+from smart_control.simulator.occupancy.enhanced_occupancy import MinuteLevelZoneOccupant
+from smart_control.simulator.occupancy.enhanced_occupancy import OccupancyStateEnum
+from smart_control.simulator.occupancy.enhanced_occupancy import WorkerType
+
+DEBUG_PRINT = False
+SEED = 511211
+TIME_STEP = pd.Timedelta(minutes=5)
+EARLIEST_EXPECTED_ARRIVAL_HOUR = 8
+LATEST_EXPECTED_ARRIVAL_HOUR = 10
+EARLIEST_EXPECTED_DEPARTURE_HOUR = 16
+LATEST_EXPECTED_DEPARTURE_HOUR = 18
+LUNCH_START_HOUR = 12
+LUNCH_END_HOUR = 14
+NUM_OCCUPANTS = 10
+DEFAULT_TIMEZONE = 'UTC'
+NO_WEEKEND_WORKERS_REGULAR_PCT = 0.0
+NO_WEEKEND_WORKERS_OCCASIONAL_PCT = 0.0
+NO_WEEKEND_WORKERS_DAILY_PROB = 0.0
+REGULAR_WEEKEND_WORKERS_REGULAR_PCT = 0.2
+REGULAR_WEEKEND_WORKERS_OCCASIONAL_PCT = 0.0
+REGULAR_WEEKEND_WORKERS_DAILY_PROB = 0.0
+TEST_SETUP_REGULAR_PCT = 0.1
+TEST_SETUP_OCCASIONAL_PCT = 0.2
+TEST_SETUP_DAILY_PROB = 0.3
+
+
+class EnhancedOccupancyTest(parameterized.TestCase):
+
+ @parameterized.parameters('UTC', 'US/Pacific', 'US/Eastern')
+ def test_average_occupancy_weekday(self, tz):
+ occupancy = EnhancedOccupancy(
+ zone_assignment=NUM_OCCUPANTS,
+ earliest_expected_arrival_hour=EARLIEST_EXPECTED_ARRIVAL_HOUR,
+ latest_expected_arrival_hour=LATEST_EXPECTED_ARRIVAL_HOUR,
+ earliest_expected_departure_hour=EARLIEST_EXPECTED_DEPARTURE_HOUR,
+ latest_expected_departure_hour=LATEST_EXPECTED_DEPARTURE_HOUR,
+ lunch_start_hour=LUNCH_START_HOUR,
+ lunch_end_hour=LUNCH_END_HOUR,
+ time_step=TIME_STEP,
+ time_zone=tz,
+ weekend_regular_pct=NO_WEEKEND_WORKERS_REGULAR_PCT,
+ weekend_occasional_pct=NO_WEEKEND_WORKERS_OCCASIONAL_PCT,
+ occasional_daily_prob=NO_WEEKEND_WORKERS_DAILY_PROB,
+ )
+
+ current_time = pd.Timestamp('2021-09-01 00:00', tz=tz)
+ occupancies = []
+ while current_time < pd.Timestamp('2021-09-02 00:00', tz=tz):
+ n = occupancy.average_zone_occupancy(
+ 'zone_0', current_time, current_time + TIME_STEP
+ )
+ occupancies.append(n)
+ current_time += TIME_STEP
+
+ early_morning_avg = np.mean(occupancies[0:48]) # 48 time steps = 4 hours
+ morning_avg = np.mean(occupancies[96:132]) # 8-11
+ lunch_avg = np.mean(occupancies[144:168]) # 12-14
+ afternoon_avg = np.mean(occupancies[180:204]) # 15-17
+ evening_avg = np.mean(occupancies[240:288]) # 20-24
+
+ self.assertEqual(early_morning_avg, 0.0)
+ self.assertEqual(evening_avg, 0.0)
+ self.assertGreater(morning_avg, 0.0)
+ self.assertGreater(afternoon_avg, 0.0)
+ self.assertLess(lunch_avg, NUM_OCCUPANTS)
+
+ def test_weekend_occupancy(self):
+ weekday_only_occupancy = EnhancedOccupancy(
+ zone_assignment=100,
+ earliest_expected_arrival_hour=EARLIEST_EXPECTED_ARRIVAL_HOUR,
+ latest_expected_arrival_hour=LATEST_EXPECTED_ARRIVAL_HOUR,
+ earliest_expected_departure_hour=EARLIEST_EXPECTED_DEPARTURE_HOUR,
+ latest_expected_departure_hour=LATEST_EXPECTED_DEPARTURE_HOUR,
+ lunch_start_hour=LUNCH_START_HOUR,
+ lunch_end_hour=LUNCH_END_HOUR,
+ time_step=TIME_STEP,
+ time_zone=DEFAULT_TIMEZONE,
+ weekend_regular_pct=NO_WEEKEND_WORKERS_REGULAR_PCT,
+ weekend_occasional_pct=NO_WEEKEND_WORKERS_OCCASIONAL_PCT,
+ occasional_daily_prob=NO_WEEKEND_WORKERS_DAILY_PROB,
+ )
+
+ weekend_regular_occupancy = EnhancedOccupancy(
+ zone_assignment=100,
+ earliest_expected_arrival_hour=EARLIEST_EXPECTED_ARRIVAL_HOUR,
+ latest_expected_arrival_hour=LATEST_EXPECTED_ARRIVAL_HOUR,
+ earliest_expected_departure_hour=EARLIEST_EXPECTED_DEPARTURE_HOUR,
+ latest_expected_departure_hour=LATEST_EXPECTED_DEPARTURE_HOUR,
+ lunch_start_hour=LUNCH_START_HOUR,
+ lunch_end_hour=LUNCH_END_HOUR,
+ time_step=TIME_STEP,
+ time_zone=DEFAULT_TIMEZONE,
+ weekend_regular_pct=REGULAR_WEEKEND_WORKERS_REGULAR_PCT,
+ weekend_occasional_pct=REGULAR_WEEKEND_WORKERS_OCCASIONAL_PCT,
+ occasional_daily_prob=REGULAR_WEEKEND_WORKERS_DAILY_PROB,
+ )
+ saturday_morning_start = pd.Timestamp(
+ '2021-09-04 08:00', tz=DEFAULT_TIMEZONE
+ )
+ saturday_morning_end = pd.Timestamp('2021-09-04 12:00', tz=DEFAULT_TIMEZONE)
+ weekday_only_occupancy = weekday_only_occupancy.average_zone_occupancy(
+ 'zone_0', saturday_morning_start, saturday_morning_end
+ )
+ weekend_regular_occupancy = (
+ weekend_regular_occupancy.average_zone_occupancy(
+ 'zone_0', saturday_morning_start, saturday_morning_end
+ )
+ )
+ self.assertEqual(weekday_only_occupancy, 0.0)
+ self.assertGreater(weekend_regular_occupancy, 0.0)
+
+ def test_worker_distribution(self):
+ occupancy = EnhancedOccupancy(
+ zone_assignment=1000,
+ earliest_expected_arrival_hour=EARLIEST_EXPECTED_ARRIVAL_HOUR,
+ latest_expected_arrival_hour=LATEST_EXPECTED_ARRIVAL_HOUR,
+ earliest_expected_departure_hour=EARLIEST_EXPECTED_DEPARTURE_HOUR,
+ latest_expected_departure_hour=LATEST_EXPECTED_DEPARTURE_HOUR,
+ lunch_start_hour=LUNCH_START_HOUR,
+ lunch_end_hour=LUNCH_END_HOUR,
+ time_step=TIME_STEP,
+ time_zone=DEFAULT_TIMEZONE,
+ weekend_regular_pct=TEST_SETUP_REGULAR_PCT,
+ weekend_occasional_pct=TEST_SETUP_OCCASIONAL_PCT,
+ occasional_daily_prob=TEST_SETUP_DAILY_PROB,
+ )
+ distribution = occupancy.get_worker_distribution('zone_0')
+ total_workers = sum(distribution.values())
+ self.assertEqual(total_workers, 1000)
+
+ expected_regular = 1000 * TEST_SETUP_REGULAR_PCT
+ expected_occasional = 1000 * TEST_SETUP_OCCASIONAL_PCT
+ expected_weekday = 1000 * (
+ 1.0 - TEST_SETUP_REGULAR_PCT - TEST_SETUP_OCCASIONAL_PCT
+ )
+ self.assertAlmostEqual(
+ distribution['weekday_only'], expected_weekday, delta=100
+ )
+ self.assertAlmostEqual(
+ distribution['weekend_regular'], expected_regular, delta=100
+ )
+ self.assertAlmostEqual(
+ distribution['weekend_occasional'], expected_occasional, delta=100
+ )
+
+ def test_parameter_variation(self):
+ occupant = MinuteLevelZoneOccupant(
+ earliest_expected_arrival_min=EARLIEST_EXPECTED_ARRIVAL_HOUR * 60,
+ latest_expected_arrival_min=LATEST_EXPECTED_ARRIVAL_HOUR * 60,
+ earliest_expected_departure_min=EARLIEST_EXPECTED_DEPARTURE_HOUR * 60,
+ latest_expected_departure_min=LATEST_EXPECTED_DEPARTURE_HOUR * 60,
+ lunch_start_min=LUNCH_START_HOUR * 60,
+ lunch_end_min=LUNCH_END_HOUR * 60,
+ step_size=TIME_STEP,
+ random_state=np.random.RandomState(seed=SEED),
+ time_zone=DEFAULT_TIMEZONE,
+ worker_type=WorkerType.WEEKDAY_ONLY,
+ weekend_work_prob=NO_WEEKEND_WORKERS_DAILY_PROB,
+ occupant_id=0,
+ )
+ day1_morning = pd.Timestamp('2021-09-01 09:00', tz=DEFAULT_TIMEZONE)
+ day1_afternoon = pd.Timestamp('2021-09-01 15:00', tz=DEFAULT_TIMEZONE)
+ day2_morning = pd.Timestamp('2021-09-02 09:00', tz=DEFAULT_TIMEZONE)
+ params1_morning = occupant._get_daily_params(day1_morning)
+ params1_afternoon = occupant._get_daily_params(day1_afternoon)
+ params2_morning = occupant._get_daily_params(day2_morning)
+ self.assertEqual(params1_morning, params1_afternoon)
+ self.assertNotEqual(params1_morning, params2_morning)
+
+ @parameterized.parameters('UTC', 'US/Eastern', 'US/Pacific')
+ def test_occupant_peek(self, tz):
+ occupant = MinuteLevelZoneOccupant(
+ earliest_expected_arrival_min=EARLIEST_EXPECTED_ARRIVAL_HOUR * 60,
+ latest_expected_arrival_min=LATEST_EXPECTED_ARRIVAL_HOUR * 60,
+ earliest_expected_departure_min=EARLIEST_EXPECTED_DEPARTURE_HOUR * 60,
+ latest_expected_departure_min=LATEST_EXPECTED_DEPARTURE_HOUR * 60,
+ lunch_start_min=LUNCH_START_HOUR * 60,
+ lunch_end_min=LUNCH_END_HOUR * 60,
+ step_size=TIME_STEP,
+ random_state=np.random.RandomState(seed=SEED),
+ time_zone=tz,
+ worker_type=WorkerType.WEEKDAY_ONLY,
+ weekend_work_prob=NO_WEEKEND_WORKERS_DAILY_PROB,
+ occupant_id=0,
+ )
+ day = pd.Timestamp('2021-09-01 00:00', tz=tz)
+ params = occupant._get_daily_params(day)
+ day1_early_morning = pd.Timestamp('2021-09-01 06:00', tz=tz)
+ day1_work_morning = pd.Timestamp('2021-09-01 10:00', tz=tz)
+ day1_afternoon = pd.Timestamp('2021-09-01 15:00', tz=tz)
+ day1_evening = pd.Timestamp('2021-09-01 20:00', tz=tz)
+ weekend = pd.Timestamp('2021-09-05 08:00', tz=tz)
+
+ def expected_state(ts: pd.Timestamp):
+ ts_local = ts.tz_convert(tz)
+ minutes = ts_local.hour * 60 + ts_local.minute
+ in_work = params['arrival_time'] <= minutes < params['departure_time']
+ in_lunch = (
+ params['lunch_start_time']
+ <= minutes
+ < params['lunch_start_time'] + params['lunch_duration']
+ )
+ return (
+ OccupancyStateEnum.WORK
+ if (in_work and not in_lunch)
+ else OccupancyStateEnum.AWAY
+ )
+
+ self.assertEqual(occupant.peek(day1_early_morning), OccupancyStateEnum.AWAY)
+ self.assertEqual(
+ occupant.peek(day1_work_morning), expected_state(day1_work_morning)
+ )
+ self.assertEqual(
+ occupant.peek(day1_afternoon), expected_state(day1_afternoon)
+ )
+ self.assertEqual(occupant.peek(day1_evening), OccupancyStateEnum.AWAY)
+ self.assertEqual(occupant.peek(weekend), OccupancyStateEnum.AWAY)
+
+ def test_occasional_worker(self):
+ occupant = MinuteLevelZoneOccupant(
+ earliest_expected_arrival_min=EARLIEST_EXPECTED_ARRIVAL_HOUR * 60,
+ latest_expected_arrival_min=LATEST_EXPECTED_ARRIVAL_HOUR * 60,
+ earliest_expected_departure_min=EARLIEST_EXPECTED_DEPARTURE_HOUR * 60,
+ latest_expected_departure_min=LATEST_EXPECTED_DEPARTURE_HOUR * 60,
+ lunch_start_min=LUNCH_START_HOUR * 60,
+ lunch_end_min=LUNCH_END_HOUR * 60,
+ step_size=TIME_STEP,
+ random_state=np.random.RandomState(seed=SEED),
+ time_zone=DEFAULT_TIMEZONE,
+ worker_type=WorkerType.WEEKEND_OCCASIONAL,
+ weekend_work_prob=0.5,
+ occupant_id=13,
+ )
+ saturday_morning = pd.Timestamp('2021-09-04 08:00', tz=DEFAULT_TIMEZONE)
+ saturday_afternoon = pd.Timestamp('2021-09-04 15:00', tz=DEFAULT_TIMEZONE)
+ work_decision_morning = occupant._should_work_today(saturday_morning)
+ work_decision_afternoon = occupant._should_work_today(saturday_afternoon)
+ self.assertEqual(work_decision_morning, work_decision_afternoon)
+
+ def test_minutes_precision(self):
+ occupancy = EnhancedOccupancy(
+ zone_assignment=NUM_OCCUPANTS,
+ earliest_expected_arrival_hour=EARLIEST_EXPECTED_ARRIVAL_HOUR,
+ latest_expected_arrival_hour=LATEST_EXPECTED_ARRIVAL_HOUR,
+ earliest_expected_departure_hour=EARLIEST_EXPECTED_DEPARTURE_HOUR,
+ latest_expected_departure_hour=LATEST_EXPECTED_DEPARTURE_HOUR,
+ lunch_start_hour=LUNCH_START_HOUR,
+ lunch_end_hour=LUNCH_END_HOUR,
+ time_step=TIME_STEP,
+ time_zone=DEFAULT_TIMEZONE,
+ weekend_regular_pct=NO_WEEKEND_WORKERS_REGULAR_PCT,
+ weekend_occasional_pct=NO_WEEKEND_WORKERS_OCCASIONAL_PCT,
+ occasional_daily_prob=NO_WEEKEND_WORKERS_DAILY_PROB,
+ )
+ time_758 = pd.Timestamp('2021-09-01 07:58', tz=DEFAULT_TIMEZONE)
+ time_759 = pd.Timestamp('2021-09-01 07:59', tz=DEFAULT_TIMEZONE)
+ morning_start = pd.Timestamp('2021-09-01 08:30', tz=DEFAULT_TIMEZONE)
+ morning_end = pd.Timestamp('2021-09-01 11:30', tz=DEFAULT_TIMEZONE)
+ occ_758 = occupancy.average_zone_occupancy(
+ 'zone_0', time_758, time_758 + pd.Timedelta(minutes=1)
+ )
+ occ_759 = occupancy.average_zone_occupancy(
+ 'zone_0', time_759, time_759 + pd.Timedelta(minutes=1)
+ )
+ morning_occupancy = occupancy.average_zone_occupancy(
+ 'zone_0', morning_start, morning_end
+ )
+ self.assertEqual(occ_758, 0.0)
+ self.assertEqual(occ_759, 0.0)
+ self.assertGreater(morning_occupancy, 0.0)
+
+
+if __name__ == '__main__':
+ absltest.main()
diff --git a/smart_control/simulator/occupancy/randomized_arrival_departure_occupancy.py b/smart_control/simulator/occupancy/randomized_arrival_departure_occupancy.py
new file mode 100644
index 00000000..e7446dd9
--- /dev/null
+++ b/smart_control/simulator/occupancy/randomized_arrival_departure_occupancy.py
@@ -0,0 +1,227 @@
+"""A randomized occupancy model for the building simulation.
+
+In this model, N occupants arrive between the earliest and latest arrival
+hour and depart between the earliest and latest departure hour. The probability
+of departure is specified so that the expected arrival and departure times
+occur halfway in the interval. For a standard Bernoulli RV, E[X] = n*p, so
+p = E[X] / n / 2, where E[X] is the expected number of arrivals, which equals 1.
+"""
+
+import datetime
+import enum
+from typing import Optional, Union
+
+import gin
+import numpy as np
+import pandas as pd
+
+from smart_control.models.base_occupancy import BaseOccupancy
+from smart_control.utils import conversion_utils
+
+
+class OccupancyStateEnum(enum.Enum):
+ AWAY = 1
+ WORK = 2
+
+
+class ZoneOccupant:
+ """Represents a single occupant in a zone.
+
+ Attributes:
+ earliest_expected_arrival_hour: earliest arrivel, 0 - 22
+ latest_expected_arrival_hour: latest arrivel, 1 - 23
+ earliest_expected_departure_hour: earliest departure, 0 - 22
+ latest_expected_departure_hour: latest departure, 1 - 23
+ random_state: random state used to generate events
+ """
+
+ def __init__(
+ self,
+ earliest_expected_arrival_hour: int,
+ latest_expected_arrival_hour: int,
+ earliest_expected_departure_hour: int,
+ latest_expected_departure_hour: int,
+ step_size: pd.Timedelta,
+ random_state: np.random.RandomState,
+ time_zone: Union[datetime.tzinfo, str] = "UTC",
+ ):
+
+ if not (
+ earliest_expected_arrival_hour
+ < latest_expected_arrival_hour
+ < earliest_expected_departure_hour
+ < latest_expected_departure_hour
+ ):
+ raise ValueError(
+ "Arrival and departure hours must be strictly increasing: "
+ "earliest_arrival < latest_arrival < earliest_departure < "
+ "latest_departure. "
+ f"Got: {earliest_expected_arrival_hour}, "
+ f"{latest_expected_arrival_hour}, "
+ f"{earliest_expected_departure_hour}, "
+ f"{latest_expected_departure_hour}."
+ )
+
+ self._earliest_expected_arrival_hour = earliest_expected_arrival_hour
+ self._latest_expected_arrival_hour = latest_expected_arrival_hour
+ self._earliest_expected_departure_hour = earliest_expected_departure_hour
+ self._latest_expected_departure_hour = latest_expected_departure_hour
+ self._step_size = step_size
+ self._occupancy_state = OccupancyStateEnum.AWAY
+ self._p_arrival = self._get_event_probability(
+ earliest_expected_arrival_hour, latest_expected_arrival_hour
+ )
+ self._p_departure = self._get_event_probability(
+ earliest_expected_departure_hour, latest_expected_departure_hour
+ )
+ self._random_state = random_state
+ self._time_zone = time_zone
+
+ def _to_local_time(self, timestamp: pd.Timestamp) -> pd.Timestamp:
+ """Converts timestamp to local time."""
+ if timestamp.tz is None:
+ return timestamp
+ else:
+ return timestamp.tz_convert(self._time_zone)
+
+ def _get_event_probability(self, start_hour, end_hour):
+ """Returns the probability of an event based on the number of time steps."""
+
+ if start_hour >= end_hour:
+ raise ValueError(
+ "Start hour must be less than end hour to calculate event "
+ f"probability: start_hour={start_hour}, end_hour={end_hour}"
+ )
+
+ # The window is the number of Bernoulli trials (i.e. tests for arrival).
+ window = pd.Timedelta(end_hour - start_hour, unit="hour")
+ # The halfway point is the firts half of the trials.
+ n_halfway = window / self._step_size / 2.0
+ # We'd like to return the probability of event happening in a single time-
+ # step. This follow a geometric distribution, where E[X] = 1/p, where
+ # E[x] is the expected number of events before the first success. If
+ # E[X] is the halfway point, then p = 1 / n_halfway.
+ return 1.0 / n_halfway
+
+ def _occupant_arrived(self, timestamp):
+ """Makes a random draw to determine whether occupant arrives."""
+
+ local_timestamp = self._to_local_time(timestamp)
+ # TODO(sipple): Consider effects when time crosses DST>
+ if (
+ local_timestamp.hour < self._earliest_expected_arrival_hour
+ or local_timestamp.hour > self._latest_expected_arrival_hour
+ ):
+ return False
+ else:
+ return self._random_state.rand() < self._p_arrival
+
+ def _occupant_departed(self, timestamp):
+ """Makes a random draw to determine whether the occupant departs."""
+ local_timestamp = self._to_local_time(timestamp)
+ if local_timestamp.hour < self._earliest_expected_departure_hour:
+ return False
+ else:
+ return self._random_state.rand() < self._p_departure
+
+ def peek(self, current_time: pd.Timestamp) -> OccupancyStateEnum:
+ """Returns the state (WORK or AWAY) of the occupant for the current_time."""
+
+ local_timestamp = self._to_local_time(current_time)
+ day = pd.Timestamp(
+ year=local_timestamp.year,
+ month=local_timestamp.month,
+ day=local_timestamp.day,
+ )
+
+ if not conversion_utils.is_work_day(day):
+ self._occupancy_state = OccupancyStateEnum.AWAY
+
+ elif self._occupancy_state == OccupancyStateEnum.AWAY:
+ if self._occupant_arrived(current_time):
+ self._occupancy_state = OccupancyStateEnum.WORK
+
+ else:
+ if self._occupant_departed(current_time):
+ self._occupancy_state = OccupancyStateEnum.AWAY
+
+ return self._occupancy_state
+
+
+@gin.configurable
+class RandomizedArrivalDepartureOccupancy(BaseOccupancy):
+ """Provides the RL agent information about how many people are in a zone.
+
+ Attributes:
+ zone_assignment: number of occupants in a zone
+ earliest_expected_arrival_hour: earliest arrivel, 0 - 22
+ latest_expected_arrival_hour: latest arrivel, 1 - 23
+ earliest_expected_departure_hour: earliest departure, 0 - 22
+ latest_expected_departure_hour: latest departure, 1 - 23
+ seed: integer used to set the random state for repeatability
+ """
+
+ def __init__(
+ self,
+ zone_assignment: int,
+ earliest_expected_arrival_hour: int,
+ latest_expected_arrival_hour: int,
+ earliest_expected_departure_hour: int,
+ latest_expected_departure_hour: int,
+ time_step_sec: int,
+ seed: Optional[int] = 17321,
+ time_zone: str = "UTC",
+ ):
+ self._zone_assignment = zone_assignment
+ self._zone_occupants = {}
+ self._step_size = pd.Timedelta(time_step_sec, unit="second")
+ self._earliest_expected_arrival_hour = earliest_expected_arrival_hour
+ self._latest_expected_arrival_hour = latest_expected_arrival_hour
+ self._earliest_expected_departure_hour = earliest_expected_departure_hour
+ self._latest_expected_departure_hour = latest_expected_departure_hour
+ self._random_state = np.random.RandomState(seed)
+ self._time_zone = time_zone
+
+ def average_zone_occupancy(
+ self, zone_id: str, start_time: pd.Timestamp, end_time: pd.Timestamp
+ ) -> float:
+ """Returns the occupancy within start_time, end_time for the zone.
+
+ If the zone is not found, implementations should raise a ValueError.
+
+ Args:
+ zone_id: specific zone identifier for the building.
+ start_time: **local time** w/ TZ for the beginning of the interval.
+ end_time: **local time** w/ TZ for the end of the interval.
+
+ Returns:
+ average number of people in the zone for the interval.
+ """
+
+ if zone_id not in self._zone_occupants:
+ self._zone_occupants[zone_id] = []
+ for _ in range(self._zone_assignment):
+ self._zone_occupants[zone_id].append(
+ ZoneOccupant(
+ self._earliest_expected_arrival_hour,
+ self._latest_expected_arrival_hour,
+ self._earliest_expected_departure_hour,
+ self._latest_expected_departure_hour,
+ self._step_size,
+ self._random_state,
+ self._time_zone,
+ )
+ )
+
+ current_time = start_time
+ total_occupants = 0.0
+ steps = 0
+ while current_time < end_time:
+ num_occupants = 0.0
+ for occupant in self._zone_occupants[zone_id]:
+ if occupant.peek(current_time) == OccupancyStateEnum.WORK:
+ num_occupants += 1.0
+ total_occupants += num_occupants
+ steps += 1
+ current_time += self._step_size
+ return total_occupants / steps if steps > 0 else 0.0
diff --git a/smart_control/simulator/occupancy/randomized_arrival_departure_occupancy_test.py b/smart_control/simulator/occupancy/randomized_arrival_departure_occupancy_test.py
new file mode 100644
index 00000000..6b705627
--- /dev/null
+++ b/smart_control/simulator/occupancy/randomized_arrival_departure_occupancy_test.py
@@ -0,0 +1,216 @@
+"""Tests for randomized_arrival_departure_occupancy."""
+
+from absl.testing import absltest
+from absl.testing import parameterized
+import numpy as np
+import pandas as pd
+
+from smart_control.simulator.occupancy import randomized_arrival_departure_occupancy
+from smart_control.simulator.occupancy.randomized_arrival_departure_occupancy import OccupancyStateEnum
+from smart_control.simulator.occupancy.randomized_arrival_departure_occupancy import RandomizedArrivalDepartureOccupancy
+
+# fmt: off
+# pylint: disable=bad-continuation
+_EXPECTED_ZONE_OCCUPANCIES_PACIFIC = [
+ 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0,
+ 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0,
+ 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0,
+ 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0,
+ 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0,
+ 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0,
+ 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0,
+ 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0,
+ 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0,
+ 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0,
+ 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0,
+ 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 2.0, 2.0, 2.0, 2.0, 4.0, 4.0, 4.0, 4.0, 4.0,
+ 4.0, 5.0, 5.0, 5.0, 5.0, 6.0, 6.0, 6.0, 6.0, 7.0, 7.0, 7.0, 7.0, 8.0, 8.0,
+ 8.0, 8.0, 8.0, 8.0, 8.0, 8.0, 8.0, 8.0, 8.0, 8.0, 8.0, 8.0, 8.0, 8.0, 8.0,
+ 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0,
+ 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0,
+ 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0,
+ 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0,
+ 9.0, 9.0, 9.0, 9.0, 9.0, 9.0,
+]
+
+_EXPECTED_ZONE_OCCUPANCIES_EASTERN = [
+ 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0,
+ 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0,
+ 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0,
+ 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0,
+ 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0,
+ 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0,
+ 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0,
+ 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0,
+ 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0,
+ 2.0, 2.0, 2.0, 2.0, 4.0, 4.0, 4.0, 4.0, 4.0, 4.0, 5.0, 5.0, 5.0, 5.0, 6.0,
+ 6.0, 6.0, 6.0, 7.0, 7.0, 7.0, 7.0, 8.0, 8.0, 8.0, 8.0, 8.0, 8.0, 8.0, 8.0,
+ 8.0, 8.0, 8.0, 8.0, 8.0, 8.0, 8.0, 8.0, 8.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0,
+ 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0,
+ 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0,
+ 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0,
+ 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0, 9.0,
+ 9.0, 8.0, 8.0, 8.0, 8.0, 7.0, 7.0, 7.0, 5.0, 5.0, 5.0, 5.0, 5.0, 5.0, 5.0,
+ 5.0, 5.0, 5.0, 4.0, 4.0, 4.0, 4.0, 3.0, 3.0, 3.0, 3.0, 3.0, 3.0, 3.0, 3.0,
+ 3.0, 3.0, 3.0, 3.0, 3.0, 3.0,
+]
+
+_EXPECTED_ZONE_OCCUPANCIES_UTC = [
+ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
+ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
+ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
+ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
+ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
+ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 2, 2, 2,
+ 2, 4, 4, 4, 4, 4, 4, 5, 5, 5, 5, 6, 6, 6, 6,
+ 7, 7, 7, 7, 8, 8, 8, 8, 8, 8, 8, 8, 8, 8, 8,
+ 8, 8, 8, 8, 8, 8, 9, 9, 9, 9, 9, 9, 9, 9, 9,
+ 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9,
+ 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9,
+ 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9,
+ 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 8, 8,
+ 8, 8, 7, 7, 7, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5,
+ 4, 4, 4, 4, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3,
+ 3, 3, 3, 3, 3, 3, 3, 2, 2, 2, 1, 1, 1, 1, 1,
+ 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1,
+ 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1,
+ 1, 1, 1, 1, 1, 1,
+]
+# pylint: disable=bad-continuation
+# fmt: off
+
+
+class RandomizedArrivalDepartureOccupancyTest(parameterized.TestCase):
+
+ @parameterized.parameters(
+ (None, _EXPECTED_ZONE_OCCUPANCIES_UTC),
+ ('UTC', _EXPECTED_ZONE_OCCUPANCIES_UTC),
+ ('US/Pacific', _EXPECTED_ZONE_OCCUPANCIES_PACIFIC),
+ ('US/Eastern', _EXPECTED_ZONE_OCCUPANCIES_EASTERN),
+ )
+ def test_average_zone_occupancy_weekday(self, tz, expected_zone_occupancies):
+ step_size = pd.Timedelta(5, unit='minute')
+
+ occupancy = randomized_arrival_departure_occupancy.RandomizedArrivalDepartureOccupancy( # pylint: disable=line-too-long
+ 10, 7, 11, 15, 20, step_size.total_seconds(), 511211, tz
+ )
+ current_time = pd.Timestamp('2021-09-01 00:00', tz='UTC')
+ actual_occupancies = []
+ while current_time < pd.Timestamp('2021-09-01 23:00', tz='UTC'):
+ n = occupancy.average_zone_occupancy(
+ 'zone_0', current_time, current_time + step_size
+ )
+ actual_occupancies.append(n)
+
+ current_time += step_size
+ self.assertSequenceEqual(expected_zone_occupancies, actual_occupancies)
+
+ def test_get_event_probability(self):
+ occupant = randomized_arrival_departure_occupancy.ZoneOccupant(
+ 8,
+ 12,
+ 13,
+ 18,
+ pd.Timedelta(5, unit='minute'),
+ np.random.RandomState(seed=55213),
+ )
+ p = occupant._get_event_probability(8, 12)
+ self.assertEqual(p, 1.0 / 24.0)
+
+ @parameterized.parameters((None), 'UTC', 'US/Eastern', 'US/Pacific')
+ def test_peek(self, tz):
+ random_state = np.random.RandomState(seed=55213)
+ occupant = randomized_arrival_departure_occupancy.ZoneOccupant(
+ 8, 12, 13, 18, pd.Timedelta(5, unit='minute'), random_state, tz
+ )
+ current_time = pd.Timestamp('2021-09-01 00:00', tz=tz)
+ while current_time < pd.Timestamp('2021-09-01 23:00', tz=tz):
+ state = occupant.peek(current_time=current_time)
+
+ if current_time < pd.Timestamp(
+ '2021-09-01 11:10', tz=tz
+ ) or current_time >= pd.Timestamp('2021-09-01 17:00', tz=tz):
+ self.assertEqual(
+ randomized_arrival_departure_occupancy.OccupancyStateEnum.AWAY,
+ state,
+ )
+ else:
+ self.assertEqual(
+ randomized_arrival_departure_occupancy.OccupancyStateEnum.WORK,
+ state,
+ )
+ current_time += pd.Timedelta(5, unit='minute')
+
+ def test_zone_occupant_invalid_hour_order(self):
+ """ValueError when arrival/departure hours are not strictly increasing."""
+ random_state = np.random.RandomState(seed=55213)
+ step_size = pd.Timedelta(5, unit='minute')
+
+ # latest_arrival >= earliest_departure is invalid
+ with self.assertRaisesRegex(
+ ValueError, 'Arrival and departure hours must be strictly increasing'
+ ):
+ randomized_arrival_departure_occupancy.ZoneOccupant(
+ earliest_expected_arrival_hour=8,
+ latest_expected_arrival_hour=14, # > earliest_departure (13)
+ earliest_expected_departure_hour=13,
+ latest_expected_departure_hour=18,
+ step_size=step_size,
+ random_state=random_state,
+ )
+
+ def test_get_event_probability_invalid_hours(self):
+ """ValueError when start_hour >= end_hour."""
+ random_state = np.random.RandomState(seed=55213)
+ step_size = pd.Timedelta(5, unit='minute')
+ occupant = randomized_arrival_departure_occupancy.ZoneOccupant(
+ earliest_expected_arrival_hour=8,
+ latest_expected_arrival_hour=12,
+ earliest_expected_departure_hour=13,
+ latest_expected_departure_hour=18,
+ step_size=step_size,
+ random_state=random_state,
+ )
+
+ with self.assertRaisesRegex(
+ ValueError, 'Start hour must be less than end hour'
+ ):
+ occupant._get_event_probability(start_hour=12, end_hour=8)
+
+ def test_average_zone_occupancy_matches_manual_two_steps(self):
+ """average_zone_occupancy should equal the mean of per-step counts."""
+ step = pd.Timedelta(minutes=5)
+ tz = 'UTC'
+
+ occ = RandomizedArrivalDepartureOccupancy(
+ zone_assignment=7,
+ earliest_expected_arrival_hour=8,
+ latest_expected_arrival_hour=12,
+ earliest_expected_departure_hour=16,
+ latest_expected_departure_hour=20,
+ time_step_sec=step.total_seconds(),
+ seed=55213,
+ time_zone=tz,
+ )
+
+ t0 = pd.Timestamp('2021-09-01 10:00', tz=tz)
+ t1 = t0 + 2 * step
+
+ # initialise the zone
+ _ = occ.average_zone_occupancy('zone_0', t0, t0 + step)
+
+ manual_counts = []
+ for cur in (t0, t0 + step):
+ c = 0.0
+ for zocc in occ._zone_occupants['zone_0']:
+ if zocc.peek(cur) == OccupancyStateEnum.WORK:
+ c += 1.0
+ manual_counts.append(c)
+ manual_avg = sum(manual_counts) / 2.0
+
+ result = occ.average_zone_occupancy('zone_0', t0, t1)
+ self.assertEqual(result, manual_avg)
+
+
+if __name__ == '__main__':
+ absltest.main()
diff --git a/smart_control/simulator/occupancy/step_function_occupancy.py b/smart_control/simulator/occupancy/step_function_occupancy.py
new file mode 100644
index 00000000..9de06f7c
--- /dev/null
+++ b/smart_control/simulator/occupancy/step_function_occupancy.py
@@ -0,0 +1,160 @@
+"""A very basic occupancy model for smart buildings smart control.
+
+Occupancy is the average number of people in a zone during a specific interval.
+
+Occupancy is an input to the RL agent reward function.
+
+This implementation assumes:
+ (a) all zones have the same occupancy patterns.
+ (b) the occupancy is constant for work periods, and
+ (c) the occupancy is constant for non-work periods (off hours, weekends and
+ holidays).
+"""
+
+from typing import Tuple
+
+import gin
+import pandas as pd
+
+from smart_control.models.base_occupancy import BaseOccupancy
+from smart_control.utils import conversion_utils
+
+
+@gin.configurable
+class StepFunctionOccupancy(BaseOccupancy):
+ """An occupancy model with constant level for work and non-work times.
+
+ This model ignores the specific zone and returns the same value for all zones.
+
+ Attributes:
+ work_start_time: time-of-day when work period starts
+ work_end_time: time-of-day when work period ends
+ work_occupancy: avg number of people in zone during work times
+ nonwork_occupancy: avg number of people in zone during non-work times
+ """
+
+ def __init__(
+ self,
+ work_start_time: pd.Timedelta,
+ work_end_time: pd.Timedelta,
+ work_occupancy: float,
+ nonwork_occupancy: float,
+ ):
+ self._check_times(work_start_time)
+ self._check_times(work_end_time)
+ self._work_start_time = work_start_time
+ self._work_end_time = work_end_time
+ self._work_occupancy = work_occupancy
+ self._nonwork_occupancy = nonwork_occupancy
+
+ def average_zone_occupancy(
+ self, zone_id: str, start_time: pd.Timestamp, end_time: pd.Timestamp
+ ) -> float:
+ """Returns the occupancy within start_time, end_time for the zone.
+
+ This model applies a constant for work time and non-work times, and returns
+ a weighted average. Evaluates weekends and holidays as non-working times.
+
+ Ignores the zone - assumes each zone is equally occupied.
+
+ Args:
+ zone_id: specific zone identifier for the building.
+ start_time: **local time** w/o TZ for the beginning of the interval.
+ end_time: **local time** w/o TZ for the end of the interval.
+
+ Returns:
+ average number of people in the zone for the interval.
+ """
+
+ if start_time >= end_time:
+ raise ValueError('End time may not occur before start time.')
+
+ work_seconds = 0.0
+ nonwork_seconds = 0.0
+
+ # Get the timestamp for midnight of the first day.
+ day = pd.Timestamp(
+ year=start_time.year, month=start_time.month, day=start_time.day
+ )
+ current_time = start_time - day
+
+ # Accumulate working and non-working hours for all days.
+ while day <= end_time:
+ day_end = min(pd.Timedelta(1, unit='day'), end_time - day)
+
+ if conversion_utils.is_work_day(day):
+ before_work, during_work, after_work = self._split_workday(
+ current_time, day_end
+ )
+ work_seconds += during_work
+ nonwork_seconds += before_work + after_work
+
+ else: # Holidays and weekends
+ nonwork_seconds += (day_end - current_time).total_seconds()
+
+ day += pd.Timedelta(1.0, unit='day')
+ current_time = pd.Timedelta(0.0, unit='sec')
+
+ # Compute the weighted average.
+ return (
+ work_seconds * self._work_occupancy
+ + nonwork_seconds * self._nonwork_occupancy
+ ) / (work_seconds + nonwork_seconds)
+
+ def _split_workday(
+ self, start_time: pd.Timedelta, end_time: pd.Timedelta
+ ) -> Tuple[float, float, float]:
+ """Splits the interval into sec before, during, and after working hours.
+
+ Args:
+ start_time: start of the interval.
+ end_time: end of the interval, may be greater than .
+
+ Returns:
+ Tuple (seconds before work, seconds during work, seconds after work)
+ """
+
+ if start_time > end_time:
+ raise ValueError('Cannot have an end time before start time.')
+ self._check_times(start_time)
+
+ before_work = 0.0
+ during_work = 0.0
+ after_work = 0.0
+
+ # Step through before work, during work, and afterwork phases, and
+ # add up the times in seconds. If the current time is before any
+ # phase, do not add any time, and truncate.
+ # If the end_time precedes any phase, truncate the phase, and do not
+ # add any time after the current phase.
+ current = start_time
+ # Don't go beyond the day or the specified end time, whichever is sooner.
+ interval_end = min(end_time, pd.Timedelta(24, unit='hour'))
+ # The next step to check is either the work start time or interval end.
+ # Before work start:
+ next_step = min(interval_end, self._work_start_time)
+
+ # Get the time between the curren time and the next step.
+ if current < next_step:
+ before_work = (next_step - current).total_seconds()
+ current = max(current, next_step)
+
+ # After work start and work end:
+ next_step = min(interval_end, self._work_end_time)
+ if current < next_step:
+ during_work = (next_step - current).total_seconds()
+ current = next_step
+
+ # Finally, between work end and modnight.
+ next_step = interval_end
+ if current < next_step:
+ after_work = (next_step - current).total_seconds()
+
+ return (before_work, during_work, after_work)
+
+ def _check_times(self, time_delta: pd.Timedelta) -> None:
+ if (
+ time_delta > pd.Timedelta(24, unit='hour')
+ or time_delta.total_seconds() < 0.0
+ ):
+ raise ValueError('Time delta must be positive and less than one day.')
diff --git a/smart_control/simulator/occupancy/step_function_occupancy_test.py b/smart_control/simulator/occupancy/step_function_occupancy_test.py
new file mode 100644
index 00000000..c55410bc
--- /dev/null
+++ b/smart_control/simulator/occupancy/step_function_occupancy_test.py
@@ -0,0 +1,168 @@
+"""Tests for step_function_occupancy_model."""
+
+from absl.testing import absltest
+from absl.testing import parameterized
+import pandas as pd
+
+from smart_control.simulator.occupancy import step_function_occupancy
+
+
+class StepFunctionOccupancyModelTest(parameterized.TestCase):
+
+ @parameterized.named_parameters(
+ (
+ "full day",
+ pd.Timedelta("00:00:00"),
+ pd.Timedelta(25, unit="hour"),
+ (8 * 3600.0, 8 * 3600.0, 8 * 3600.0),
+ ),
+ (
+ "workday, 20 min",
+ pd.Timedelta("15:10:00"),
+ pd.Timedelta("15:30:00"),
+ (0.0, 1200.0, 0.0),
+ ),
+ (
+ "before workday, 30 min",
+ pd.Timedelta("03:15:00"),
+ pd.Timedelta("03:45:00"),
+ (1800.0, 0.0, 0.0),
+ ),
+ (
+ "after workday, 120 min",
+ pd.Timedelta("18:15:00"),
+ pd.Timedelta("20:15:00"),
+ (0.0, 0.0, 7200.0),
+ ),
+ (
+ "during 20 min, after workday 120 min",
+ pd.Timedelta("15:40:00"),
+ pd.Timedelta("18:00:00"),
+ (0.0, 1200.0, 7200.0),
+ ),
+ (
+ "before 1 min, during workday 1 min",
+ pd.Timedelta("07:59:00"),
+ pd.Timedelta("08:01:00"),
+ (60.0, 60.0, 0.0),
+ ),
+ )
+ def test_split_workday_valid(self, start_time, end_time, expected_split):
+ occupancy = step_function_occupancy.StepFunctionOccupancy(
+ pd.Timedelta("08:00:00"), pd.Timedelta("16:00:00"), 10.0, 0.2
+ )
+ split = occupancy._split_workday(start_time, end_time)
+ self.assertEqual(expected_split, split)
+
+ def test_split_workday_invalid(self):
+ occupancy = step_function_occupancy.StepFunctionOccupancy(
+ pd.Timedelta("08:00:00"), pd.Timedelta("16:00:00"), 10.0, 0.2
+ )
+ with self.assertRaises(ValueError):
+ _ = occupancy._split_workday(
+ pd.Timedelta("10:00:00"), pd.Timedelta("09:59:00")
+ )
+
+ with self.assertRaises(ValueError):
+ _ = occupancy._split_workday(
+ pd.Timedelta(25, unit="hours"), pd.Timedelta("09:59:00")
+ )
+
+ @parameterized.named_parameters(
+ (
+ "full day",
+ pd.Timestamp("2021-05-10 00:00:00"),
+ pd.Timestamp("2021-05-11 00:00:00"),
+ 3.46666,
+ ),
+ (
+ "two days",
+ pd.Timestamp("2021-05-10 00:00:00"),
+ pd.Timestamp("2021-05-12 00:00:00"),
+ 3.46666,
+ ),
+ (
+ "full day holiday",
+ pd.Timestamp("2022-12-25 00:00:00"),
+ pd.Timestamp("2022-12-26 00:00:00"),
+ 0.2,
+ ),
+ (
+ "full day weekend",
+ pd.Timestamp("2021-05-09 00:00:00"),
+ pd.Timestamp("2021-05-10 00:00:00"),
+ 0.2,
+ ),
+ (
+ "split before",
+ pd.Timestamp("2021-05-10 06:00:00"),
+ pd.Timestamp("2021-05-10 10:00:00"),
+ 5.1,
+ ),
+ (
+ "split after",
+ pd.Timestamp("2021-05-10 15:55:00"),
+ pd.Timestamp("2021-05-10 16:05:00"),
+ 5.1,
+ ),
+ (
+ "before only",
+ pd.Timestamp("2021-05-10 07:55:00"),
+ pd.Timestamp("2021-05-10 07:56:00"),
+ 0.2,
+ ),
+ (
+ "after only",
+ pd.Timestamp("2021-05-10 17:55:00"),
+ pd.Timestamp("2021-05-10 17:56:00"),
+ 0.2,
+ ),
+ (
+ "during only",
+ pd.Timestamp("2021-05-10 13:16:00"),
+ pd.Timestamp("2021-05-10 13:22:00"),
+ 10.0,
+ ),
+ (
+ "between years",
+ pd.Timestamp("2022-12-30 18:00:00"),
+ pd.Timestamp("2023-01-01 10:00"),
+ 0.2,
+ ),
+ )
+ def test_average_zone_occupancy(
+ self, start_time, end_time, expected_occupancy
+ ):
+ occupancy = step_function_occupancy.StepFunctionOccupancy(
+ pd.Timedelta("08:00:00"), pd.Timedelta("16:00:00"), 10.0, 0.2
+ )
+ average_occupancy = occupancy.average_zone_occupancy(
+ "any zone", start_time, end_time
+ )
+ self.assertAlmostEqual(expected_occupancy, average_occupancy, places=4)
+
+ def test_invalid_average_zone_occupancy(self):
+ occupancy = step_function_occupancy.StepFunctionOccupancy(
+ pd.Timedelta("08:00:00"), pd.Timedelta("16:00:00"), 10.0, 0.2
+ )
+ with self.assertRaises(ValueError):
+ _ = occupancy.average_zone_occupancy(
+ "any zone",
+ pd.Timestamp("2021-12-30 14:00:00"),
+ pd.Timestamp("2021-12-30 13:00:00"),
+ )
+
+ def test_invalid_average_zone_occupancy_init(self):
+ with self.assertRaises(ValueError):
+ _ = step_function_occupancy.StepFunctionOccupancy(
+ pd.Timedelta("-08:00:00"), pd.Timedelta("16:00:00"), 10.0, 0.2
+ )
+
+ with self.assertRaises(ValueError):
+ _ = step_function_occupancy.StepFunctionOccupancy(
+ pd.Timedelta("08:00:00"), pd.Timedelta("-16:00:00"), 10.0, 0.2
+ )
+
+
+if __name__ == "__main__":
+ absltest.main()
diff --git a/smart_control/simulator/occupancy/stochastic_occupancy.py b/smart_control/simulator/occupancy/stochastic_occupancy.py
new file mode 100644
index 00000000..a93239b1
--- /dev/null
+++ b/smart_control/simulator/occupancy/stochastic_occupancy.py
@@ -0,0 +1,300 @@
+"""A stochastic occupancy model for building simulation.
+
+This model simulates the behavior of occupants in a building by defining
+arrival, departure, and lunch break times based on random sampling. Each zone
+is assigned a specified number of occupants, and their schedules are generated
+using cumulative probability functions (CPFs) to ensure realistic variability.
+
+For each occupant, arrival and departure times are sampled within defined
+earliest and latest bounds. Lunch break times and durations are also generated
+stochastically. The model determines whether an occupant is present in the
+work zone or away at any given time, accounting for work hours, lunch breaks,
+and holidays.
+
+The `LIGHTSWITCHOccupancy` class calculates the average occupancy for a zone
+over a specified time interval, enabling integration with larger building
+simulation frameworks.
+
+Debugging features are included to provide insights into sampling and state
+transition processes when `debug_print` is enabled.
+"""
+
+# Modify the code to include debug prints
+import datetime
+import enum
+from typing import Optional, Union
+
+import gin
+import numpy as np
+import pandas as pd
+
+from smart_control.models.base_occupancy import BaseOccupancy
+from smart_control.utils import conversion_utils
+
+debug_print = False # Set to False to disable debugging
+
+
+class OccupancyStateEnum(enum.Enum):
+ AWAY = 1
+ WORK = 2
+
+
+class ZoneOccupant:
+ """Zone Occupant."""
+
+ def __init__(
+ self,
+ earliest_expected_arrival_hour: int,
+ latest_expected_arrival_hour: int,
+ earliest_expected_departure_hour: int,
+ latest_expected_departure_hour: int,
+ lunch_start_hour: int,
+ lunch_end_hour: int,
+ step_size: pd.Timedelta,
+ random_state: np.random.RandomState,
+ time_zone: Union[datetime.tzinfo, str] = "UTC",
+ ):
+ # Validate that the time bounds are in chronological order
+ if not (
+ earliest_expected_arrival_hour
+ < latest_expected_arrival_hour
+ < earliest_expected_departure_hour
+ < latest_expected_departure_hour
+ ):
+ raise ValueError(
+ "Arrival and departure hours must be strictly increasing: "
+ "earliest_arrival < latest_arrival < earliest_departure < "
+ "latest_departure. "
+ f"Got: {earliest_expected_arrival_hour}, "
+ f"{latest_expected_arrival_hour}, "
+ f"{earliest_expected_departure_hour}, "
+ f"{latest_expected_departure_hour}."
+ )
+
+ # Validate lunch time bounds
+ if lunch_start_hour >= lunch_end_hour:
+ raise ValueError("lunch_start_hour must be before lunch_end_hour.")
+
+ self._earliest_expected_arrival_hour = earliest_expected_arrival_hour
+ self._latest_expected_arrival_hour = latest_expected_arrival_hour
+ self._earliest_expected_departure_hour = earliest_expected_departure_hour
+ self._latest_expected_departure_hour = latest_expected_departure_hour
+ self._lunch_start_hour = lunch_start_hour
+ self._lunch_end_hour = lunch_end_hour
+ self._step_size = step_size
+ self._random_state = random_state
+ self._time_zone = time_zone
+ self._occupancy_state = OccupancyStateEnum.AWAY
+
+ # Sample times using CPF-based sampling
+ self._arrival_time = self._sample_event_time(
+ self._earliest_expected_arrival_hour, self._latest_expected_arrival_hour
+ )
+
+ self._departure_time = self._sample_event_time(
+ self._earliest_expected_departure_hour,
+ self._latest_expected_departure_hour,
+ )
+ self._lunch_start_time = self._sample_event_time(
+ self._lunch_start_hour, self._lunch_end_hour
+ )
+ self._lunch_duration = self._sample_lunch_duration()
+
+ if debug_print:
+ print(
+ f"ZoneOccupant initialized with: arrival_time={self._arrival_time},"
+ f" departure_time={self._departure_time},"
+ f" lunch_start_time={self._lunch_start_time},"
+ f" lunch_duration={self._lunch_duration}"
+ )
+
+ def _generate_cpf(self, start, end):
+ values = np.arange(start, end + 1)
+ probabilities = self._random_state.rand(len(values))
+ cumulative_probabilities = np.cumsum(probabilities / probabilities.sum())
+ return values, cumulative_probabilities
+
+ def _sample_event_time(self, start, end):
+ values, cumulative_probabilities = self._generate_cpf(start, end)
+ random_value = self._random_state.rand()
+ index = np.searchsorted(cumulative_probabilities, random_value)
+ if debug_print:
+ print(
+ f"Sampled event time: start={start}, end={end}, value={values[index]}"
+ )
+ return values[index]
+
+ def _sample_lunch_duration(self):
+ values, cumulative_probabilities = self._generate_cpf(30, 90)
+ random_value = self._random_state.rand()
+ index = np.searchsorted(cumulative_probabilities, random_value)
+ if debug_print:
+ print(f"Sampled lunch duration: {values[index]} minutes")
+ return values[index]
+
+ def _to_local_time(self, timestamp: pd.Timestamp) -> pd.Timestamp:
+ if timestamp.tz is None:
+ return timestamp
+ return timestamp.tz_convert(self._time_zone)
+
+ def _occupant_arrived(self, timestamp: pd.Timestamp) -> bool:
+ local_timestamp = self._to_local_time(timestamp)
+ arrived = local_timestamp.hour >= self._arrival_time
+ if debug_print:
+ print(
+ f"Check arrival: local_time_hour={local_timestamp.hour},"
+ f" arrival_time={self._arrival_time}, arrived={arrived}"
+ )
+ return arrived
+
+ def _occupant_departed(self, timestamp: pd.Timestamp) -> bool:
+ local_timestamp = self._to_local_time(timestamp)
+ departed = local_timestamp.hour >= self._departure_time
+ if debug_print:
+ print(
+ f"Check departure: local_time_hour={local_timestamp.hour},"
+ f" departure_time={self._departure_time}, departed={departed}"
+ )
+ return departed
+
+ def peek(self, current_time: pd.Timestamp) -> OccupancyStateEnum:
+ """Checks the current occupancy state based on the provided timestamp.
+
+ This method determines the occupancy state (AWAY or WORK) based on
+ the current time, considering workdays, arrival/departure times,
+ and a lunch break.
+
+ Args:
+ current_time: The current timestamp to evaluate.
+
+ Returns:
+ The current `OccupancyStateEnum` (AWAY or WORK).
+ """
+ local_timestamp = self._to_local_time(current_time)
+ local_time = local_timestamp.time() # a datetime.time object
+ if debug_print:
+ print(
+ f"Peek called: current_time={current_time},"
+ f" local_time={local_timestamp}, state={self._occupancy_state}"
+ )
+
+ day = pd.Timestamp(
+ year=local_timestamp.year,
+ month=local_timestamp.month,
+ day=local_timestamp.day,
+ )
+
+ # Check if it's a workday
+ if not conversion_utils.is_work_day(day):
+ self._occupancy_state = OccupancyStateEnum.AWAY
+ return self._occupancy_state
+
+ # Check arrival and departure
+ if self._occupant_arrived(current_time) and not self._occupant_departed(
+ current_time
+ ):
+ self._occupancy_state = OccupancyStateEnum.WORK
+ else:
+ self._occupancy_state = OccupancyStateEnum.AWAY
+
+ # Handle lunch break
+ if self._occupancy_state == OccupancyStateEnum.WORK:
+ lunch_start_time = datetime.time(hour=self._lunch_start_time, minute=0)
+ lunch_end_time = (
+ datetime.datetime.combine(datetime.date.today(), lunch_start_time)
+ + pd.Timedelta(minutes=self._lunch_duration)
+ ).time()
+ if lunch_start_time <= local_time < lunch_end_time:
+ self._occupancy_state = OccupancyStateEnum.AWAY
+ return OccupancyStateEnum.AWAY
+
+ if debug_print:
+ print(f"Occupancy state: {self._occupancy_state}")
+
+ return self._occupancy_state
+
+
+@gin.configurable
+class LIGHTSWITCHOccupancy(BaseOccupancy):
+ """Light Switch Occupancy.
+
+ Calculates the average occupancy for a zone over a specified time interval,
+ enabling integration with larger building simulation frameworks.
+ """
+
+ def __init__(
+ self,
+ zone_assignment: int,
+ earliest_expected_arrival_hour: int,
+ latest_expected_arrival_hour: int,
+ earliest_expected_departure_hour: int,
+ latest_expected_departure_hour: int,
+ lunch_start_hour: int = 12,
+ lunch_end_hour: int = 14,
+ time_step_sec: int = 3600,
+ seed: Optional[int] = 511211,
+ time_zone: str = "UTC",
+ ):
+ self._zone_assignment = zone_assignment
+ self._zone_occupants = {}
+ self._step_size = pd.Timedelta(seconds=time_step_sec)
+ self._earliest_expected_arrival_hour = earliest_expected_arrival_hour
+ self._latest_expected_arrival_hour = latest_expected_arrival_hour
+ self._earliest_expected_departure_hour = earliest_expected_departure_hour
+ self._latest_expected_departure_hour = latest_expected_departure_hour
+ self._lunch_start_hour = lunch_start_hour
+ self._lunch_end_hour = lunch_end_hour
+ self._random_state = np.random.RandomState(seed)
+ self._time_zone = time_zone
+
+ def _initialize_zone(self, zone_id: str):
+ if zone_id not in self._zone_occupants:
+ self._zone_occupants[zone_id] = []
+ for _ in range(self._zone_assignment):
+ self._zone_occupants[zone_id].append(
+ ZoneOccupant(
+ self._earliest_expected_arrival_hour,
+ self._latest_expected_arrival_hour,
+ self._earliest_expected_departure_hour,
+ self._latest_expected_departure_hour,
+ self._lunch_start_hour,
+ self._lunch_end_hour,
+ self._step_size,
+ self._random_state,
+ self._time_zone,
+ )
+ )
+
+ def average_zone_occupancy(
+ self, zone_id: str, start_time: pd.Timestamp, end_time: pd.Timestamp
+ ) -> float:
+ """Calculates the average occupancy within a time interval for a zone.
+
+ Args:
+ zone_id: specific zone identifier for the building.
+ start_time: **local time** with TZ for the beginning of the interval.
+ end_time: **local time** with TZ for the end of the interval.
+
+ Returns:
+ Average number of people in the zone for the interval.
+ """
+ self._initialize_zone(zone_id)
+
+ current_time = start_time
+ total_occupancy = 0
+ steps = 0
+
+ while current_time < end_time:
+ num_occupants = 0
+ for occupant in self._zone_occupants[zone_id]:
+ state = occupant.peek(current_time)
+ if state == OccupancyStateEnum.WORK:
+ num_occupants += 1
+
+ # print(f"Current time: {current_time}, Occupancy count: {num_occupants}")
+ total_occupancy += num_occupants
+ steps += 1
+ current_time += self._step_size
+
+ # Avoid division by zero
+ return total_occupancy / steps if steps > 0 else 0.0
diff --git a/smart_control/simulator/occupancy/stochastic_occupancy_test.py b/smart_control/simulator/occupancy/stochastic_occupancy_test.py
new file mode 100644
index 00000000..d342a2c6
--- /dev/null
+++ b/smart_control/simulator/occupancy/stochastic_occupancy_test.py
@@ -0,0 +1,139 @@
+"""Tests for stochastic_occupancy."""
+
+from absl.testing import absltest
+from absl.testing import parameterized
+import numpy as np
+import pandas as pd
+
+from smart_control.simulator.occupancy.stochastic_occupancy import LIGHTSWITCHOccupancy
+from smart_control.simulator.occupancy.stochastic_occupancy import OccupancyStateEnum
+from smart_control.simulator.occupancy.stochastic_occupancy import ZoneOccupant
+
+# fmt: off
+# pylint: disable=line-too-long
+_EXPECTED_ZONE_OCCUPANCIES_UTC = [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 4.0, 4.0, 4.0, 4.0, 4.0, 4.0, 4.0, 4.0, 4.0, 4.0, 4.0, 4.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 7.0, 7.0, 7.0, 7.0, 7.0, 7.0, 7.0, 7.0, 7.0, 7.0, 7.0, 7.0, 3.0, 4.0, 5.0, 5.0, 6.0, 6.0, 6.0, 6.0, 6.0, 6.0, 7.0, 8.0, 5.0, 5.0, 6.0, 7.0, 7.0, 7.0, 7.0, 7.0, 7.0, 8.0, 8.0, 9.0, 9.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 6.0, 6.0, 6.0, 6.0, 6.0, 6.0, 6.0, 6.0, 6.0, 6.0, 6.0, 6.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0]
+_EXPECTED_ZONE_OCCUPANCIES_PACIFIC = [2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 4.0, 4.0, 4.0, 4.0, 4.0, 4.0, 4.0, 4.0, 4.0, 4.0, 4.0, 4.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 7.0, 7.0, 7.0, 7.0, 7.0, 7.0, 7.0, 7.0, 7.0, 7.0, 7.0, 7.0, 3.0, 4.0, 5.0, 5.0, 6.0, 6.0, 6.0, 6.0, 6.0, 6.0, 7.0, 8.0, 5.0, 5.0, 6.0, 7.0, 7.0, 7.0, 7.0, 7.0, 7.0, 8.0, 8.0, 9.0, 9.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0]
+_EXPECTED_ZONE_OCCUPANCIES_EASTERN = [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 4.0, 4.0, 4.0, 4.0, 4.0, 4.0, 4.0, 4.0, 4.0, 4.0, 4.0, 4.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 7.0, 7.0, 7.0, 7.0, 7.0, 7.0, 7.0, 7.0, 7.0, 7.0, 7.0, 7.0, 3.0, 4.0, 5.0, 5.0, 6.0, 6.0, 6.0, 6.0, 6.0, 6.0, 7.0, 8.0, 5.0, 5.0, 6.0, 7.0, 7.0, 7.0, 7.0, 7.0, 7.0, 8.0, 8.0, 9.0, 9.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0, 6.0, 6.0, 6.0, 6.0, 6.0, 6.0, 6.0, 6.0, 6.0, 6.0, 6.0, 6.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 2.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0]
+# pylint: enable=line-too-long
+# fmt: on
+
+# Constants for test cases
+debug_print = False
+SEED = 511211
+STEP_SIZE = pd.Timedelta(minutes=5)
+EARLIEST_EXPECTED_ARRIVAL_HOUR = 8
+LATEST_EXPECTED_ARRIVAL_HOUR = 10
+EARLIEST_EXPECTED_DEPARTURE_HOUR = 16
+LATEST_EXPECTED_DEPARTURE_HOUR = 18
+NUM_OCCUPANTS = 10
+
+
+class LIGHTSWITCHOccupancyTest(parameterized.TestCase):
+
+ @parameterized.parameters(
+ (None, _EXPECTED_ZONE_OCCUPANCIES_UTC),
+ ('UTC', _EXPECTED_ZONE_OCCUPANCIES_UTC),
+ ('US/Pacific', _EXPECTED_ZONE_OCCUPANCIES_PACIFIC),
+ ('US/Eastern', _EXPECTED_ZONE_OCCUPANCIES_EASTERN),
+ )
+ def test_average_zone_occupancy_weekday(self, tz, expected_zone_occupancies):
+
+ occupancy = LIGHTSWITCHOccupancy(
+ zone_assignment=NUM_OCCUPANTS,
+ earliest_expected_arrival_hour=EARLIEST_EXPECTED_ARRIVAL_HOUR,
+ latest_expected_arrival_hour=LATEST_EXPECTED_ARRIVAL_HOUR,
+ earliest_expected_departure_hour=EARLIEST_EXPECTED_DEPARTURE_HOUR,
+ latest_expected_departure_hour=LATEST_EXPECTED_DEPARTURE_HOUR,
+ lunch_start_hour=12,
+ lunch_end_hour=14,
+ time_step_sec=STEP_SIZE.total_seconds(),
+ seed=SEED,
+ time_zone=tz,
+ )
+ current_time = pd.Timestamp('2021-09-01 00:00', tz='UTC')
+ actual_occupancies = []
+ while current_time < pd.Timestamp('2021-09-01 23:00', tz='UTC'):
+ n = occupancy.average_zone_occupancy(
+ 'zone_0', current_time, current_time + STEP_SIZE
+ )
+ actual_occupancies.append(n)
+
+ current_time += STEP_SIZE
+ self.assertSequenceEqual(expected_zone_occupancies, actual_occupancies)
+
+ @parameterized.parameters(None, 'UTC', 'US/Eastern', 'US/Pacific')
+ def test_peek(self, tz):
+ random_state = np.random.RandomState(seed=SEED)
+ occupant = ZoneOccupant(
+ earliest_expected_arrival_hour=EARLIEST_EXPECTED_ARRIVAL_HOUR,
+ latest_expected_arrival_hour=LATEST_EXPECTED_ARRIVAL_HOUR,
+ earliest_expected_departure_hour=EARLIEST_EXPECTED_DEPARTURE_HOUR,
+ latest_expected_departure_hour=LATEST_EXPECTED_DEPARTURE_HOUR,
+ lunch_start_hour=12,
+ lunch_end_hour=14,
+ step_size=STEP_SIZE,
+ random_state=random_state,
+ time_zone=tz,
+ )
+ current_time = pd.Timestamp('2021-09-01 00:00', tz=tz)
+
+ while current_time < pd.Timestamp('2021-09-01 23:00', tz=tz):
+ state = occupant.peek(current_time=current_time)
+
+ # Debugging information
+ # print(f"Current time: {current_time}, Local time: {local_time}")
+ # print(f"Current state: {state}")
+
+ if current_time < pd.Timestamp(
+ '2021-09-01 10:00', tz=tz
+ ) or current_time >= pd.Timestamp('2021-09-01 17:00', tz=tz):
+ self.assertEqual(OccupancyStateEnum.AWAY, state)
+ elif (
+ pd.Timestamp('2021-09-01 13:00', tz=tz)
+ <= current_time
+ < pd.Timestamp('2021-09-01 13:54', tz=tz)
+ ):
+ self.assertEqual(OccupancyStateEnum.AWAY, state) # Lunch break
+ else:
+ self.assertEqual(OccupancyStateEnum.WORK, state)
+ current_time += STEP_SIZE
+
+ def test_zone_occupant_invalid_hour_order(self):
+ """ValueError when arrival/departure hours are not strictly increasing."""
+ random_state = np.random.RandomState(seed=SEED)
+
+ with self.assertRaisesRegex(
+ ValueError, 'Arrival and departure hours must be strictly increasing'
+ ):
+ ZoneOccupant(
+ earliest_expected_arrival_hour=8,
+ latest_expected_arrival_hour=17, # > earliest_departure (16)
+ earliest_expected_departure_hour=16,
+ latest_expected_departure_hour=18,
+ lunch_start_hour=12,
+ lunch_end_hour=14,
+ step_size=STEP_SIZE,
+ random_state=random_state,
+ )
+
+ def test_zone_occupant_invalid_lunch_hours(self):
+ """ValueError when lunch_start_hour >= lunch_end_hour."""
+ random_state = np.random.RandomState(seed=SEED)
+
+ with self.assertRaisesRegex(
+ ValueError, 'lunch_start_hour must be before lunch_end_hour'
+ ):
+ ZoneOccupant(
+ earliest_expected_arrival_hour=8,
+ latest_expected_arrival_hour=10,
+ earliest_expected_departure_hour=16,
+ latest_expected_departure_hour=18,
+ lunch_start_hour=14, # >= lunch_end_hour
+ lunch_end_hour=12,
+ step_size=STEP_SIZE,
+ random_state=random_state,
+ )
+
+
+if __name__ == '__main__':
+ absltest.main()