#!/usr/bin/env python3 import zipfile import json import re from datetime import datetime import argparse import statistics import logging, logging.config import sys def calculate_power_hour_ratio(start, end): time_format = "%d.%m.%y, %H:%M" start_time = datetime.strptime(start['time'], time_format) end_time = datetime.strptime(end['time'], time_format) time_diff_hours = (end_time - start_time).total_seconds() / 3600 logger.debug(f"{time_diff_hours=}, {start_time=}, {end_time=}, {start['power']=}, {end['power']=}, {start['charging']=}, {end['charging']=}") if time_diff_hours == 0: return None return (end['power'] - start['power']) / time_diff_hours def process_zip(zip_file_path): logger.info(f"Processing zip file at '{zip_file_path}'...") data_points = [] with zipfile.ZipFile(zip_file_path, 'r') as zip_file: sorted_files = [] for file_name in zip_file.namelist(): if re.match(r'^[^/]+-(\d+)\.json$', file_name): sorted_files.append((re.search(r'^[^/]+-(\d+)\.json$', file_name).group(1), file_name)) elif re.match(r'^[^/]+\.json$', file_name): sorted_files.append(("1", file_name)) for _, file_name in sorted_files: logger.debug(f"Parsing file: {file_name}") with zip_file.open(file_name) as json_file: data = json.load(json_file) data_points.append(data) logger.info(f"Zip file successfully processed, {len(data_points)} data-points extracted...") data_points.sort(key=lambda x: datetime.strptime(x["time"], "%d.%m.%y, %H:%M")) return data_points def calculate_power_ratios(data_points): logger.info(f"Calculating power ratios...") discharging = [] charging = [] ignored_discharging_periods = 0 ignored_charging_periods = 0 for i in range(1, len(data_points)): start = data_points[i-1] end = data_points[i] power_hour_ratio = calculate_power_hour_ratio(start, end) if start['charging'] == False and end['charging'] == True: if power_hour_ratio is None or power_hour_ratio > 0: ignored_discharging_periods += 1 logger.debug(f"Ignoring discharging period: {start} - {end} --> {power_hour_ratio}...") continue discharging.append(power_hour_ratio) elif start['charging'] == True and end['charging'] == False: if power_hour_ratio is None or power_hour_ratio < 0: ignored_charging_periods += 1 logger.debug(f"Ignoring charging period: {start} - {end}...") continue charging.append(power_hour_ratio) # elif start['charging'] == False and end['charging'] == False: # if power_hour_ratio is None or power_hour_ratio > 0: # ignored_discharging_periods += 1 # logger.debug(f"Ignoring discharging period: {start} - {end} --> {power_hour_ratio}...") # continue # discharging.append(power_hour_ratio) elif start['charging'] == True and end['charging'] == True: if power_hour_ratio is None or power_hour_ratio < 0: ignored_charging_periods += 1 logger.debug(f"Ignoring charging period: {start} - {end} --> {power_hour_ratio}...") continue charging.append(power_hour_ratio) else: logger.error(f"Unexpected ({'unusable' if power_hour_ratio is None else 'usable'}) datapoints: {start}, {end}") continue discharging_median = statistics.median(discharging) if len(discharging)>0 else 0 discharging_mean = statistics.mean(discharging) if len(discharging)>0 else 0 charging_median = statistics.median(charging) if len(charging)>0 else 0 charging_mean = statistics.mean(charging) if len(charging)>0 else 0 logger.info(f"Power ratios calculated: {len(discharging)+ignored_discharging_periods} discharging periods ({len(discharging)} usable), {len(charging)+ignored_charging_periods} charging periods ({len(charging)} usable)...") return discharging, discharging_median, discharging_mean, charging, charging_median, charging_mean parser = argparse.ArgumentParser(description="Process a zip file of JSON files containing power, time, and charging data.") parser.add_argument('--file', metavar='file.zip', required=True, help="Path to the zip file") parser.add_argument("--log", metavar='LOGLEVEL', help="Loglevel to log", default="INFO") args = parser.parse_args() logging.config.dictConfig({ "version": 1, "disable_existing_loggers": False, "formatters": { "simple": { "format": "%(asctime)s [%(levelname)-7s] %(name)s {%(threadName)s} %(filename)s:%(lineno)d: %(message)s", "color": True } }, "handlers": { "stderr": { "class": "logging.StreamHandler", "level": args.log, "formatter": "simple" }, "ignore": { "class": "logging.NullHandler", "level": "DEBUG" } }, "loggers": { "": { "level": "DEBUG", "handlers": ["stderr"] } } }) logger = logging.getLogger(__name__) data_points = process_zip(args.file) discharging, discharging_median, discharging_mean, charging, charging_median, charging_mean = calculate_power_ratios(data_points) print(f"{args.file}: Discharging ratios (median: {discharging_median:.3f}, mean: {discharging_mean:.3f}):", discharging) print(f"{args.file}: Charging ratios (median: {charging_median:.3f}, mean: {charging_mean:.3f}):", charging)