From 5142122548ffae072381771af6a08e93adb391e2 Mon Sep 17 00:00:00 2001 From: geoffhancock Date: Tue, 1 Sep 2026 14:53:03 -0400 Subject: [PATCH 1/3] Fix include_imputed_marker being passed as include_meta in get_historical_csv get_historical_csv passed include_imputed_marker as the sixth positional argument to get_historical_pandas, whose sixth parameter is include_meta. As a result, get_historical_csv(..., include_imputed_marker=True) wrote a CSV containing a meta column and no imputed_data_used column. Pass all defaulted parameters by keyword in the internal call chain (get_historical_csv -> get_historical_pandas -> get_historical_jsons) so argument order differences between signatures cannot misroute a flag. Public signatures are unchanged. Adds a regression test asserting the CSV contains imputed_data_used and not meta when include_imputed_marker=True. Co-Authored-By: Claude Fable 5 --- tests/test_sdk.py | 18 ++++++++++++++++++ watttime/api.py | 14 ++++++++++++-- 2 files changed, 30 insertions(+), 2 deletions(-) diff --git a/tests/test_sdk.py b/tests/test_sdk.py index 557cee3a..a79ab392 100644 --- a/tests/test_sdk.py +++ b/tests/test_sdk.py @@ -289,6 +289,24 @@ def test_get_historical_csv(self): assert fp.exists() fp.unlink() + def test_get_historical_csv_include_imputed(self): + start = parse("2025-01-01 00:00Z") + end = parse("2025-01-02 00:00Z") + self.historical.get_historical_csv( + start, end, REGION, include_imputed_marker=True + ) + + fp = ( + Path.home() + / "watttime_historical_csvs" + / f"{REGION}_co2_moer_{start.date()}_{end.date()}.csv" + ) + assert fp.exists() + df = pd.read_csv(fp) + self.assertIn("imputed_data_used", df.columns) + self.assertNotIn("meta", df.columns) + fp.unlink() + def test_multi_model_range(self): """If model is not specified, we should only return the most recent model data""" myaccess = WattTimeMyAccess() diff --git a/watttime/api.py b/watttime/api.py index 81d372c4..7f04fae5 100644 --- a/watttime/api.py +++ b/watttime/api.py @@ -467,7 +467,12 @@ def get_historical_pandas( pd.DataFrame: _description_ """ responses = self.get_historical_jsons( - start, end, region, signal_type, model, include_imputed_marker + start, + end, + region, + signal_type=signal_type, + model=model, + include_imputed_marker=include_imputed_marker, ) df = pd.json_normalize( responses, record_path="data", meta=["meta"] if include_meta else [] @@ -503,7 +508,12 @@ def get_historical_csv( None, results are saved to a csv file in the user's home directory. """ df = self.get_historical_pandas( - start, end, region, signal_type, model, include_imputed_marker + start, + end, + region, + signal_type=signal_type, + model=model, + include_imputed_marker=include_imputed_marker, ) out_dir = Path.home() / "watttime_historical_csvs" From 9d334f8a7ed718b57f66151dbcb02a43c801d88d Mon Sep 17 00:00:00 2001 From: geoffhancock Date: Fri, 18 Sep 2026 13:15:44 -0400 Subject: [PATCH 2/3] Expose request timeout and historical chunk size; hint at both on timeout Add a keyword-only `timeout` to WattTimeBase.__init__ (default (10, 60), matching the existing login/register literals) and pass it at all three HTTP call sites. The data request previously used a scalar 60, so its default connect timeout drops from 60 s to 10 s; default read timeouts are unchanged. Add a keyword-only `chunk_size` to get_historical_jsons/_pandas/_csv, forwarded to _get_chunks, which now accepts None (30 days) and rejects sizes of 5 minutes or less -- the per-chunk trim -- instead of looping forever or producing inverted chunks. When a request fails on a timeout, append a hint to the RuntimeError naming both knobs. Once the session's retries are exhausted, requests raises ConnectionError -> MaxRetryError -> ReadTimeoutError rather than ReadTimeout, so _is_timeout walks the exception chain. Motivation: in August 2026, 30-day historical pulls for some regions took over 180 s server-side while the same pulls at 10 days took ~11 s. Users hit an opaque read timeout after ~4 minutes with no supported lever; the only workaround was monkeypatching _get_chunks. Co-Authored-By: Claude Fable 5.1 --- tests/test_sdk.py | 176 ++++++++++++++++++++++++++++++++++++++++++++++ watttime/api.py | 84 +++++++++++++++++++--- 2 files changed, 249 insertions(+), 11 deletions(-) diff --git a/tests/test_sdk.py b/tests/test_sdk.py index a79ab392..9db18a42 100644 --- a/tests/test_sdk.py +++ b/tests/test_sdk.py @@ -1,3 +1,4 @@ +import inspect import unittest import unittest.mock as mock from unittest.mock import patch @@ -15,6 +16,8 @@ ) from pathlib import Path import pytest +import requests +import urllib3.exceptions from shapely.geometry import shape, Polygon, MultiPolygon import pandas as pd @@ -74,6 +77,119 @@ def test_login_sets_user_agent_header(self): self.base.headers["User-Agent"], f"watttime-python-sdk-{VERSION}" ) + def _mock_login_response(self): + mock_rsp = mock.Mock() + mock_rsp.json.return_value = {"token": "fake-token"} + mock_rsp.raise_for_status.return_value = None + return mock_rsp + + def _authed_base(self, **kwargs): + """A client with a fake, unexpired token so requests skip _login().""" + base = WattTimeBase(**kwargs) + base.token = "fake-token" + base.token_valid_until = datetime.now() + timedelta(minutes=30) + base.headers = {"Authorization": "Bearer fake-token"} + return base + + def test_timeout_reaches_all_three_call_sites(self): + """The default (10, 60) is passed to login, the data request, and register.""" + base = WattTimeBase() + self.assertEqual(base.timeout, (10, 60)) + + mock_rsp = mock.Mock() + mock_rsp.json.return_value = {"token": "fake-token", "data": [], "meta": {}} + mock_rsp.raise_for_status.return_value = None + + with mock.patch.object(base.session, "get", return_value=mock_rsp) as mock_get: + base._login() + self.assertEqual(mock_get.call_args.kwargs["timeout"], (10, 60)) + + base._make_rate_limited_request("https://api.watttime.org/v3/test", {}) + self.assertEqual(mock_get.call_args.kwargs["timeout"], (10, 60)) + + with mock.patch.object( + base.session, "post", return_value=mock_rsp + ) as mock_post: + base.register(email="someone@example.com") + self.assertEqual(mock_post.call_args.kwargs["timeout"], (10, 60)) + + base.session.close() + + def test_timeout_accepts_tuple_and_scalar_and_none(self): + """Both `requests` timeout forms, and None, are stored and forwarded as given.""" + for value in [(3, 5), 30, 12.5, None]: + with self.subTest(timeout=value): + base = WattTimeBase(timeout=value) + self.assertEqual(base.timeout, value) + + with mock.patch.object( + base.session, "get", return_value=self._mock_login_response() + ) as mock_get: + base._login() + + self.assertEqual(mock_get.call_args.kwargs["timeout"], value) + base.session.close() + + def test_new_parameters_are_keyword_only(self): + """New optional parameters are keyword-only, so existing positional calls + are unaffected and the positional contract stays where it is today.""" + keyword_only = inspect.Parameter.KEYWORD_ONLY + checks = [ + (WattTimeBase.__init__, "timeout"), + (WattTimeHistorical.get_historical_jsons, "chunk_size"), + (WattTimeHistorical.get_historical_pandas, "chunk_size"), + (WattTimeHistorical.get_historical_csv, "chunk_size"), + ] + for func, name in checks: + with self.subTest(f"{func.__qualname__}({name})"): + param = inspect.signature(func).parameters[name] + self.assertIs(param.kind, keyword_only) + + client = WattTimeHistorical(None, None, False, 10, 4) + self.assertEqual(client.worker_count, 4) + self.assertEqual(client.timeout, (10, 60)) + client.session.close() + + with self.assertRaises(TypeError): + WattTimeHistorical(None, None, False, 10, 4, (3, 5)) + + def test_timeout_error_message_points_at_knobs(self): + """A timeout is wrapped in RuntimeError with __cause__ preserved and a hint + naming both knobs -- including when urllib3's retry machinery has wrapped it + as ConnectionError -> MaxRetryError -> ReadTimeoutError, which is what the + session raises once its retries are exhausted.""" + url = "https://api.watttime.org/v3/test" + reason = urllib3.exceptions.ReadTimeoutError(None, url, "Read timed out.") + exhausted = requests.exceptions.ConnectionError( + urllib3.exceptions.MaxRetryError(None, url, reason=reason) + ) + + for exc in [requests.exceptions.ReadTimeout("too slow"), exhausted]: + with self.subTest(exc=type(exc).__name__): + base = self._authed_base(timeout=(10, 60)) + with mock.patch.object(base.session, "get", side_effect=exc): + with self.assertRaises(RuntimeError) as ctx: + base._make_rate_limited_request(url, {}) + + self.assertIs(ctx.exception.__cause__, exc) + message = str(ctx.exception) + self.assertIn("Hint", message) + self.assertIn("chunk_size", message) + self.assertIn("`timeout`", message) + self.assertIn("(10, 60)", message) + base.session.close() + + def test_non_timeout_error_message_has_no_hint(self): + base = self._authed_base() + error = requests.exceptions.HTTPError("403 Client Error: Forbidden") + + with mock.patch.object(base.session, "get", side_effect=error): + with self.assertRaises(RuntimeError) as ctx: + base._make_rate_limited_request("https://api.watttime.org/v3/test", {}) + + self.assertNotIn("Hint", str(ctx.exception)) + base.session.close() + @patch("time.sleep", return_value=None) def test_apply_rate_limit(self, mock_sleep): """Test _apply_rate_limit (single-threaded) triggers sleep when rate limit is exceeded.""" @@ -179,6 +295,34 @@ def test_get_chunks_rejects_inverted_span(self): with self.assertRaises(ValueError): self.base._get_chunks(start, end) + def test_get_chunks_honors_chunk_size(self): + start = datetime(2025, 1, 1, tzinfo=UTC) + end = datetime(2025, 1, 31, tzinfo=UTC) + + chunks = self.base._get_chunks(start, end, chunk_size=timedelta(days=10)) + + self.assertEqual(len(chunks), 3) + self.assertEqual(chunks[0][0], start) + self.assertEqual(chunks[-1][1], end) + self.assertEqual(chunks[1][0] - chunks[0][1], timedelta(minutes=5)) + + # None means the 30-day default + self.assertEqual( + self.base._get_chunks(start, end, chunk_size=None), + self.base._get_chunks(start, end), + ) + + def test_get_chunks_rejects_chunk_size_at_or_below_trim(self): + # 5 minutes is trimmed from every chunk end but the last, so a chunk that + # short would be empty or inverted; zero or negative would never terminate. + start = datetime(2025, 1, 1, tzinfo=UTC) + end = datetime(2025, 1, 2, tzinfo=UTC) + + for bad in [timedelta(minutes=5), timedelta(0), timedelta(minutes=-10)]: + with self.subTest(chunk_size=bad): + with self.assertRaises(ValueError): + self.base._get_chunks(start, end, chunk_size=bad) + @mock.patch("watttime.requests.Session.post", side_effect=mocked_register) def test_mock_register(self, mock_post): resp = self.base.register(email=os.getenv("WATTTIME_EMAIL")) @@ -202,6 +346,38 @@ def setUp(self): def tearDown(self): self.historical.session.close() + def test_get_historical_jsons_chunk_size_controls_request_count(self): + """30 days at 10-day chunks is three requests instead of one.""" + fake_response = {"data": [], "meta": {"model": {"date": "2024-01-01"}}} + + with mock.patch.object( + self.historical, "_fetch_data", return_value=[fake_response] + ) as mock_fetch: + self.historical.get_historical_jsons( + "2025-01-01 00:00Z", + "2025-01-31 00:00Z", + REGION, + chunk_size=timedelta(days=10), + ) + + param_chunks = mock_fetch.call_args.args[1] + self.assertEqual(len(param_chunks), 3) + + def test_get_historical_pandas_forwards_chunk_size(self): + fake_response = { + "data": [{"point_time": "2025-01-01T00:00:00+00:00", "value": 1.0}], + "meta": {}, + } + + with mock.patch.object( + self.historical, "get_historical_jsons", return_value=[fake_response] + ) as mock_jsons: + self.historical.get_historical_pandas( + "2025-01-01", "2025-01-02", REGION, chunk_size=timedelta(days=1) + ) + + self.assertEqual(mock_jsons.call_args.kwargs["chunk_size"], timedelta(days=1)) + def test_get_historical_jsons_3_months(self): start = "2025-01-01 00:00Z" end = "2025-03-31 00:00Z" diff --git a/watttime/api.py b/watttime/api.py index 7f04fae5..8884140a 100644 --- a/watttime/api.py +++ b/watttime/api.py @@ -18,6 +18,7 @@ from dateutil.parser import parse from pytz import UTC from requests.adapters import HTTPAdapter +import urllib3.exceptions from urllib3.util.retry import Retry try: @@ -27,6 +28,30 @@ VERSION = "0.0.0" +def _is_timeout(exc: BaseException) -> bool: + """ + True if `exc` is, or wraps, a connect/read timeout. + + Once the session's retries are exhausted, requests surfaces a read timeout as + ConnectionError -> MaxRetryError -> ReadTimeoutError rather than as ReadTimeout, + so the whole chain has to be walked, not just the top-level exception. + """ + seen = set() + stack = [exc] + while stack: + e = stack.pop() + if e is None or id(e) in seen: + continue + seen.add(id(e)) + if isinstance( + e, (requests.exceptions.Timeout, urllib3.exceptions.TimeoutError) + ): + return True + stack.extend([e.__cause__, e.__context__, getattr(e, "reason", None)]) + stack.extend(a for a in e.args if isinstance(a, BaseException)) + return False + + class WattTimeAPIWarning: def __init__(self, url: str, params: Dict[str, Any], warning_message: str): self.url = url @@ -71,6 +96,8 @@ def __init__( multithreaded: bool = False, rate_limit: int = 10, worker_count: int = min(10, (os.cpu_count() or 1) * 2), + *, + timeout: Optional[Union[float, Tuple[float, float]]] = (10, 60), ): """ Initializes a new instance of the class. @@ -81,6 +108,7 @@ def __init__( multithreaded (bool): Whether to use multithreading for requests. Default is False. rate_limit (int): The maximum number of requests to make per second. Default is 10 as this algins well with WattTime's API rate limiting policy. worker_count (int): The number of worker threads to use for multithreading. Default is min(10, (os.cpu_count() or 1) * 2). + timeout (Optional[Union[float, Tuple[float, float]]]): The timeout passed to every HTTP request, in seconds, using the standard `requests` forms: a single value applied to both the connect and read phases, or a (connect, read) tuple. `None` disables timeouts entirely. Default is (10, 60). Note that this is a per-attempt timeout: the session retries each request up to 3 times with backoff, so a request can take several times this long before raising. """ @@ -100,6 +128,7 @@ def __init__( self.multithreaded = multithreaded self.rate_limit = rate_limit + self.timeout = timeout self._last_request_times = [] self.worker_count = worker_count self.raised_warnings: List[WattTimeAPIWarning] = [] @@ -136,7 +165,7 @@ def _login(self): auth=requests.auth.HTTPBasicAuth( os.getenv("WATTTIME_USER"), os.getenv("WATTTIME_PASSWORD") ), - timeout=(10, 60), + timeout=self.timeout, ) rsp.raise_for_status() self.token = rsp.json().get("token", None) @@ -184,7 +213,10 @@ def _parse_dates( return start, end def _get_chunks( - self, start: datetime, end: datetime, chunk_size: timedelta = timedelta(days=30) + self, + start: datetime, + end: datetime, + chunk_size: Optional[timedelta] = None, ) -> List[Tuple[datetime, datetime]]: """ Generate a list of tuples representing chunks of time within a given time range. @@ -192,15 +224,23 @@ def _get_chunks( Args: start (datetime): The start datetime of the time range. end (datetime): The end datetime of the time range. - chunk_size (timedelta, optional): The size of each chunk. Defaults to timedelta(days=30). + chunk_size (Optional[timedelta], optional): The size of each chunk. None means the default of 30 days. + Must be longer than 5 minutes, since 5 minutes is trimmed from the end of every chunk but the last. Returns: List[Tuple[datetime, datetime]]: A list of tuples representing the chunks of time. If start == end, a single zero-length chunk is returned. Raises: - ValueError: If start is after end. + ValueError: If start is after end, or chunk_size is not longer than 5 minutes. """ + if chunk_size is None: + chunk_size = timedelta(days=30) + if chunk_size <= timedelta(minutes=5): + raise ValueError( + f"chunk_size must be longer than 5 minutes, got {chunk_size}" + ) + if start > end: raise ValueError(f"start ({start}) must not be after end ({end})") @@ -240,7 +280,7 @@ def register(self, email: str, organization: Optional[str] = None) -> None: "org": organization, } - rsp = self.session.post(url, json=params, timeout=(10, 60)) + rsp = self.session.post(url, json=params, timeout=self.timeout) rsp.raise_for_status() LOG.info( f"Successfully registered {os.getenv('WATTTIME_USER')}, please check {email} for a verification email" @@ -298,13 +338,23 @@ def _make_rate_limited_request(self, url: str, params: Dict[str, Any]) -> Dict: self._apply_rate_limit(ts) try: - rsp = self.session.get(url, headers=self.headers, params=params, timeout=60) + rsp = self.session.get( + url, headers=self.headers, params=params, timeout=self.timeout + ) rsp.raise_for_status() j = rsp.json() except requests.exceptions.RequestException as e: - raise RuntimeError( - f"API Request Failed: {e}\nURL: {url}\nParams: {params}" - ) from e + msg = f"API Request Failed: {e}\nURL: {url}\nParams: {params}" + if _is_timeout(e): + msg += ( + f"\nHint: the request exceeded the client timeout ({self.timeout}) " + "on every retry. This usually means the API was slow to respond, " + "which is more likely for requests covering a large time span. " + "Either pass a smaller `chunk_size` to the historical methods " + "(e.g. timedelta(days=10)) so each request covers less time, or " + "construct the client with a longer `timeout`." + ) + raise RuntimeError(msg) from e meta = j.get("meta", {}) warnings = meta.get("warnings") @@ -398,6 +448,8 @@ def get_historical_jsons( ] = "co2_moer", model: Optional[Union[str, date]] = None, include_imputed_marker: bool = False, + *, + chunk_size: Optional[timedelta] = None, ) -> List[dict]: """ Base function to scrape historical data, returning a list of .json responses. @@ -409,6 +461,9 @@ def get_historical_jsons( signal_type (str, optional): one of ['co2_moer', 'co2_aoer', 'health_damage']. Defaults to "co2_moer". model (Optional[Union[str, date]], optional): Optionally provide a model, used for versioning models. Defaults to None. + chunk_size (Optional[timedelta], optional): The span of each request the date range is split into. + Defaults to 30 days. Smaller spans mean more, faster requests; use this when the API is + slow to answer large spans and requests are hitting the read timeout. Raises: Exception: Scraping failed for some reason @@ -423,7 +478,7 @@ def get_historical_jsons( params["include_imputed_marker"] = "true" start, end = self._parse_dates(start, end) - chunks = self._get_chunks(start, end) + chunks = self._get_chunks(start, end, chunk_size=chunk_size) # No model will default to the most recent model version available if model is not None: @@ -453,12 +508,14 @@ def get_historical_pandas( model: Optional[Union[str, date]] = None, include_meta: bool = False, include_imputed_marker: bool = False, + *, + chunk_size: Optional[timedelta] = None, ): """ Return a pd.DataFrame with point_time, and values. Args: - See .get_hist_jsons() for shared arguments. + See .get_historical_jsons() for shared arguments. include_meta (bool, optional): adds additional columns to the output dataframe, containing the metadata information. Note that metadata is returned for each API response, not for each point_time. @@ -473,6 +530,7 @@ def get_historical_pandas( signal_type=signal_type, model=model, include_imputed_marker=include_imputed_marker, + chunk_size=chunk_size, ) df = pd.json_normalize( responses, record_path="data", meta=["meta"] if include_meta else [] @@ -492,6 +550,8 @@ def get_historical_csv( ] = "co2_moer", model: Optional[Union[str, date]] = None, include_imputed_marker: bool = False, + *, + chunk_size: Optional[timedelta] = None, ): """ Retrieves historical data from a specified start date to an end date and saves it as a CSV file. @@ -503,6 +563,7 @@ def get_historical_csv( region (str): The region for which historical data is requested. signal_type (Optional[Literal["co2_moer", "co2_aoer", "health_damage"]]): The type of signal for which historical data is requested. Default is "co2_moer". model (Optional[Union[str, date]]): The date of the model for which historical data is requested. It can be a string in the format "YYYY-MM-DD" or a date object. Default is None. + chunk_size (Optional[timedelta]): See .get_historical_jsons(). Default is None (30 days). Returns: None, results are saved to a csv file in the user's home directory. @@ -514,6 +575,7 @@ def get_historical_csv( signal_type=signal_type, model=model, include_imputed_marker=include_imputed_marker, + chunk_size=chunk_size, ) out_dir = Path.home() / "watttime_historical_csvs" From ed4f2c498eb24b3e040f683cba4d8a510b5d3ff8 Mon Sep 17 00:00:00 2001 From: geoffhancock Date: Fri, 18 Sep 2026 13:19:01 -0400 Subject: [PATCH 3/3] Document timeout and chunk_size in the README Add a short "Tuning requests that time out" note under the historical data example showing both options and that the timeout is per attempt. Co-Authored-By: Claude Fable 5.1 --- README.md | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/README.md b/README.md index 6a680742..11faf371 100644 --- a/README.md +++ b/README.md @@ -74,6 +74,29 @@ wt_hist.get_historical_csv( ) ``` +#### Tuning requests that time out + +Historical pulls are split into 30-day requests, each with a 60 s read timeout. If the API is slow to answer a large span, a request can time out and the error message will say so. Two options, which can be combined: + +```python +from datetime import timedelta +from watttime import WattTimeHistorical + +# allow each request more time: (connect, read) seconds +wt_hist = WattTimeHistorical(username, password, timeout=(10, 300)) + +# or make each request cover less time +moers = wt_hist.get_historical_pandas( + start = '2022-01-01 00:00Z', + end = '2023-01-01 00:00Z', + region = 'CAISO_NORTH', + signal_type = 'co2_moer', + chunk_size = timedelta(days=10) +) +``` + +The timeout applies to each attempt; the client retries a failed request up to three times before raising. + You could also combine these classes to iterate through all regions where you have access to data: ```python