Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions .cspell/custom-dictionary-workspace.txt
Original file line number Diff line number Diff line change
Expand Up @@ -119,6 +119,8 @@ gridconsumptionpower
growatt
HACS
hadashboard
hahistory
hainterface
hanres
HAOS
hass
Expand Down Expand Up @@ -277,6 +279,7 @@ sigenergy
sigenstor
Slee
socb
socketloop
socs
sofar
SolarEdge
Expand Down Expand Up @@ -332,6 +335,7 @@ weblink
welink
workmode
writeonly
wrongsha
xaxis
xaxistooltip
xlabel
Expand Down
3 changes: 3 additions & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -94,3 +94,6 @@ apps.mod.yaml
apps.yaml.mark
comparisons.yaml
it

# Runtime-generated manifest
apps/predbat/manifest.yaml
269 changes: 224 additions & 45 deletions apps/predbat/download.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,62 @@

import os
import requests
import yaml
import hashlib


def get_github_directory_listing(tag):
"""
Get the list of files in the apps/predbat directory from GitHub

Args:
tag (str): The tag to query (e.g. v1.0.0)
Returns:
list: List of file metadata dicts from GitHub API, or None on failure
"""
url = "https://api.github.com/repos/springfall2008/batpred/contents/apps/predbat?ref={}".format(tag)
print("Fetching directory listing from {}".format(url))
try:
r = requests.get(url, headers={})
if r.ok:
data = r.json()
# Filter out directories, keep only files with full metadata
files = []
for item in data:
if item.get("type") == "file":
files.append(item)
print("Found {} files in directory".format(len(files)))
return files
else:
print("Error: Failed to fetch directory listing, status code: {}".format(r.status_code))
return None
except Exception as e:
print("Error: Exception while fetching directory listing: {}".format(e))
return None


def compute_file_sha1(filepath):
"""
Compute Git blob SHA1 hash of a file (matches GitHub's SHA)
Git computes SHA as: sha1("blob " + filesize + "\0" + contents)

Args:
filepath (str): Path to the file
Returns:
str: Git blob SHA1 hash as hex string, or None on error
"""
try:
sha1 = hashlib.sha1()
with open(filepath, "rb") as f:
data = f.read()

# Compute Git blob SHA: sha1("blob " + size + "\0" + contents)
header = "blob {}\0".format(len(data)).encode("utf-8")
sha1.update(header + data)
return sha1.hexdigest()
except Exception as e:
print("Error: Failed to compute SHA1 for {}: {}".format(filepath, e))
return None


def download_predbat_file_from_github(tag, filename, new_filename):
Expand Down Expand Up @@ -42,6 +98,8 @@ def predbat_update_move(version, files):
"""
Move the updated files into place
"""
if not files:
return False
tag_split = version.split(" ")
if tag_split:
tag = tag_split[0]
Expand All @@ -55,40 +113,88 @@ def predbat_update_move(version, files):
return False


def get_files_from_predbat(predbat_code):
files = ["predbat.py"]
for line in predbat_code.split("\n"):
if line.startswith("PREDBAT_FILES"):
files = line.split("=")[1].strip()
files = files.replace("[", "")
files = files.replace("]", "")
files = files.replace('"', "")
files = files.replace(" ", "")
files = files.split(",")
break
return files


def check_install():
def check_install(version):
"""
Check if Predbat is installed correctly

Args:
version (str): The version string (e.g. v8.30.8)
"""
this_path = os.path.dirname(__file__)
predbat_file = os.path.join(this_path, "predbat.py")
if os.path.exists(predbat_file):
with open(predbat_file, "r") as han:
predbat_code = han.read()
files = get_files_from_predbat(predbat_code)
for file in files:
filepath = os.path.join(this_path, file)
if not os.path.exists(filepath):
print("Error: File {} is missing".format(filepath))
return False
if os.path.getsize(filepath) == 0:
print("Error: File {} is zero bytes".format(filepath))
return False
return True
return False
manifest_file = os.path.join(this_path, "manifest.yaml")

# Check if manifest exists
if not os.path.exists(manifest_file):
print("Warn: Manifest file {} is missing, bypassing checks...".format(manifest_file))
# Try to download manifest from GitHub
tag_split = version.split(" ")
if tag_split:
tag = tag_split[0]
file_list = get_github_directory_listing(tag)
if file_list:
# Sort files alphabetically
file_list_sorted = sorted(file_list, key=lambda x: x["name"])
# Create manifest
try:
with open(manifest_file, "w") as f:
yaml.dump(file_list_sorted, f, default_flow_style=False, sort_keys=False)
print("Downloaded and created manifest file")
except Exception as e:
print("Error: Failed to write manifest: {}".format(e))
return True, False # Continue without manifest
else:
print("Warn: Failed to download manifest from GitHub")
return True, False # Continue without manifest
else:
return True, False # Continue without manifest

# Load and validate against manifest
try:
with open(manifest_file, "r") as f:
files = yaml.safe_load(f)

if not files:
print("Error: Manifest is empty")
return False

validation_passed = True
validation_modified = False

for file_info in files:
filename = file_info.get("name")
expected_size = file_info.get("size", 0)
expected_sha = file_info.get("sha")
filepath = os.path.join(this_path, filename)

# Check file exists
if not os.path.exists(filepath):
print("Error: File {} is missing".format(filepath))
validation_passed = False
continue

# Check file is not zero bytes
actual_size = os.path.getsize(filepath)
if actual_size == 0:
print("Error: File {} is zero bytes".format(filepath))
validation_passed = False
continue

# Warn on size mismatch but don't fail
if actual_size != expected_size:
print("Warn: File {} size mismatch: expected {}, got {}".format(filepath, expected_size, actual_size))
validation_modified = True
elif expected_sha:
# Warn on SHA mismatch but don't fail
actual_sha = compute_file_sha1(filepath)
if actual_sha and actual_sha != expected_sha:
print("Warn: File {} SHA mismatch: expected {}, got {}".format(filepath, expected_sha, actual_sha))
validation_modified = True

return validation_passed, validation_modified

except Exception as e:
print("Error: Failed to load manifest: {}".format(e))
return False


def predbat_update_download(version):
Expand All @@ -100,19 +206,92 @@ def predbat_update_download(version):
if tag_split:
tag = tag_split[0]

# Download predbat.py
file = "predbat.py"
predbat_code = download_predbat_file_from_github(tag, file, os.path.join(this_path, file + "." + tag))
if predbat_code:
# Get the list of other files to download by searching for PREDBAT_FILES in predbat.py
files = get_files_from_predbat(predbat_code)

# Download the remaining files
if files:
for file in files:
# Download the remaining files
if file != "predbat.py":
if not download_predbat_file_from_github(tag, file, os.path.join(this_path, file + "." + tag)):
return None
return files
# Get the list of files from GitHub API
file_list = get_github_directory_listing(tag)
if not file_list:
print("Error: Failed to get file list from GitHub")
return None

# Download all files
downloaded_files = []
for file_info in file_list:
filename = file_info["name"]
if not download_predbat_file_from_github(tag, filename, os.path.join(this_path, filename + "." + tag)):
print("Error: Failed to download {}".format(filename))
return None
downloaded_files.append(filename)

# Sort files alphabetically
file_list_sorted = sorted(file_list, key=lambda x: x["name"])

# Generate manifest.yaml (just the sorted file list from GitHub API)
manifest_filename = os.path.join(this_path, "manifest.yaml." + tag)
try:
with open(manifest_filename, "w") as f:
yaml.dump(file_list_sorted, f, default_flow_style=False, sort_keys=False)
print("Generated manifest: {}".format(manifest_filename))
except Exception as e:
print("Error: Failed to write manifest: {}".format(e))
return None

# Return list of files including manifest
downloaded_files.append("manifest.yaml")
return downloaded_files
return None


def main(): # pragma: no cover
"""
Main function for standalone testing of download functionality
"""
import argparse
import sys

# Add parent directory to path so we can import download module
parent_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if parent_dir not in sys.path:
sys.path.insert(0, parent_dir)

parser = argparse.ArgumentParser(description="Test Predbat download functionality")
parser.add_argument("--check", metavar="VERSION", help="Check if Predbat is installed correctly for given version (e.g. v8.30.8)")
parser.add_argument("--download", metavar="VERSION", help="Download Predbat version from GitHub (e.g. v8.30.8)")

args = parser.parse_args()

if args.check:
print("=" * 60)
print("Checking Predbat installation for version: {}".format(args.check))
print("=" * 60)
result, modified = check_install(args.check)
if result:
if modified:
print("Warn: Installation check PASSED with modifications")
else:
print("\n✓ Installation check PASSED")
sys.exit(0)
else:
print("\n✗ Installation check FAILED")
sys.exit(1)

elif args.download:
print("=" * 60)
print("Downloading Predbat version: {}".format(args.download))
print("=" * 60)
files = predbat_update_download(args.download)
if files:
print("\n✓ Download successful!")
print("Files downloaded: {}".format(", ".join(files)))
predbat_update_move(args.download, files)
print("Files moved into place.")
sys.exit(0)
else:
print("\n✗ Download FAILED")
sys.exit(1)

else:
parser.print_help()
sys.exit(1)


if __name__ == "__main__":
main()
9 changes: 6 additions & 3 deletions apps/predbat/fox.py
Original file line number Diff line number Diff line change
Expand Up @@ -1517,8 +1517,11 @@ async def automatic_config(self):
self.set_arg("export_limit", [f"number.predbat_fox_{device}_setting_exportlimit" for device in batteries])
self.set_arg("schedule_write_button", [f"switch.predbat_fox_{device}_battery_schedule_charge_write" for device in batteries])

if len(batteries):
self.set_arg("battery_temperature_history", f"sensor.predbat_fox_{batteries[0]}_battemperature")

class MockBase:

class MockBase: # pragma: no cover
"""Mock base class for testing"""

def __init__(self):
Expand All @@ -1533,7 +1536,7 @@ def dashboard_item(self, *args, **kwargs):
print(f"DASHBOARD: {args}, {kwargs}")


async def test_fox_api(sn, api_key):
async def test_fox_api(sn, api_key): # pragma: no cover
"""
Run a test
"""
Expand Down Expand Up @@ -1625,7 +1628,7 @@ async def test_fox_api(sn, api_key):
# print(res)


def main():
def main(): # pragma: no cover
"""
Main function for command line execution
"""
Expand Down
20 changes: 10 additions & 10 deletions apps/predbat/ha.py
Original file line number Diff line number Diff line change
Expand Up @@ -734,17 +734,17 @@ def api_call(self, endpoint, data_in=None, post=False, core=True, silent=False):
"Content-Type": "application/json",
"Accept": "application/json",
}
if post:
if data_in:
response = requests.post(url, headers=headers, json=data_in, timeout=TIMEOUT)
else:
response = requests.post(url, headers=headers, timeout=TIMEOUT)
else:
if data_in:
response = requests.get(url, headers=headers, params=data_in, timeout=TIMEOUT)
else:
response = requests.get(url, headers=headers, timeout=TIMEOUT)
try:
if post:
if data_in:
response = requests.post(url, headers=headers, json=data_in, timeout=TIMEOUT)
else:
response = requests.post(url, headers=headers, timeout=TIMEOUT)
else:
if data_in:
response = requests.get(url, headers=headers, params=data_in, timeout=TIMEOUT)
else:
response = requests.get(url, headers=headers, timeout=TIMEOUT)
data = response.json()
self.api_errors = 0
except requests.exceptions.JSONDecodeError:
Expand Down
Loading