diff --git a/citylearn/citylearn.py b/citylearn/citylearn.py index d4bc5a50..a04a2824 100644 --- a/citylearn/citylearn.py +++ b/citylearn/citylearn.py @@ -157,14 +157,14 @@ def __init__(self, if requested_render_mode not in {'none', 'during', 'end'}: raise ValueError("render_mode must be one of {'none', 'during', 'end'}.") self.render_mode = requested_render_mode - self._buffer_render = self.render_mode == 'end' - self._defer_render_flush = False - self._render_buffer = defaultdict(list) - self._render_start_date = self._parse_render_start_date(start_date if start_date is not None else schema_start_date) - self.previous_month = None - self.current_day = self._render_start_date.day - self.year = self._render_start_date.year - self._final_kpis_exported = False + self._buffer_render = self.render_mode == 'end' + self._defer_render_flush = False + self._render_buffer = defaultdict(list) + self._render_start_date = self._parse_render_start_date(start_date if start_date is not None else schema_start_date) + self.previous_month = None + self.current_day = self._render_start_date.day + self.year = self._render_start_date.year + self._final_kpis_exported = False self.__rewards = None self.buildings = [] self.random_seed = self.schema.get('random_seed', None) if random_seed is None else random_seed @@ -267,8 +267,8 @@ def __init__(self, self.new_folder_path = None self._render_start_datetime = None - if self.render_enabled: - self._ensure_render_output_dir(ensure_exists=False) + if self.render_enabled: + self._ensure_render_output_dir(ensure_exists=False) @property def render_start_date(self) -> datetime.date: @@ -653,33 +653,6 @@ def net_electricity_consumption_without_storage(self) -> np.ndarray: for b in self.buildings ]).sum(axis = 0, min_count = 1).to_numpy() - @property - def net_electricity_consumption_emission_without_storage(self) -> np.ndarray: - """Summed `Building.net_electricity_consumption_emission_without_storage` time series, in [kg_co2].""" - - return pd.DataFrame([ - b.net_electricity_consumption_emission_without_storage - for b in self.buildings - ]).sum(axis = 0, min_count = 1).tolist() - - @property - def net_electricity_consumption_cost_without_storage(self) -> np.ndarray: - """Summed `Building.net_electricity_consumption_cost_without_storage` time series, in [$].""" - - return pd.DataFrame([ - b.net_electricity_consumption_cost_without_storage - for b in self.buildings - ]).sum(axis = 0, min_count = 1).to_numpy() - - @property - def net_electricity_consumption_without_storage(self) -> np.ndarray: - """Summed `Building.net_electricity_consumption_without_storage` time series, in [kWh].""" - - return pd.DataFrame([ - b.net_electricity_consumption_without_storage - for b in self.buildings - ]).sum(axis = 0, min_count = 1).to_numpy() - @property def net_electricity_consumption_emission(self) -> List[float]: """Summed `Building.net_electricity_consumption_emission` time series, in [kg_co2].""" @@ -941,7 +914,7 @@ def time_step_ratio(self, time_step_ratio: int): Environment.time_step_ratio.fset(self, time_step_ratio) for b in self.buildings: - b.time_step_ratio = self.time_step_ratio + b.time_step_ratio = self.time_step_ratio def get_metadata(self) -> Mapping[str, Any]: return { @@ -1026,41 +999,41 @@ def step(self, actions: List[List[float]]) -> Tuple[List[List[float]], List[floa # Advance to next timestep t+1 self.next_time_step() - # store episode reward summary at the end of episode (upon reaching final timestep) - if self.terminated: - if self.render_mode == 'during' and self.render_enabled: - # Final step was already streamed during the most recent `next_time_step` call. - pass - rewards = np.array(self.__rewards[1:], dtype='float32') - self.__episode_rewards.append({ - 'min': rewards.min(axis=0).tolist(), - 'max': rewards.max(axis=0).tolist(), - 'sum': rewards.sum(axis=0).tolist(), - 'mean': rewards.mean(axis=0).tolist() - }) - if self.render_mode == 'end' and self.render_enabled: - if self.time_step > 0: - final_index = min(self.time_steps - 1, self.time_step - 1) - else: - final_index = 0 - - has_buffered_rows = any(self._render_buffer.values()) - - if not has_buffered_rows: - state_snapshot = self._override_render_time_step(final_index) - self._defer_render_flush = True - try: - self.render() - finally: - self._restore_render_time_step(state_snapshot) - self._defer_render_flush = False - - self._flush_render_buffer() - - if self.render_enabled and not self._final_kpis_exported: - self.export_final_kpis() - - return self.observations, reward, self.terminated, self.truncated, self.get_info() + # store episode reward summary at the end of episode (upon reaching final timestep) + if self.terminated: + if self.render_mode == 'during' and self.render_enabled: + # Final step was already streamed during the most recent `next_time_step` call. + pass + rewards = np.array(self.__rewards[1:], dtype='float32') + self.__episode_rewards.append({ + 'min': rewards.min(axis=0).tolist(), + 'max': rewards.max(axis=0).tolist(), + 'sum': rewards.sum(axis=0).tolist(), + 'mean': rewards.mean(axis=0).tolist() + }) + if self.render_mode == 'end' and self.render_enabled: + if self.time_step > 0: + final_index = min(self.time_steps - 1, self.time_step - 1) + else: + final_index = 0 + + has_buffered_rows = any(self._render_buffer.values()) + + if not has_buffered_rows: + state_snapshot = self._override_render_time_step(final_index) + self._defer_render_flush = True + try: + self.render() + finally: + self._restore_render_time_step(state_snapshot) + self._defer_render_flush = False + + self._flush_render_buffer() + + if self.render_enabled and not self._final_kpis_exported: + self.export_final_kpis() + + return self.observations, reward, self.terminated, self.truncated, self.get_info() def get_info(self) -> Mapping[Any, Any]: """Other information to return from the `citylearn.CityLearnEnv.step` function.""" @@ -1122,7 +1095,7 @@ def _parse_actions(self, actions: List[List[float]]) -> List[Mapping[str, float] action_dict['electric_vehicle_storage_actions'] = electric_vehicle_actions # aqui podes criar dicionario if washing_machine_actions: - action_dict['washing_machine_actions'] = washing_machine_actions + action_dict['washing_machine_actions'] = washing_machine_actions # Fill missing actions with default NaN for k in building.action_metadata: @@ -1357,78 +1330,78 @@ def next_time_step(self): #It basicly associates an EV to a Building.Charger self.associate_chargers_to_electric_vehicles() - def associate_chargers_to_electric_vehicles(self): - r"""Associate charger to its corresponding electric_vehicle based on charger simulation state.""" - - def _resolve_arrival_soc(simulation: ChargerSimulation, step: int, prev_state: float, prev_id: Union[str, None], ev_identifier: str) -> Union[float, None]: - """Return expected SOC (as fraction) for an EV connecting at `step`, or ``None`` when unavailable.""" - - candidate_index = None - - if prev_state in (2, 3) and step > 0: - if isinstance(prev_id, str) and prev_id.strip() not in {"", "nan"} and prev_id != ev_identifier: - raise ValueError( - f"Charger dataset EV mismatch: expected '{ev_identifier}' but found '{prev_id}' at time step {step - 1}." - ) - candidate_index = step - 1 - - elif 0 <= step < len(simulation.electric_vehicle_estimated_soc_arrival): - candidate_index = step - - soc_value = None - - if candidate_index is not None and 0 <= candidate_index < len(simulation.electric_vehicle_estimated_soc_arrival): - candidate = simulation.electric_vehicle_estimated_soc_arrival[candidate_index] - if isinstance(candidate, (float, np.floating)) and not np.isnan(candidate) and candidate >= 0: - soc_value = float(candidate) - - if soc_value is None and 0 <= step < len(simulation.electric_vehicle_required_soc_departure): - fallback = simulation.electric_vehicle_required_soc_departure[step] - if isinstance(fallback, (float, np.floating)) and not np.isnan(fallback) and fallback >= 0: - soc_value = float(fallback) - - return soc_value - - for building in self.buildings: - if building.electric_vehicle_chargers is None: - continue - - for charger in building.electric_vehicle_chargers: - sim = charger.charger_simulation - state = sim.electric_vehicle_charger_state[self.time_step] - - if np.isnan(state) or state not in [1, 2]: - continue # Skip if no EV is connected or incoming - - ev_id = sim.electric_vehicle_id[self.time_step] - prev_state = np.nan - prev_ev_id = None - if self.time_step > 0: - idx = self.time_step - 1 - if idx < len(sim.electric_vehicle_charger_state): - prev_state = sim.electric_vehicle_charger_state[idx] - if idx < len(sim.electric_vehicle_id): - prev_ev_id = sim.electric_vehicle_id[idx] - - if isinstance(ev_id, str) and ev_id.strip() not in ["", "nan"]: - for ev in self.electric_vehicles: - if ev.name == ev_id: - if state == 1: - charger.plug_car(ev) - is_new_connection = ( - prev_state != 1 - or not isinstance(prev_ev_id, str) - or prev_ev_id != ev_id - ) - if is_new_connection: - soc_value = _resolve_arrival_soc(sim, self.time_step, prev_state, prev_ev_id, ev_id) - if soc_value is not None: - ev.battery.force_set_soc(soc_value) - elif state == 2: - charger.associate_incoming_car(ev) - - def simulate_unconnected_ev_soc(self): - """Simulate SOC changes for EVs that are not under charger control at t+1.""" + def associate_chargers_to_electric_vehicles(self): + r"""Associate charger to its corresponding electric_vehicle based on charger simulation state.""" + + def _resolve_arrival_soc(simulation: ChargerSimulation, step: int, prev_state: float, prev_id: Union[str, None], ev_identifier: str) -> Union[float, None]: + """Return expected SOC (as fraction) for an EV connecting at `step`, or ``None`` when unavailable.""" + + candidate_index = None + + if prev_state in (2, 3) and step > 0: + if isinstance(prev_id, str) and prev_id.strip() not in {"", "nan"} and prev_id != ev_identifier: + raise ValueError( + f"Charger dataset EV mismatch: expected '{ev_identifier}' but found '{prev_id}' at time step {step - 1}." + ) + candidate_index = step - 1 + + elif 0 <= step < len(simulation.electric_vehicle_estimated_soc_arrival): + candidate_index = step + + soc_value = None + + if candidate_index is not None and 0 <= candidate_index < len(simulation.electric_vehicle_estimated_soc_arrival): + candidate = simulation.electric_vehicle_estimated_soc_arrival[candidate_index] + if isinstance(candidate, (float, np.floating)) and not np.isnan(candidate) and candidate >= 0: + soc_value = float(candidate) + + if soc_value is None and 0 <= step < len(simulation.electric_vehicle_required_soc_departure): + fallback = simulation.electric_vehicle_required_soc_departure[step] + if isinstance(fallback, (float, np.floating)) and not np.isnan(fallback) and fallback >= 0: + soc_value = float(fallback) + + return soc_value + + for building in self.buildings: + if building.electric_vehicle_chargers is None: + continue + + for charger in building.electric_vehicle_chargers: + sim = charger.charger_simulation + state = sim.electric_vehicle_charger_state[self.time_step] + + if np.isnan(state) or state not in [1, 2]: + continue # Skip if no EV is connected or incoming + + ev_id = sim.electric_vehicle_id[self.time_step] + prev_state = np.nan + prev_ev_id = None + if self.time_step > 0: + idx = self.time_step - 1 + if idx < len(sim.electric_vehicle_charger_state): + prev_state = sim.electric_vehicle_charger_state[idx] + if idx < len(sim.electric_vehicle_id): + prev_ev_id = sim.electric_vehicle_id[idx] + + if isinstance(ev_id, str) and ev_id.strip() not in ["", "nan"]: + for ev in self.electric_vehicles: + if ev.name == ev_id: + if state == 1: + charger.plug_car(ev) + is_new_connection = ( + prev_state != 1 + or not isinstance(prev_ev_id, str) + or prev_ev_id != ev_id + ) + if is_new_connection: + soc_value = _resolve_arrival_soc(sim, self.time_step, prev_state, prev_ev_id, ev_id) + if soc_value is not None: + ev.battery.force_set_soc(soc_value) + elif state == 2: + charger.associate_incoming_car(ev) + + def simulate_unconnected_ev_soc(self): + """Simulate SOC changes for EVs that are not under charger control at t+1.""" t = self.time_step if t + 1 >= self.episode_tracker.episode_time_steps: return @@ -1488,30 +1461,30 @@ def simulate_unconnected_ev_soc(self): new_soc = np.clip(last_soc * variability, 0.0, 1.0) ev.battery.force_set_soc(new_soc) - def export_final_kpis(self, model: 'citylearn.agents.base.Agent' = None, filepath: str = "exported_kpis.csv"): - """Export episode KPIs to csv. - - Parameters - ---------- - model: citylearn.agents.base.Agent, optional - Agent whose environment should be evaluated. Defaults to the current environment. - filepath: str, default: ``"exported_kpis.csv"`` - Output filename placed inside :pyattr:`new_folder_path`. - """ - # Ensure output directory exists even if rendering was disabled - self._ensure_render_output_dir() - file_path = os.path.join(self.new_folder_path, filepath) - if model is not None and getattr(model, 'env', None) is not None: - kpis = model.env.evaluate() - else: - kpis = self.evaluate() - kpis = kpis.pivot(index='cost_function', columns='name', values='value').round(3) - kpis = kpis.dropna(how='all') - kpis = kpis.fillna('') - kpis = kpis.reset_index() - kpis = kpis.rename(columns={'cost_function': 'KPI'}) - kpis.to_csv(file_path, index=False, encoding='utf-8') - self._final_kpis_exported = True + def export_final_kpis(self, model: 'citylearn.agents.base.Agent' = None, filepath: str = "exported_kpis.csv"): + """Export episode KPIs to csv. + + Parameters + ---------- + model: citylearn.agents.base.Agent, optional + Agent whose environment should be evaluated. Defaults to the current environment. + filepath: str, default: ``"exported_kpis.csv"`` + Output filename placed inside :pyattr:`new_folder_path`. + """ + # Ensure output directory exists even if rendering was disabled + self._ensure_render_output_dir() + file_path = os.path.join(self.new_folder_path, filepath) + if model is not None and getattr(model, 'env', None) is not None: + kpis = model.env.evaluate() + else: + kpis = self.evaluate() + kpis = kpis.pivot(index='cost_function', columns='name', values='value').round(3) + kpis = kpis.dropna(how='all') + kpis = kpis.fillna('') + kpis = kpis.reset_index() + kpis = kpis.rename(columns={'cost_function': 'KPI'}) + kpis.to_csv(file_path, index=False, encoding='utf-8') + self._final_kpis_exported = True def render(self): """ @@ -1563,107 +1536,107 @@ def render(self): self._save_to_csv(ev_filename, {"timestamp": iso_timestamp, **ev.as_dict()}) - def _save_to_csv(self, filename, data): - """ - Saves data to a CSV file, appending it if the file exists. When `render_mode='end'`, - rows may be buffered in memory until a flush is requested. - """ - if self._buffer_render and getattr(self, '_defer_render_flush', False): - self._render_buffer[filename].append(dict(data)) - return - - self._write_render_rows(filename, [dict(data)]) - - def _flush_render_buffer(self): - """Write any buffered render rows to disk.""" - if not getattr(self, '_render_buffer', None): - return - - has_pending_rows = any(self._render_buffer.values()) - if not has_pending_rows: - self._render_buffer.clear() - return - - try: - target_dir = Path(self.new_folder_path) - except Exception: - target_dir = None - - if target_dir is not None: - print(f"Writing buffered render exports to {target_dir} ...") - - original_defer = self._defer_render_flush - original_buffer_state = self._buffer_render - self._defer_render_flush = False - self._buffer_render = False - - try: - for filename, rows in list(self._render_buffer.items()): - if rows: - self._write_render_rows(filename, rows) - finally: - self._render_buffer.clear() - self._buffer_render = original_buffer_state - self._defer_render_flush = original_defer - - def _write_render_rows(self, filename: str, rows: List[Mapping[str, Any]]): - """Write one or more render rows to disk with minimal rewrites.""" - - file_path = Path(self.new_folder_path) / filename - file_path.parent.mkdir(parents=True, exist_ok=True) - if not rows: - return - - buffered_fieldnames = list( - dict.fromkeys(field for row in rows for field in row.keys()) - ) - - if not file_path.exists(): - fieldnames = buffered_fieldnames - with file_path.open('w', newline='') as csvfile: - writer = csv.DictWriter(csvfile, fieldnames=fieldnames) - writer.writeheader() - for row in rows: - writer.writerow({field: row.get(field, '') for field in fieldnames}) - return - - # File exists – inspect current header. - needs_header_extension = False - with file_path.open('r', newline='') as existing: - reader = csv.DictReader(existing) - existing_fieldnames = reader.fieldnames or [] - for field in buffered_fieldnames: - if field not in existing_fieldnames: - needs_header_extension = True - break - if needs_header_extension: - existing_rows = list(reader) - else: - existing_rows = None - - if not needs_header_extension: - with file_path.open('a', newline='') as csvfile: - writer = csv.DictWriter(csvfile, fieldnames=existing_fieldnames) - for row in rows: - writer.writerow({field: row.get(field, '') for field in existing_fieldnames}) - return - - # Need to rewrite with the expanded header. - extended_fieldnames = list( - dict.fromkeys(existing_fieldnames + [f for f in buffered_fieldnames if f not in existing_fieldnames]) - ) - - existing_rows = existing_rows or [] - for row in existing_rows: - for field in extended_fieldnames: - row.setdefault(field, '') - - with file_path.open('w', newline='') as csvfile: - writer = csv.DictWriter(csvfile, fieldnames=extended_fieldnames) - writer.writeheader() - writer.writerows(existing_rows) - for row in rows: - writer.writerow({field: row.get(field, '') for field in extended_fieldnames}) + def _save_to_csv(self, filename, data): + """ + Saves data to a CSV file, appending it if the file exists. When `render_mode='end'`, + rows may be buffered in memory until a flush is requested. + """ + if self._buffer_render and getattr(self, '_defer_render_flush', False): + self._render_buffer[filename].append(dict(data)) + return + + self._write_render_rows(filename, [dict(data)]) + + def _flush_render_buffer(self): + """Write any buffered render rows to disk.""" + if not getattr(self, '_render_buffer', None): + return + + has_pending_rows = any(self._render_buffer.values()) + if not has_pending_rows: + self._render_buffer.clear() + return + + try: + target_dir = Path(self.new_folder_path) + except Exception: + target_dir = None + + if target_dir is not None: + print(f"Writing buffered render exports to {target_dir} ...") + + original_defer = self._defer_render_flush + original_buffer_state = self._buffer_render + self._defer_render_flush = False + self._buffer_render = False + + try: + for filename, rows in list(self._render_buffer.items()): + if rows: + self._write_render_rows(filename, rows) + finally: + self._render_buffer.clear() + self._buffer_render = original_buffer_state + self._defer_render_flush = original_defer + + def _write_render_rows(self, filename: str, rows: List[Mapping[str, Any]]): + """Write one or more render rows to disk with minimal rewrites.""" + + file_path = Path(self.new_folder_path) / filename + file_path.parent.mkdir(parents=True, exist_ok=True) + if not rows: + return + + buffered_fieldnames = list( + dict.fromkeys(field for row in rows for field in row.keys()) + ) + + if not file_path.exists(): + fieldnames = buffered_fieldnames + with file_path.open('w', newline='') as csvfile: + writer = csv.DictWriter(csvfile, fieldnames=fieldnames) + writer.writeheader() + for row in rows: + writer.writerow({field: row.get(field, '') for field in fieldnames}) + return + + # File exists – inspect current header. + needs_header_extension = False + with file_path.open('r', newline='') as existing: + reader = csv.DictReader(existing) + existing_fieldnames = reader.fieldnames or [] + for field in buffered_fieldnames: + if field not in existing_fieldnames: + needs_header_extension = True + break + if needs_header_extension: + existing_rows = list(reader) + else: + existing_rows = None + + if not needs_header_extension: + with file_path.open('a', newline='') as csvfile: + writer = csv.DictWriter(csvfile, fieldnames=existing_fieldnames) + for row in rows: + writer.writerow({field: row.get(field, '') for field in existing_fieldnames}) + return + + # Need to rewrite with the expanded header. + extended_fieldnames = list( + dict.fromkeys(existing_fieldnames + [f for f in buffered_fieldnames if f not in existing_fieldnames]) + ) + + existing_rows = existing_rows or [] + for row in existing_rows: + for field in extended_fieldnames: + row.setdefault(field, '') + + with file_path.open('w', newline='') as csvfile: + writer = csv.DictWriter(csvfile, fieldnames=extended_fieldnames) + writer.writeheader() + writer.writerows(existing_rows) + for row in rows: + writer.writerow({field: row.get(field, '') for field in extended_fieldnames}) def _parse_render_start_date(self, start_date: Union[str, datetime.date]) -> datetime.date: """Return a valid start date for rendering timestamps.""" @@ -1689,61 +1662,61 @@ def _parse_render_start_date(self, start_date: Union[str, datetime.date]) -> dat "CityLearnEnv start_date must be a date, datetime, or ISO format string." ) - def _ensure_render_output_dir(self, *, ensure_exists: bool = True): - """Prepare the render output directory and optionally create it on disk. - - Parameters - ---------- - ensure_exists: bool, default: True - When ``True`` the directory tree is created (and legacy exports removed when - reusing :pyattr:`render_session_name`). When ``False`` only internal state is - updated so that paths can be materialized later on demand. - """ - base_render_path = Path(getattr(self, 'render_output_root', Path(__file__).resolve().parents[1] / 'render_logs')).expanduser() - - if ensure_exists: - try: - base_render_path.mkdir(parents=True, exist_ok=True) - except PermissionError: - fallback = (Path.cwd() / 'render_logs').resolve() - fallback.mkdir(parents=True, exist_ok=True) - self.render_output_root = fallback - base_render_path = fallback - - render_dir = getattr(self, '_render_directory_path', None) - needs_new_dir = render_dir is None - - if not needs_new_dir and ensure_exists: - render_dir = Path(render_dir) - try: - needs_new_dir = not render_dir.is_relative_to(base_render_path) - except AttributeError: - needs_new_dir = base_render_path not in render_dir.parents and render_dir != base_render_path - - if needs_new_dir: - if self.render_session_name: - render_dir = (base_render_path / Path(self.render_session_name)).expanduser().resolve() - else: - if getattr(self, '_render_timestamp', None) is None: - self._render_timestamp = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S") - render_dir = (base_render_path / self._render_timestamp).resolve() - - self._render_directory_path = render_dir - else: - render_dir = Path(self._render_directory_path) - - if ensure_exists: - render_dir.mkdir(parents=True, exist_ok=True) - if not self._render_dir_initialized: - if self.render_session_name: - for csv_file in render_dir.glob('exported_*.csv'): - try: - csv_file.unlink() - except OSError: - pass - self._render_dir_initialized = True - - self.new_folder_path = str(render_dir) + def _ensure_render_output_dir(self, *, ensure_exists: bool = True): + """Prepare the render output directory and optionally create it on disk. + + Parameters + ---------- + ensure_exists: bool, default: True + When ``True`` the directory tree is created (and legacy exports removed when + reusing :pyattr:`render_session_name`). When ``False`` only internal state is + updated so that paths can be materialized later on demand. + """ + base_render_path = Path(getattr(self, 'render_output_root', Path(__file__).resolve().parents[1] / 'render_logs')).expanduser() + + if ensure_exists: + try: + base_render_path.mkdir(parents=True, exist_ok=True) + except PermissionError: + fallback = (Path.cwd() / 'render_logs').resolve() + fallback.mkdir(parents=True, exist_ok=True) + self.render_output_root = fallback + base_render_path = fallback + + render_dir = getattr(self, '_render_directory_path', None) + needs_new_dir = render_dir is None + + if not needs_new_dir and ensure_exists: + render_dir = Path(render_dir) + try: + needs_new_dir = not render_dir.is_relative_to(base_render_path) + except AttributeError: + needs_new_dir = base_render_path not in render_dir.parents and render_dir != base_render_path + + if needs_new_dir: + if self.render_session_name: + render_dir = (base_render_path / Path(self.render_session_name)).expanduser().resolve() + else: + if getattr(self, '_render_timestamp', None) is None: + self._render_timestamp = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S") + render_dir = (base_render_path / self._render_timestamp).resolve() + + self._render_directory_path = render_dir + else: + render_dir = Path(self._render_directory_path) + + if ensure_exists: + render_dir.mkdir(parents=True, exist_ok=True) + if not self._render_dir_initialized: + if self.render_session_name: + for csv_file in render_dir.glob('exported_*.csv'): + try: + csv_file.unlink() + except OSError: + pass + self._render_dir_initialized = True + + self.new_folder_path = str(render_dir) def _get_iso_timestamp(self): # Reset time tracking if this is the first step of a new episode @@ -1773,81 +1746,81 @@ def _get_series_value(series, index, default): next_hour = _get_series_value(hour_series, next_index, hour) next_minutes = _get_series_value(minutes_series, next_index, minutes) - raw_hour = hour - timestamp_year = self.year - timestamp_month = month - timestamp_day = self.current_day - hour_for_timestamp = raw_hour % 24 - next_hour_mod = next_hour % 24 - next_minutes_clamped = max(0, min(59, next_minutes)) - minute_for_timestamp = max(0, min(59, minutes)) - - if raw_hour >= 24: - if next_month != month: - timestamp_month = next_month - - if next_month < month: - timestamp_year = self.year + 1 - timestamp_day = 1 - else: - # Keep the current day; the day roll-over is handled via next_day logic. - timestamp_day = self.current_day - - timestamp = f"{timestamp_year:04d}-{int(timestamp_month):02d}-{timestamp_day:02d}T{hour_for_timestamp:02d}:{minute_for_timestamp:02d}:00" - - next_year = timestamp_year - next_day = timestamp_day - - if next_month != month: - if next_month < month: - next_year = timestamp_year + 1 - next_day = 1 - elif next_hour_mod <= hour_for_timestamp and next_minutes_clamped <= minute_for_timestamp: - next_day = timestamp_day + 1 - - self.year = next_year - self.current_day = next_day - - return timestamp - - def _override_render_time_step(self, index: int): - """Temporarily set time_step to `index` for the environment and descendants.""" - - snapshot = [] - - def _record(obj): - if hasattr(obj, 'time_step'): - snapshot.append((obj, obj.time_step)) - obj.time_step = index - - _record(self) - for building in getattr(self, 'buildings', []): - _record(building) - electrical_storage = getattr(building, 'electrical_storage', None) - if electrical_storage is not None: - _record(electrical_storage) - - for charger in getattr(building, 'electric_vehicle_chargers', []) or []: - _record(charger) - - for washing_machine in getattr(building, 'washing_machines', []) or []: - _record(washing_machine) - - for ev in getattr(self, 'electric_vehicles', []): - _record(ev) - battery = getattr(ev, 'battery', None) - if battery is not None: - _record(battery) - - return snapshot - - @staticmethod - def _restore_render_time_step(snapshot): - for obj, value in snapshot: - try: - obj.time_step = value - except AttributeError: - pass + raw_hour = hour + timestamp_year = self.year + timestamp_month = month + timestamp_day = self.current_day + hour_for_timestamp = raw_hour % 24 + next_hour_mod = next_hour % 24 + next_minutes_clamped = max(0, min(59, next_minutes)) + minute_for_timestamp = max(0, min(59, minutes)) + + if raw_hour >= 24: + if next_month != month: + timestamp_month = next_month + + if next_month < month: + timestamp_year = self.year + 1 + timestamp_day = 1 + else: + # Keep the current day; the day roll-over is handled via next_day logic. + timestamp_day = self.current_day + + timestamp = f"{timestamp_year:04d}-{int(timestamp_month):02d}-{timestamp_day:02d}T{hour_for_timestamp:02d}:{minute_for_timestamp:02d}:00" + + next_year = timestamp_year + next_day = timestamp_day + + if next_month != month: + if next_month < month: + next_year = timestamp_year + 1 + next_day = 1 + elif next_hour_mod <= hour_for_timestamp and next_minutes_clamped <= minute_for_timestamp: + next_day = timestamp_day + 1 + + self.year = next_year + self.current_day = next_day + + return timestamp + + def _override_render_time_step(self, index: int): + """Temporarily set time_step to `index` for the environment and descendants.""" + + snapshot = [] + + def _record(obj): + if hasattr(obj, 'time_step'): + snapshot.append((obj, obj.time_step)) + obj.time_step = index + + _record(self) + for building in getattr(self, 'buildings', []): + _record(building) + electrical_storage = getattr(building, 'electrical_storage', None) + if electrical_storage is not None: + _record(electrical_storage) + + for charger in getattr(building, 'electric_vehicle_chargers', []) or []: + _record(charger) + + for washing_machine in getattr(building, 'washing_machines', []) or []: + _record(washing_machine) + + for ev in getattr(self, 'electric_vehicles', []): + _record(ev) + battery = getattr(ev, 'battery', None) + if battery is not None: + _record(battery) + + return snapshot + + @staticmethod + def _restore_render_time_step(snapshot): + for obj, value in snapshot: + try: + obj.time_step = value + except AttributeError: + pass def _reset_time_tracking(self): """Reset all time tracking variables.""" @@ -1881,9 +1854,9 @@ def reset(self, seed: int = None, options: Mapping[str, Any] = None) -> Tuple[Li Override :meth"`get_info` to get custom key-value pairs in `info`. """ - # object reset - super().reset() - self._final_kpis_exported = False + # object reset + super().reset() + self._final_kpis_exported = False # update seed if seed is not None: @@ -1985,24 +1958,24 @@ def load_agent(self, agent: Union[str, 'citylearn.agents.base.Agent'] = None, ** else: agent_type = self.schema['agent']['type'] - if kwargs is not None and len(kwargs) > 0: - agent_attributes = dict(kwargs) - - elif agent is None: - agent_attributes = dict(self.schema['agent'].get('attributes', {})) - - else: - agent_attributes = {} - - if 'env' not in agent_attributes: - agent_attributes['env'] = self - - agent_module = '.'.join(agent_type.split('.')[0:-1]) - agent_name = agent_type.split('.')[-1] - agent_constructor = getattr(importlib.import_module(agent_module), agent_name) - agent = agent_constructor(**agent_attributes) - - return agent + if kwargs is not None and len(kwargs) > 0: + agent_attributes = dict(kwargs) + + elif agent is None: + agent_attributes = dict(self.schema['agent'].get('attributes', {})) + + else: + agent_attributes = {} + + if 'env' not in agent_attributes: + agent_attributes['env'] = self + + agent_module = '.'.join(agent_type.split('.')[0:-1]) + agent_name = agent_type.split('.')[-1] + agent_constructor = getattr(importlib.import_module(agent_module), agent_name) + agent = agent_constructor(**agent_attributes) + + return agent def _load(self, schema: Mapping[str, Any], **kwargs) -> Tuple[Union[Path, str], List[Building], List[ElectricVehicle], Union[int, List[Tuple[int, int]]], bool, bool, float, RewardFunction, bool, List[str], EpisodeTracker]: """Return `CityLearnEnv` and `Controller` objects as defined by the `schema`. @@ -2056,7 +2029,7 @@ def _load(self, schema: Mapping[str, Any], **kwargs) -> Tuple[Union[Path, str], key: value for key, value in schema["observations"].items() if key not in set(schema['chargers_observations_helper']) | set(schema['washing_machine_observations_helper']) - } + } schema['actions'] = { key: value for key, value in schema['actions'].items() @@ -2447,7 +2420,7 @@ def process_metadata(self, schema, building_schema, chargers_list, washing_machi observation_metadata = {k: v['active'] for k, v in schema['observations'].items()} # Since minutes is Optional, in case the schema has minutes as observation metadata and some energy simulation building csv doesn't contain minutes, remove it from observation if 'minutes' in observation_metadata and energy_simulation.minutes is None: - observation_metadata.pop('minutes', None) + observation_metadata.pop('minutes', None) chargers_observations_metadata_helper = {k: v['active'] for k, v in schema['chargers_observations_helper'].items()} washing_machine_observations_metadata_helper = {k: v['active'] for k, v in schema['washing_machine_observations_helper'].items()} @@ -2584,7 +2557,7 @@ def process_metadata(self, schema, building_schema, chargers_list, washing_machi if washing_machine_actions_metadata_helper.get("washing_machine", False): - action_metadata[f'{washing_machine_name}'] = True + action_metadata[f'{washing_machine_name}'] = True return observation_metadata, action_metadata @@ -2626,7 +2599,7 @@ def _load_electric_vehicle(self, electric_vehicle_name: str, schema: dict, elect ) return ev - + def _load_washing_machine( self, washing_machine_name: str, @@ -2673,7 +2646,7 @@ def _load_washing_machine( ) return wm - + def __str__(self) -> str: """ Return a string representation of the current simulation state. @@ -2694,19 +2667,19 @@ def as_dict(self) -> dict: dict Dictionary with energy and environmental metrics for the current step. """ - if len(self.net_electricity_consumption) == 0: - idx = 0 - else: - idx = max(0, min(self.time_step, len(self.net_electricity_consumption) - 1)) - - return { - "Net Electricity Consumption-kWh": self.net_electricity_consumption[idx], - "Self Consumption-kWh": self.total_self_consumption[idx], - "Stored energy by community- kWh": self.energy_to_electrical_storage[idx], - "Total Solar Generation-kWh": self.solar_generation[idx], - "CO2-kg_co2": self.net_electricity_consumption_emission[idx], - "Price-$": self.net_electricity_consumption_cost[idx], - } + if len(self.net_electricity_consumption) == 0: + idx = 0 + else: + idx = max(0, min(self.time_step, len(self.net_electricity_consumption) - 1)) + + return { + "Net Electricity Consumption-kWh": self.net_electricity_consumption[idx], + "Self Consumption-kWh": self.total_self_consumption[idx], + "Stored energy by community- kWh": self.energy_to_electrical_storage[idx], + "Total Solar Generation-kWh": self.solar_generation[idx], + "CO2-kg_co2": self.net_electricity_consumption_emission[idx], + "Price-$": self.net_electricity_consumption_cost[idx], + } class Error(Exception): """Base class for other exceptions."""