From 5cf970df62fa9885436e1e7e26e98b2d3b76594d Mon Sep 17 00:00:00 2001 From: Luke Calladine Date: Mon, 12 Jan 2026 20:27:00 +0000 Subject: [PATCH 1/3] Add simple HanchuESS Helper --- .env.example | 24 ++++++++ backend/requirements.txt | 5 +- backend/src/main.py | 11 ++++ backend/src/service/hanchu_service.py | 80 +++++++++++++++++++++++++++ docker-compose.yml | 13 ++++- 5 files changed, 131 insertions(+), 2 deletions(-) create mode 100644 .env.example create mode 100644 backend/src/service/hanchu_service.py diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..0f59a5b --- /dev/null +++ b/.env.example @@ -0,0 +1,24 @@ +# HanchuESS API Configuration +# Copy this file to .env and fill in your actual values + +# Required: AES encryption key (must be 16, 24, or 32 bytes) +HANCHU_AES_KEY=your_aes_key_here + +# Required: AES initialization vector (must be 16 bytes) +HANCHU_AES_IV=your_aes_iv_here + +# Required: Login URL for the HanchuESS API +HANCHU_LOGIN_URL=https://api.example.com/login + +# Optional: Login type (default: ACCOUNT) +HANCHU_LOGIN_TYPE=ACCOUNT + +# Optional: HTTP timeout in seconds (default: 10) +HANCHU_HTTP_TIMEOUT=10 + +# Optional: Verify SSL certificates (default: true, set to false for self-signed certs) +HANCHU_VERIFY_SSL=true + +# Optional: Username and password +HANCHU_USERNAME= +HANCHU_PASSWORD= diff --git a/backend/requirements.txt b/backend/requirements.txt index 8e0578a..e88472a 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -1,2 +1,5 @@ fastapi -uvicorn[standard] \ No newline at end of file +uvicorn[standard] +requests>=2.31.0 +pycryptodome>=3.20.0 +python-dotenv>=1.0.1 \ No newline at end of file diff --git a/backend/src/main.py b/backend/src/main.py index 66cc0bc..fb0d721 100644 --- a/backend/src/main.py +++ b/backend/src/main.py @@ -6,6 +6,17 @@ app = FastAPI(title="HanchuESS Solar Backend API") def root(): return {"message": "Welcome to the HanchuESS Solar Backend API!"} +@app.get("/get_access_token", tags=["HanchuESS"]) +def get_access_token(): + from service.hanchu_service import HanchuESSService + + hanchu_service = HanchuESSService() + try: + access_token = hanchu_service.get_access_token() + return {"access_token": access_token} + except Exception as e: + return {"error": str(e)} + if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8050) \ No newline at end of file diff --git a/backend/src/service/hanchu_service.py b/backend/src/service/hanchu_service.py new file mode 100644 index 0000000..9272dda --- /dev/null +++ b/backend/src/service/hanchu_service.py @@ -0,0 +1,80 @@ +import base64 +import json +import os +import requests + +from Crypto.Cipher import AES +from Crypto.Util.Padding import pad +from dotenv import load_dotenv + + +load_dotenv() + + +class HanchuESSService: + def __init__(self): + self.name = "HanchuESS Service" + + # Load config from environment + self.aes_key = os.environ["HANCHU_AES_KEY"].encode("utf-8") + self.aes_iv = os.environ["HANCHU_AES_IV"].encode("utf-8") + self.login_url = os.environ["HANCHU_LOGIN_URL"] + self.login_type = os.getenv("HANCHU_LOGIN_TYPE", "ACCOUNT") + self.timeout = int(os.getenv("HANCHU_HTTP_TIMEOUT", "10")) + self.verify_ssl = os.getenv("HANCHU_VERIFY_SSL", "true").lower() == "true" + self.hanchu_username = os.getenv("HANCHU_USERNAME", "") + self.hanchu_password = os.getenv("HANCHU_PASSWORD", "") + + # Safety checks + if len(self.aes_key) not in (16, 24, 32): + raise ValueError("AES key must be 16, 24, or 32 bytes") + if len(self.aes_iv) != 16: + raise ValueError("AES IV must be exactly 16 bytes") + + def encrypt_payload(self, data: dict | str) -> str: + """ + Encrypt payload using AES-CBC and return base64 string. + """ + if not isinstance(data, str): + data = json.dumps(data, separators=(",", ":")) + + cipher = AES.new(self.aes_key, AES.MODE_CBC, self.aes_iv) + ciphertext = cipher.encrypt(pad(data.encode("utf-8"), AES.block_size)) + return base64.b64encode(ciphertext).decode("utf-8") + + def get_access_token(self) -> str: + """ + Authenticate with Hanchu ESS and return access token. + """ + + payload = { + "account": self.hanchu_username, + "password": self.hanchu_password, + "loginType": self.login_type, + } + + encrypted_payload = self.encrypt_payload(payload) + + headers = { + "Content-Type": "application/json", + "Accept": "application/json", + "User-Agent": "Mozilla/5.0", + } + + response = requests.post( + self.login_url, + json={"data": encrypted_payload}, + headers=headers, + timeout=self.timeout, + verify=self.verify_ssl, + ) + + response.raise_for_status() + result = response.json() + + try: + return result["data"]["accessToken"] + except (KeyError, TypeError): + raise RuntimeError( + f"Hanchu login failed: {json.dumps(result, ensure_ascii=False)}" + ) diff --git a/docker-compose.yml b/docker-compose.yml index c74fef8..3245747 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -5,4 +5,15 @@ services: dockerfile: docker/Dockerfile container_name: hanchuess-solar-backend ports: - - "8050:8050" \ No newline at end of file + - "8050:8050" + env_file: + - .env + environment: + - HANCHU_AES_KEY=${HANCHU_AES_KEY} + - HANCHU_AES_IV=${HANCHU_AES_IV} + - HANCHU_LOGIN_URL=${HANCHU_LOGIN_URL} + - HANCHU_LOGIN_TYPE=${HANCHU_LOGIN_TYPE:-ACCOUNT} + - HANCHU_HTTP_TIMEOUT=${HANCHU_HTTP_TIMEOUT:-10} + - HANCHU_VERIFY_SSL=${HANCHU_VERIFY_SSL:-true} + - HANCHU_USERNAME=${HANCHU_USERNAME:-} + - HANCHU_PASSWORD=${HANCHU_PASSWORD:-} \ No newline at end of file -- 2.49.1 From 680335fba1bde44b2c0bcedce9fe7d89bba5c7eb Mon Sep 17 00:00:00 2001 From: Luke Calladine Date: Tue, 13 Jan 2026 16:14:20 +0000 Subject: [PATCH 2/3] Initial HanchuESS services --- backend/src/main.py | 62 ++++++++ backend/src/service/hanchu_service.py | 202 +++++++++++++++++++++++++- 2 files changed, 256 insertions(+), 8 deletions(-) diff --git a/backend/src/main.py b/backend/src/main.py index fb0d721..1c04cb1 100644 --- a/backend/src/main.py +++ b/backend/src/main.py @@ -1,13 +1,20 @@ from fastapi import FastAPI +from pydantic import BaseModel app = FastAPI(title="HanchuESS Solar Backend API") + +class DecryptRequest(BaseModel): + encrypted_payload: str + + @app.get("/", tags=["Root"]) def root(): return {"message": "Welcome to the HanchuESS Solar Backend API!"} @app.get("/get_access_token", tags=["HanchuESS"]) def get_access_token(): + """Get access token by logging into HanchuESS""" from service.hanchu_service import HanchuESSService hanchu_service = HanchuESSService() @@ -17,6 +24,61 @@ def get_access_token(): except Exception as e: return {"error": str(e)} +@app.post("/decrypt_payload", tags=["Payload"]) +def decrypt_payload(request: DecryptRequest): + """Decrypt an AES-encrypted HanchuESS payload""" + from service.hanchu_service import HanchuESSService + + try: + hanchu_service = HanchuESSService() + decrypted_data = hanchu_service.decrypt_payload(request.encrypted_payload) + + return { + "decrypted_data": decrypted_data, + "data_type": type(decrypted_data).__name__ + } + except Exception as e: + import traceback + return {"error": str(e), "traceback": traceback.format_exc()} + +@app.get("/get_power_chart", tags=["HanchuESS"]) +def get_power_chart(): + """Get 65-second power chart data from HanchuESS""" + from service.hanchu_service import HanchuESSService + + try: + hanchu_service = HanchuESSService() + + # Get power chart data (will automatically handle authentication) + power_data = hanchu_service.get_power_chart() + + return power_data + except Exception as e: + import traceback + return {"error": str(e), "traceback": traceback.format_exc()} + +@app.get("/get_power_minute_chart", tags=["HanchuESS"]) +def get_power_minute_chart(start_ts: int = None, end_ts: int = None): + """Get minute-by-minute power chart data from HanchuESS + + Args: + start_ts: Optional start timestamp in milliseconds + end_ts: Optional end timestamp in milliseconds + """ + from service.hanchu_service import HanchuESSService + + try: + hanchu_service = HanchuESSService() + + # Get minute chart data (will automatically handle authentication) + chart_data = hanchu_service.get_power_minute_chart(start_ts=start_ts, end_ts=end_ts) + + return chart_data + except Exception as e: + import traceback + return {"error": str(e), "traceback": traceback.format_exc()} + + if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8050) \ No newline at end of file diff --git a/backend/src/service/hanchu_service.py b/backend/src/service/hanchu_service.py index 9272dda..dca04eb 100644 --- a/backend/src/service/hanchu_service.py +++ b/backend/src/service/hanchu_service.py @@ -3,8 +3,9 @@ import json import os import requests -from Crypto.Cipher import AES -from Crypto.Util.Padding import pad +from Crypto.Cipher import AES, PKCS1_v1_5 +from Crypto.PublicKey import RSA +from Crypto.Util.Padding import pad, unpad from dotenv import load_dotenv @@ -12,6 +13,13 @@ load_dotenv() class HanchuESSService: + # RSA public key from the JavaScript code + RSA_PUBLIC_KEY = """-----BEGIN PUBLIC KEY----- +MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQCVg7RFDLMGM4O98d1zWKI5RQan +jci3iY4qlpgsH76fUn3GnZtqjbRk37lCQDv6AhgPNXRPpty81+g909/c4yzySKaP +CcDZv7KdCRB1mVxkq+0z4EtKx9EoTXKnFSDBaYi2srdal1tM3gGOsNTDN58CzYPX +nDGPX7+EHS1Mm4aVDQIDAQAB +-----END PUBLIC KEY-----""" def __init__(self): self.name = "HanchuESS Service" @@ -24,6 +32,10 @@ class HanchuESSService: self.verify_ssl = os.getenv("HANCHU_VERIFY_SSL", "true").lower() == "true" self.hanchu_username = os.getenv("HANCHU_USERNAME", "") self.hanchu_password = os.getenv("HANCHU_PASSWORD", "") + self.base_serial_number = os.getenv("BASE_SERIAL_NUMBER", "") + + # Cache for access token + self._access_token = None # Safety checks if len(self.aes_key) not in (16, 24, 32): @@ -42,28 +54,72 @@ class HanchuESSService: ciphertext = cipher.encrypt(pad(data.encode("utf-8"), AES.block_size)) return base64.b64encode(ciphertext).decode("utf-8") + def decrypt_payload(self, encrypted_data: str) -> dict | str: + """ + Decrypt base64-encoded AES-CBC payload and return the original data. + """ + ciphertext = base64.b64decode(encrypted_data) + cipher = AES.new(self.aes_key, AES.MODE_CBC, self.aes_iv) + decrypted = unpad(cipher.decrypt(ciphertext), AES.block_size) + decrypted_str = decrypted.decode("utf-8") + + # Try to parse as JSON, return string if it fails + try: + return json.loads(decrypted_str) + except json.JSONDecodeError: + return decrypted_str + + def encrypt_password_rsa(self, password: str) -> str: + """ + Encrypt password using RSA public key (matches JavaScript GO function). + Returns base64-encoded encrypted password. + """ + public_key = RSA.import_key(self.RSA_PUBLIC_KEY) + cipher = PKCS1_v1_5.new(public_key) + encrypted = cipher.encrypt(password.encode('utf-8')) + return base64.b64encode(encrypted).decode('utf-8') + def get_access_token(self) -> str: """ Authenticate with Hanchu ESS and return access token. + Uses double encryption: RSA for password, then AES for entire payload. + Caches the token to avoid unnecessary logins. """ + # Return cached token if available + if self._access_token: + return self._access_token + + # Step 1: RSA encrypt the password + encrypted_password = self.encrypt_password_rsa(self.hanchu_password) + # Step 2: Build payload with encrypted password payload = { "account": self.hanchu_username, - "password": self.hanchu_password, + "pwd": encrypted_password, "loginType": self.login_type, } + # Step 3: AES encrypt the entire payload encrypted_payload = self.encrypt_payload(payload) + # Step 4: Send to API with correct headers headers = { - "Content-Type": "application/json", - "Accept": "application/json", - "User-Agent": "Mozilla/5.0", + "Content-Type": "text/plain", + "Accept": "application/json, text/plain, */*", + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", + "appplat": "iess", + "locale": "en", + "timezone": "Africa/Accra", + "timeselected": "GMT", + "version": "1.0", + "crypto-version": "1.0.0", + "Origin": "https://iess3.hanchuess.com", + "Referer": "https://iess3.hanchuess.com/login", } response = requests.post( self.login_url, - json={"data": encrypted_payload}, + data=encrypted_payload, headers=headers, timeout=self.timeout, verify=self.verify_ssl, @@ -73,8 +129,138 @@ class HanchuESSService: result = response.json() try: - return result["data"]["accessToken"] + # The token is directly in the 'data' field as a JWT string + if result.get("success") and result.get("data"): + self._access_token = result["data"] + return self._access_token + else: + raise RuntimeError( + f"Hanchu login failed: {json.dumps(result, ensure_ascii=False)}" + ) except (KeyError, TypeError): raise RuntimeError( f"Hanchu login failed: {json.dumps(result, ensure_ascii=False)}" ) + + def get_power_chart(self, access_token: str = None) -> dict: + """ + Get 65-second power chart data from HanchuESS. + + Args: + access_token: Optional JWT token from login. If not provided, will get one automatically. + + Returns: + Power chart data from the API + """ + # Get access token if not provided + if not access_token: + # Check if we have a cached token first + if self._access_token: + access_token = self._access_token + else: + access_token = self.get_access_token() + + # Build payload with serial number + payload = {"sn": self.base_serial_number} + + # AES encrypt the payload + encrypted_payload = self.encrypt_payload(payload) + + # Send to API with access token + url = "https://iess3.hanchuess.com/gateway/platform/pcs/powerChart" + headers = { + "Content-Type": "text/plain", + "Accept": "application/json, text/plain, */*", + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", + "appplat": "iess", + "locale": "en", + "timezone": "Africa/Accra", + "timeselected": "GMT", + "version": "1.0", + "crypto-version": "1.0.0", + "access-token": access_token, + "Origin": "https://iess3.hanchuess.com", + "Referer": "https://iess3.hanchuess.com/", + } + + response = requests.post( + url, + data=encrypted_payload, + headers=headers, + timeout=self.timeout, + verify=self.verify_ssl, + ) + + response.raise_for_status() + return response.json() + + def get_power_minute_chart(self, access_token: str = None, start_ts: int = None, end_ts: int = None) -> dict: + """ + Get minute-by-minute power chart data from HanchuESS. + + Args: + access_token: Optional JWT token from login. If not provided, will get one automatically. + start_ts: Start timestamp in milliseconds. If not provided, defaults to start of today. + end_ts: End timestamp in milliseconds. If not provided, defaults to end of today. + + Returns: + Power minute chart data from the API + """ + # Get access token if not provided + if not access_token: + # Check if we have a cached token first + if self._access_token: + access_token = self._access_token + else: + access_token = self.get_access_token() + + # Set default timestamps if not provided (today's data) + if not start_ts or not end_ts: + from datetime import datetime, timezone, timedelta + now = datetime.now(timezone.utc) + # 30 minutes ago + start_time = now - timedelta(minutes=30) + start_ts = int(start_time.timestamp() * 1000) + # Current time + end_ts = int(now.timestamp() * 1000) + + # Build payload + payload = { + "sn": self.base_serial_number, + "devType": "2", + "maxCount": 1440, # 24 hours * 60 minutes + "dataTimeTsEnd": end_ts, + "dataTimeTsStart": start_ts, + "masterSum": True + } + + # AES encrypt the payload + encrypted_payload = self.encrypt_payload(payload) + + # Send to API with access token + url = "https://iess3.hanchuess.com/gateway/platform/pcs/powerMinuteChart" + headers = { + "Content-Type": "text/plain", + "Accept": "application/json, text/plain, */*", + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", + "appplat": "iess", + "locale": "en", + "timezone": "Africa/Accra", + "timeselected": "GMT", + "version": "1.0", + "crypto-version": "1.0.0", + "access-token": access_token, + "Origin": "https://iess3.hanchuess.com", + "Referer": "https://iess3.hanchuess.com/", + } + + response = requests.post( + url, + data=encrypted_payload, + headers=headers, + timeout=self.timeout, + verify=self.verify_ssl, + ) + + response.raise_for_status() + return response.json() -- 2.49.1 From d073456c2f29280f8713445655d2364a0f5681e8 Mon Sep 17 00:00:00 2001 From: Preteck <123873832+preteckdotdev@users.noreply.github.com> Date: Tue, 13 Jan 2026 19:37:22 +0000 Subject: [PATCH 3/3] Update README.md --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index c712fd2..b884f44 100644 --- a/README.md +++ b/README.md @@ -8,4 +8,4 @@ The aims of this project are to: - [ ] Scrape historical data of a HanchuESS station. - [ ] Scrape current data of a HanchuESS station. - [ ] Send commands to the HanchuESS station. -- [ ] Create an integration API that provides QOL improvements to the HanchuESS station. \ No newline at end of file +- [ ] Create an integration API that provides QOL improvements to the HanchuESS station. -- 2.49.1