-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
f16acbb
commit 3b6e479
Showing
5 changed files
with
268 additions
and
7 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,180 @@ | ||
| # == csv_to_google_sheet.py Author: Zuinige Rijder ===================================== | ||
| """ | ||
| Simple Python3 script to update google sheet with | ||
| csv input files generated by smart_plug_mini.py | ||
| """ | ||
| import configparser | ||
| import os | ||
| import traceback | ||
| import time | ||
| from datetime import datetime | ||
| from pathlib import Path | ||
| from typing import Generator | ||
| import gspread | ||
|
|
||
|
|
||
| def log(msg: str) -> None: | ||
| """log a message prefixed with a date/time format yyyymmdd hh:mm:ss""" | ||
| print(datetime.now().strftime("%Y%m%d %H:%M:%S") + ": " + msg) | ||
|
|
||
|
|
||
| def read_reverse_order(file_name: str) -> Generator[str, None, None]: | ||
| """read in reverse order""" | ||
| # Open file for reading in binary mode | ||
| with open(file_name, "rb") as read_obj: | ||
| # Move the cursor to the end of the file | ||
| read_obj.seek(0, os.SEEK_END) | ||
| # Get the current position of pointer i.e eof | ||
| pointer_location = read_obj.tell() | ||
| # Create a buffer to keep the last read line | ||
| buffer = bytearray() | ||
| # Loop till pointer reaches the top of the file | ||
| while pointer_location >= 0: | ||
| # Move the file pointer to the location pointed by pointer_location | ||
| read_obj.seek(pointer_location) | ||
| # Shift pointer location by -1 | ||
| pointer_location = pointer_location - 1 | ||
| # read that byte / character | ||
| new_byte = read_obj.read(1) | ||
| # If the read byte is newline character then one line is read | ||
| if new_byte == b"\n": | ||
| # Fetch the line from buffer and yield it | ||
| yield buffer.decode()[::-1] | ||
| # Reinitialize the byte array to save next line | ||
| buffer = bytearray() | ||
| else: | ||
| # If last read character is not eol then add it in buffer | ||
| buffer.extend(new_byte) | ||
| # If there is still data in buffer, then it is first line. | ||
| if len(buffer) > 0: | ||
| # Yield the first line too | ||
| yield buffer.decode()[::-1] | ||
|
|
||
|
|
||
| def get_last_line(filename: Path) -> str: | ||
| """get last line of filename""" | ||
| last_line = "" | ||
| if filename.is_file(): | ||
| with open(filename.name, "rb") as file: | ||
| try: | ||
| file.seek(-2, os.SEEK_END) | ||
| while file.read(1) != b"\n": | ||
| file.seek(-2, os.SEEK_CUR) | ||
| except OSError: | ||
| file.seek(0) | ||
| last_line = file.readline().decode().strip() | ||
| print(f"# last line {filename.name}: {last_line}") | ||
| return last_line | ||
|
|
||
|
|
||
| def read_csv_and_write_to_sheet( | ||
| array: list, name: str, period: str, params: tuple[str, int, int, int, int, int] | ||
| ) -> list: | ||
| """read_csv_and_write_to_sheet""" | ||
| last_line = params[0].split(",") | ||
| last_line_index = params[1] | ||
| startrow = params[2] | ||
| endrow = startrow + params[3] | ||
| strip_begin = params[4] | ||
| strip_end = params[5] | ||
|
|
||
| row = startrow | ||
| array.append({"range": f"A{row}:B{row}", "values": [[period, "kWh"]]}) | ||
| row += 1 | ||
| if len(last_line) == 7: | ||
| date_str = last_line[0].strip()[strip_begin:strip_end] | ||
| kwh = float(last_line[last_line_index].strip()) | ||
| array.append({"range": f"A{row}:B{row}", "values": [[date_str, kwh]]}) | ||
| row += 1 | ||
| if period == "Hour": | ||
| csv_postfix = "" | ||
| else: | ||
| csv_postfix = f".{period.lower()}s" | ||
| csv_filename = Path(f"{name}{csv_postfix}.csv") | ||
| if csv_filename.is_file(): | ||
| log(f"### read_reverse_csv_file: {csv_filename.name} " + "#" * 30) | ||
| for line in read_reverse_order(csv_filename.name): | ||
| line = line.strip() | ||
| if line == "": | ||
| continue | ||
| print(line) | ||
| splitted = line.split(",") | ||
| if len(splitted) > 2 and splitted[0].startswith("20"): | ||
| date_str = splitted[0].strip()[strip_begin:strip_end] | ||
| kwh = float(splitted[2].strip()) | ||
| array.append({"range": f"A{row}:B{row}", "values": [[date_str, kwh]]}) | ||
| row += 1 | ||
|
|
||
| if row > endrow: | ||
| break # finished | ||
| else: | ||
| log(f"Warning: csv file {csv_filename.name} does not exist") | ||
|
|
||
| return array | ||
|
|
||
|
|
||
| def write_to_sheet(name: str, sheet: list) -> None: | ||
| """write_to_sheet""" | ||
| array = [] | ||
| row = 1 | ||
| last = get_last_line(Path(f"{name}.csv")) | ||
| last_split = last.split(",") | ||
| if len(last_split) != 7 or not last_split[0].strip().startswith("20"): | ||
| log(f"ERROR: unexpected last line in {name}.csv: {last}") | ||
| return | ||
| header = [["Date", "Time", "kWh", "Hour", "Day", "Week", "Month", "Year"]] | ||
| array.append({"range": f"A{row}:H{row}", "values": header}) | ||
| row += 1 | ||
| first_line = [ | ||
| [ | ||
| last_split[0].strip().split(" ")[0].strip()[2:], | ||
| last_split[0].strip().split(" ")[1].strip(), | ||
| float(last_split[1].strip()), | ||
| float(last_split[2].strip()), | ||
| float(last_split[3].strip()), | ||
| float(last_split[4].strip()), | ||
| float(last_split[5].strip()), | ||
| float(last_split[6].strip()), | ||
| ] | ||
| ] | ||
| array.append({"range": f"A{row}:H{row}", "values": first_line}) | ||
|
|
||
| # params(last_line, last_line_index, start, rows, strip_start, strip_end) | ||
| array = read_csv_and_write_to_sheet(array, name, "Hour", ("", 2, 3, 49, 8, 16)) | ||
| array = read_csv_and_write_to_sheet(array, name, "Day", (last, 3, 54, 33, 5, 10)) | ||
| array = read_csv_and_write_to_sheet(array, name, "Week", (last, 4, 89, 28, 5, 10)) | ||
| array = read_csv_and_write_to_sheet(array, name, "Month", (last, 5, 119, 26, 0, 7)) | ||
| array = read_csv_and_write_to_sheet(array, name, "Year", (last, 6, 147, 25, 0, 4)) | ||
|
|
||
| sheet.batch_update(array) | ||
|
|
||
|
|
||
| def main() -> None: | ||
| """main""" | ||
| parser = configparser.ConfigParser() | ||
| parser.read("smart_plug_mini.cfg") | ||
| sp3s_settings = dict(parser.items("smart_plug_mini")) | ||
|
|
||
| client = gspread.service_account() | ||
| for name in sp3s_settings["device_names"].split(","): | ||
| name = name.strip() | ||
| retries = 2 | ||
| while retries > 0: | ||
| try: | ||
| spreadsheet_name = f"{name}.SP" | ||
| log(f"##### Writing {spreadsheet_name} " + "#" * 60) | ||
| spreadsheet = client.open(spreadsheet_name) | ||
| sheet = spreadsheet.sheet1 | ||
| sheet.clear() | ||
| write_to_sheet(name, sheet) | ||
|
|
||
| retries = 0 | ||
| except Exception as ex: # pylint: disable=broad-except | ||
| log("Exception: " + str(ex)) | ||
| traceback.print_exc() | ||
| retries -= 1 | ||
| log("Sleeping a minute") | ||
| time.sleep(60) | ||
|
|
||
|
|
||
| main() |
Binary file not shown.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,3 +1,4 @@ | ||
| python_version>=3.9 | ||
| broadlink==0.18.3 | ||
| python_dateutil==2.8.2 | ||
| gspread==5.6.2 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters