Move parse code to its own module

ideenergy/client.py

@@ -22,10 +22,11 @@
 import functools
 import json
 import logging
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, date
 from typing import Any, Dict, List, Optional, Union
 
 import aiohttp
+from . import parsers
 
 _BASE_URL = "https://www.i-de.es/consumidores/rest"
 _CONSUMPTION_PERIOD_ENDPOINT = (
@@ -369,41 +370,33 @@ async def _get_historical_generic_data(
     ) -> Dict[Any, Any]:
         start = min([start, end])
         end = max([start, end])
-
         url = url_template.format(start=start, end=end)
 
         data = await self.request_json("GET", url, encoding="iso-8859-1")
 
-        base = datetime(start.year, start.month, start.day)
-        historical = data["y"]["data"][0]
-        historical = [x for x in historical if x is not None]
-        historical = [
-            (base + timedelta(hours=idx), x.get("valor", None))
-            for (idx, x) in enumerate(historical)
-        ]
-        historical = [(dt, float(x) if x is not None else x) for (dt, x) in historical]
+        base_date = datetime(start.year, start.month, start.day)
+        ret = parsers.parser_generic_historical_data(data, base_date)
 
-        return {
-            "accumulated": float(data["acumulado"]),
-            "accumulated-co2": float(data["acumuladoCO2"]),
-            "historical": historical,
-        }
+        return ret
 
     @auth_required
-    async def get_historical_power_demand(self):
+    async def get_historical_power_demand_limits(self):
         url = _POWER_DEMAND_LIMITS_ENDPOINT
 
         data = await self.request_json("GET", url)
         assert data.get("resultado") == "correcto"
 
-        url = _POWER_DEMAND_PERIOD_ENDPOINT.format(
-            fecMin=data["fecMin"], fecMax=data["fecMax"]
-        )
+        return data
+
+    @auth_required
+    async def get_historical_power_demand(self):
+        limits = await self.get_historical_power_demand_limits()
+        url = _POWER_DEMAND_PERIOD_ENDPOINT.format(**limits)
 
         data = await self.request_json("GET", url)
-        assert data.get("resultado") == "correcto"
 
-        return data["potMaxMens"]
+        ret = parsers.parse_historical_power_demand_data(data)
+        return ret
 
     def __repr__(self):
         return f"<ideenergy.Client username={self.username}, contract={self._contract}>"

ideenergy/parsers.py

@@ -0,0 +1,69 @@
+# -*- coding: utf-8 -*-
+
+# Copyright (C) 2021-2022 Luis López <[email protected]>
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License
+# as published by the Free Software Foundation; either version 2
+# of the License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301,
+# USA.
+
+
+import itertools
+import datetime
+from typing import Dict
+
+
+def parser_generic_historical_data(data, base_dt: datetime.datetime) -> Dict:
+    def _normalize_historical_item(idx: int, item: Dict | None) -> Dict | None:
+        if item is None:
+            return None
+
+        start = base_dt + datetime.timedelta(hours=idx)
+        try:
+            value = float(item["valor"])
+        except (KeyError, ValueError, TypeError):
+            value = None
+
+        return {
+            "start": start,
+            "end": start + datetime.timedelta(hours=1),
+            "value": value,
+        }
+
+    historical = data["y"]["data"][0]
+    historical = [
+        _normalize_historical_item(idx, item) for (idx, item) in enumerate(historical)
+    ]
+    historical = [x for x in historical if x is not None]
+
+    return {
+        "accumulated": float(data["acumulado"]),
+        "accumulated-co2": float(data["acumuladoCO2"]),
+        "historical": historical,
+    }
+
+
+def parse_historical_power_demand_data(data) -> Dict:
+    def _normalize_power_spike_item(item: Dict):
+        return {
+            "dt": datetime.datetime.strptime(item["name"], "%d/%m/%Y %H:%M"),
+            "value": item["y"],
+        }
+
+    assert data.get("resultado") == "correcto"
+
+    potMaxMens = data["potMaxMens"]
+    potMaxMens = list(itertools.chain.from_iterable([x for x in potMaxMens]))
+    potMaxMens = [_normalize_power_spike_item(x) for x in potMaxMens]
+
+    return potMaxMens

tests/test_client.py

@@ -71,6 +71,21 @@ async def test_historical_consumption(self, _):
             self.assertEqual(ret["accumulated"], 97871.0)
             self.assertEqual(ret["accumulated-co2"], 23586.91)
 
+    # @patch("ideenergy.Client.is_logged", return_value=True)
+    # async def test_historical_power_demand(self, _):
+    #     with patch(
+    #         "ideenergy.Client.request_bytes",
+    #         new_class=AsyncMock,
+    #         side_effect=[
+    #             read_fixture("historical-power-demand-limits"),
+    #             read_fixture("historical-power-demand"),
+    #         ],
+    #     ):
+    #         ret = await self.client.get_historical_power_demand()
+
+    #         self.assertEqual(ret["accumulated"], 97871.0)
+    #         self.assertEqual(ret["accumulated-co2"], 23586.91)
+
 
 if __name__ == "__main__":
     unittest.main()