add script to update google sheet

.github/workflows/update_record.yml

@@ -0,0 +1,21 @@
+name: Update PetKit Record
+on:
+  schedule:
+      # * is a special character in YAML so you have to quote this string
+      - cron:  '*/15 * * * *'
+
+jobs:
+  Update-Records-Action:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: Setup Python
+        uses: actions/setup-python@v2
+        with:
+          python-version: '3.8'
+      - name: Install Tox and any other packages
+        run: pip install tox
+      - name: Build venv
+        run: tox -e venv
+      - name: Run script
+        run:  venv/bin/python -m scripts.update

.vscode/settings.json

@@ -0,0 +1,4 @@
+{
+    "python.formatting.provider": "autopep8",
+    "python.pythonPath": "venv/bin/python"
+}

petkit_exporter/googlesheet_exporter.py

@@ -0,0 +1,115 @@
+from collections import defaultdict
+import string
+
+from google.oauth2 import service_account
+from googleapiclient.discovery import build
+
+from petkit_exporter.petkit import CleanEvent
+from petkit_exporter.petkit import PetEvent
+
+
+class GoogleSheetExporter:
+
+    def __init__(self, spreadsheet_id=None, auth_json=None):
+        self.creds = service_account.Credentials.from_service_account_info(auth_json)
+        self.sheets_service = build("sheets", "v4", credentials=self.creds).spreadsheets()
+        self.spreadsheet_id = spreadsheet_id
+
+    def create_sheet_and_share(self, file_name, share_email, pet_names):
+        spread_sheet = self.sheets_service.create(
+            body={"properties": {"title": file_name}},
+            fields="spreadsheetId,spreadsheetUrl"
+        ).execute()
+        drive_service = build("drive", "v3", credentials=self.creds)
+        drive_service.permissions().create(
+            fileId=spread_sheet["spreadsheetId"],
+            body={
+                "type": "user",
+                "role": "writer",
+                "emailAddress": share_email
+            }
+        ).execute()
+        self.spreadsheet_id = spread_sheet["spreadsheetId"]
+        self.spreadsheet_url = spread_sheet["spreadsheetUrl"]
+        # create sheets
+        sheets_names = pet_names + ["unknown", "other"]
+        self.sheets_service.batchUpdate(
+            spreadsheetId=spread_sheet["spreadsheetId"],
+            body={
+                "requests": [
+                    {
+                        "addSheet": {"properties": {"title": name}}
+                    }
+                    for name in sheets_names
+                ]
+            }
+        ).execute()
+        for sheet in pet_names + ["unknown"]:
+            range = f"{sheet}!A1:" + string.ascii_uppercase[len(PetEvent._fields) -1] + "1"
+            self.sheets_service.values().update(
+                spreadsheetId=self.spreadsheet_id,
+                range=range,
+                valueInputOption="RAW",
+                body={"values": [PetEvent._fields]}
+            ).execute()
+        self.sheets_service.values().update(
+            spreadsheetId=self.spreadsheet_id,
+            range="other!A1",
+            valueInputOption="RAW",
+            body={"values": [["last_modified"]]}
+        ).execute()
+        self.sheets_service.values().update(
+            spreadsheetId=self.spreadsheet_id,
+            range="other!A2:" + string.ascii_uppercase[len(CleanEvent._fields) - 1] + "2",
+            valueInputOption="RAW",
+            body={"values": [CleanEvent._fields]}
+        ).execute()
+        return spread_sheet
+
+    def get_latest_updated_timestamp(self):
+        val = self.sheets_service.values().get(
+            spreadsheetId=self.spreadsheet_id,
+            range="other!B1"
+        ).execute().get("values")
+        if val is None:
+            return None
+        return int(val[0][0])
+
+    def set_latest_updated_timestamp(self, value):
+        self.sheets_service.values().update(
+            spreadsheetId=self.spreadsheet_id,
+            range="other!B1",
+            valueInputOption="RAW",
+            body={"values": [[value]]}
+        ).execute()
+
+    def update(self, records):
+        last_timestamp = self.get_latest_updated_timestamp()
+        sheets = defaultdict(list)
+        new_ts = None
+        for ts, event in records:
+            if new_ts is None or ts > new_ts:
+                new_ts = ts
+            if last_timestamp is not None and ts <= last_timestamp:
+                # we've already had this logged
+                continue
+            if isinstance(event, PetEvent):
+                pet_name = event.name or "unknown"
+                sheets[pet_name].append(list(event))
+            if isinstance(event, CleanEvent):
+                sheets["other"].append(list(event))
+
+        for sheet_name, rows in sheets.items():
+            range = 'A:' + string.ascii_uppercase[
+                len(rows[0]) - 1]
+            self.sheets_service.values().append(
+                spreadsheetId=self.spreadsheet_id,
+                valueInputOption="RAW",
+                range=f"{sheet_name}!{range}",
+                body={
+                    "majorDimension": "ROWS",
+                    "values": rows
+                }
+            ).execute()
+        self.set_latest_updated_timestamp(new_ts)
+        print(new_ts)

petkit_exporter/petkit.py

@@ -0,0 +1,176 @@
+import datetime
+import hashlib
+from collections import namedtuple
+from typing import Dict, List, Optional
+
+import requests
+
+
+PETKIT_API = "http://api.petkt.com"
+
+EVENT_TYPES = {
+    5: "cleaning",
+    10: "pet in the litter box",
+    8: "deorder"
+}
+
+START_REASON = {
+    0: "auto",
+    1: "periodic",
+    2: "manual"
+}
+
+PetEvent = namedtuple(
+    "PetEvent", [
+        "time_start",
+        "time_end",
+        "duration",
+        "name",
+        "weight",
+    ],
+    defaults=(None,) * 5
+)
+
+CleanEvent = namedtuple(
+    "CleanEvent", [
+        "time_start",
+        "time_end",
+        "duration",
+        "event_name",
+        "trigger_reason",
+        "litter_percent",
+        "need_clean",
+        "deoder_percent",
+        "refill_deoder",
+    ],
+    defaults=(None,) * 9
+)
+
+
+class PetkitURL:
+    LOGIN = "/latest/user/login"
+    USER_DETAILS = "/latest/user/details2"
+    DISCOVERY = "/latest/discovery/device_roster"
+    PURAX_DETAILS = "/latest/t3/device_detail"
+    PURAX_RECORDS = "/latest/t3/getDeviceRecord"
+
+
+class PetKit:
+
+    def __init__(self, user_name: str, password: str) -> None:
+        self.user_name = user_name
+        self.password = hashlib.md5(password.encode("utf-8")).hexdigest()
+        self.access_token: Optional[str] = None
+        self.access_token_expiration: Optional[datetime.datetime] = None
+        self.user: Optional[Dict] = None
+
+    def maybe_login(self) -> None:
+        if (
+            self.access_token is not None
+            and self.access_token_expiration > datetime.datetime.utcnow()
+        ):
+            return
+        r = requests.post(
+            f"{PETKIT_API}{PetkitURL.LOGIN}",
+            data={
+                "username": self.user_name,
+                "password": self.password,
+                "encrypt": 1
+            }
+        )
+        r.raise_for_status()
+        session = r.json()["result"]["session"]
+        self.access_token = session["id"]
+        self.access_token_expiration = datetime.datetime.strptime(
+            session["createdAt"], "%Y-%m-%dT%H:%M:%S.%fZ"
+        ) + datetime.timedelta(seconds=session["expiresIn"])
+
+    def _query(self, path: str) -> Dict:
+        self.maybe_login()
+        r = requests.post(
+            f"{PETKIT_API}{path}", headers={"X-Session": self.access_token}
+        )
+        r.raise_for_status()
+        response = r.json()
+        if response.get("error") is not None:
+            raise ValueError(response["error"]["msg"])
+        return response
+
+    def _format_time(self, ts: int) -> str:
+        return datetime.datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S")
+
+    def get_user_details(self) -> None:
+        r = self._query(PetkitURL.USER_DETAILS)
+        self.user = r["result"]["user"]
+
+    def discover_devices(self) -> List[Dict]:
+        r = self._query(PetkitURL.DISCOVERY)
+        return r["result"]["devices"]
+
+    def get_device_details(self, device_id: int) -> Dict:
+        r = self._query(f"{PetkitURL.PURAX_DETAILS}?id={device_id}")
+        return r["result"]
+
+    def get_device_records(self, device_id: int) -> List[Dict]:
+        r = self._query(f"{PetkitURL.PURAX_RECORDS}?deviceId={device_id}")
+        return [self.parse_record(row) for row in r["result"]]
+
+    def parse_record(self, record):
+        if record["eventType"] == 10:
+            # Pet in Litter box
+            pet = self.find_most_possible_pet(record["content"]["petWeight"])
+            return (
+                record["timestamp"],
+                PetEvent(
+                    self._format_time(record["content"]["timeIn"]),
+                    self._format_time(record["content"]["timeOut"]),
+                    record["content"]["timeOut"] - record["content"]["timeIn"],
+                    (pet or {}).get("name"),
+                    record["content"]["petWeight"]
+                )
+            )
+
+        if record["eventType"] == 5:
+            # cleaning
+            return(
+                record["timestamp"],
+                CleanEvent(
+                    self._format_time(record["content"]["startTime"]),
+                    self._format_time(record["timestamp"]),
+                    record["timestamp"] - record["content"]["startTime"],
+                    "clean",
+                    START_REASON[record["content"]["startReason"]],
+                    litter_percent=record["content"]["litterPercent"],
+                    need_clean=record["content"]["boxFull"]
+                )
+            )
+        if record["eventType"] == 8:
+            # deorder
+            return (
+                record["timestamp"],
+                CleanEvent(
+                    self._format_time(record["content"]["startTime"]),
+                    self._format_time(record["timestamp"]),
+                    record["timestamp"] - record["content"]["startTime"],
+                    "deorder",
+                    START_REASON[record["content"]["startReason"]],
+                    deoder_percent=record["content"]["liquid"],
+                    refill_deoder=record["content"]["liquidLack"]
+                )
+            )
+
+    def find_most_possible_pet(self, weight):
+        if self.user is None:
+            self.get_user_details()
+        pet = sorted(
+            self.user["dogs"],
+            key=lambda p: abs(p["weight"] * 1000 - weight)
+        )[0]
+        if pet["weight"] > 600:
+            return None
+        return pet
+
+    def get_pet_names(self):
+        if self.user is None:
+            self.get_user_details()
+        return [p["name"] for p in self.user["dogs"]]

script/update.py

@@ -0,0 +1,20 @@
+import os
+import json
+
+from petkit_exporter.googlesheet_exporter import GoogleSheetExporter
+from petkit_exporter.petkit import PetKit
+
+
+google_auth_json = json.loads(os.environ["AUTH_JSON"])
+petkit_username = os.environ["PETKIT_USERNAME"]
+petkit_password = os.environ["PETKIT_PASSWORD"]
+spreadsheet_id = os.environ["SPREADSHEET_ID"]
+
+petkit = PetKit(petkit_username, petkit_password)
+google_sheet = GoogleSheetExporter(
+    auth_json=google_auth_json,
+    spreadsheet_id=spreadsheet_id
+)
+
+device_id = petkit.discover_devices()[0]["data"]["id"]
+google_sheet.update(petkit.get_device_records(device_id))

tox.ini

@@ -0,0 +1,15 @@
+[tox]
+envlist = py38
+skipsdist = True
+
+[testenv]
+deps =
+  requests
+  google-api-python-client
+  google-auth-httplib2
+  google-auth-oauthlib
+
+[testenv:venv]
+basepython = python
+envdir = venv
+commands =