diff --git a/tools/README.md b/tools/README.md new file mode 100644 index 0000000..5dc6532 --- /dev/null +++ b/tools/README.md @@ -0,0 +1,48 @@ +# Tools + +During my investigation, I wanted to see the claimed wattage that corresponds to a given payload, but in an automated way. Since Emporia lets you export CSV data, I made a quick and dirty script to match up the `*1SEC.csv` file to the data I sniffed over UART. + +## emporia_sniff_parsing.py + +This script was crafted to deal with the firmware V7's payload, but you can adjust it to parse other kinds of payloads. + +### How to run + +`python3 .\emporia_sniff_parsing.py --sniffed_data capture_from_MGM111.txt --emporia_csv emporia_export_1SEC.csv` + +### What it expects + +`capture_from_MGM111.txt` should have contents like this: +``` +"2023-12-08 09:05:11",2401722C18480100000025877D16090000010000251A14970000000103002201000002030022E803000004002AA704000D"2023-12-08 09:05:41",2401722C18490100000025907D16090000010000251A14970000000103002201000002030022E803000004002A8F04000D"2023-12-08 09:06:11",2401722C184A01000000259A7D16090000010000251A14970000000103002201000002030022E803000004002AA304000D"2023-12-08 09:06:41",2401722C184B0100000025A37D16090000010000251A14970000000103002201000002030022E803000004002A5404000D"2023-12-08 09:07:11",2401722C184C0100000025AC7D16090000010000251A14970000000103002201000002030022E803000004002A8E04000D"2023-12-08 09:07:41",2401722C184D0100000025B77D16090000010000251A14970000000103002201000002030022E803000004002AD104000D +``` + +This was a format returned by [Realterm](https://realterm.sourceforge.io/), but you can adjust the code to match the format of your sniffed data. + +`emporia_export_1SEC.csv` should have contents like this: +``` +12/08/2023 08:55:40,1.3850 +12/08/2023 08:55:41,1.3850 +12/08/2023 08:55:42,1.3850 +12/08/2023 08:55:43,1.3850 +12/08/2023 08:55:44,1.3850 +12/08/2023 08:55:45,1.3850 +12/08/2023 08:55:46,1.4140 +12/08/2023 08:55:47,1.4140 +12/08/2023 08:55:48,1.4140 +``` + +### What to expect + +``` +Hex3: C90600 Calculated W: 1737 Pairing: [('1.6560', '2401722C18210100000025A97B16090000010000251A14970000000103002201000002030022E803000004002AC906000D')] +Hex3: 780600 Calculated W: 1656 Pairing: [('1.6790', '2401722C18230100000025C67B16090000010000251A14970000000103002201000002030022E803000004002A7806000D')] +Hex3: 8F0600 Calculated W: 1679 Pairing: [('1.6820', '2401722C18250100000025E07B16090000010000251A14970000000103002201000002030022E803000004002A8F06000D')] +Hex3: 7F0600 Calculated W: 1663 Pairing: [('1.6370', '2401722C18290100000025197C16090000010000251A14970000000103002201000002030022E803000004002A7F06000D')] +Hex3: 650600 Calculated W: 1637 Pairing: [('1.5340', '2401722C182B0100000025347C16090000010000251A14970000000103002201000002030022E803000004002A6506000D')] +Hex3: FE0500 Calculated W: 1534 Pairing: [('1.5260', '2401722C182D01000000254F7C16090000010000251A14970000000103002201000002030022E803000004002AFE05000D')] +Hex3: F60500 Calculated W: 1526 Pairing: [('1.5820', '2401722C182F0100000025697C16090000010000251A14970000000103002201000002030022E803000004002AF605000D')] +Hex3: 2E0600 Calculated W: 1582 Pairing: [('1.4740', '2401722C18310100000025827C16090000010000251A14970000000103002201000002030022E803000004002A2E06000D')] +Hex3: C20500 Calculated W: 1474 Pairing: [('1.3850', '2401722C183301000000259B7C16090000010000251A14970000000103002201000002030022E803000004002AC205000D')] +Hex3: 690500 Calculated W: 1385 Pairing: [('1.4140', '2401722C18350100000025B47C16090000010000251A14970000000103002201000002030022E803000004002A6905000D')] +``` \ No newline at end of file diff --git a/tools/emporia_sniff_parsing.py b/tools/emporia_sniff_parsing.py new file mode 100644 index 0000000..950deb9 --- /dev/null +++ b/tools/emporia_sniff_parsing.py @@ -0,0 +1,142 @@ +import argparse +import csv +from collections import defaultdict +import re +from datetime import datetime + +def parse_sniffed_packet(csv_data): + # Regular expression pattern for the timestamp + pattern = r'("\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}",)' + # Split the data using the pattern, keeping the delimiters + entries = re.split(pattern, csv_data)[1:] # first element is always empty + + parsed_data = {} + for i in range(0, len(entries), 2): + timestamp_str = entries[i].strip('",') + value = entries[i + 1] + + # Parse the timestamp string into a datetime object + timestamp = datetime.strptime(timestamp_str, '%Y-%m-%d %H:%M:%S') + + if timestamp in parsed_data: + # Concatenate value if timestamp already exists + parsed_data[timestamp] += value + else: + # Otherwise, just add the new entry + parsed_data[timestamp] = value + + return parsed_data + +def parse_emporia(csv_data): + parsed_data = {} + last_value = None + + reader = csv.reader(csv_data.splitlines()) + for row in reader: + # print(row) + # Skip blank lines + if not row or len(row) < 2: + continue + + timestamp_str, value = row + + # Skip if the value is the same as the last one + if value == last_value: + continue + + # Parse the timestamp string into a datetime object + try: + timestamp = datetime.strptime(timestamp_str, '%m/%d/%Y %H:%M:%S') + except ValueError: + continue + + # Update the last value and add the new entry + last_value = value + parsed_data[timestamp] = value + + return parsed_data + +def find_nearest_before(from_emporia, sniffed): + result = [] + for keyA, valueA in from_emporia.items(): + # Initialize nearest datetime and its value + nearest_time = None + nearest_value = None + # Go through each item in sniffed + for keyB, valueB in sniffed.items(): + # Check if keyB is before keyA and within 1-5 seconds + if 0 < (keyA - keyB).total_seconds() <= 5: + # Update nearest_time if it's closer to keyA than the previous nearest_time + if nearest_time is None or (keyA - keyB) < (keyA - nearest_time): + nearest_time = keyB + nearest_value = valueB + # Append to result if a match is found + if nearest_time: + result.append((nearest_time, valueA, nearest_value)) + return result + +def find_nearest_after(sniffed, from_emporia): + result = [] + for keyA, valueA in sniffed.items(): + # Initialize nearest datetime and its value + nearest_time = None + nearest_value = None + # Go through each item in dictB + for keyB, valueB in from_emporia.items(): + # Check if keyB is after keyA and within 1-5 seconds + if 1 < (keyB - keyA).total_seconds() <= 5: + # Update nearest_time if it's closer to keyA than the previous nearest_time + if nearest_time is None or (keyB - keyA) < (nearest_time - keyA): + nearest_time = keyB + nearest_value = valueB + # Append to result if a match is found + if nearest_time: + result.append((keyA, valueA, nearest_value)) + else: + print('have time', keyA, 'that doesnt have a nearest value') + return result + +def swap_endianness(hex_string): + # Split the string into bytes (two characters each) + bytes_list = [hex_string[i:i+2] for i in range(0, len(hex_string), 2)] + # Reverse the list of bytes and join them back into a string + reversed_hex = ''.join(reversed(bytes_list)) + return reversed_hex + +def hex_to_decimal(hex_string): + try: + # First, swap the endianness + swapped_hex = swap_endianness(hex_string) + # Then, convert the hexadecimal string to decimal + return int(swapped_hex, 16) + except ValueError: + return "Invalid hexadecimal number" + +def read_file(file_path): + with open(file_path, 'r') as file: + return file.read() + +def main(): + parser = argparse.ArgumentParser(description="Merge the data.") + parser.add_argument("--sniffed_data", required=True, help="Filepath to the sniffed UART traffic.") + parser.add_argument("--emporia_csv", required=True, help="Filepath to the Emporia 1SEC CSV.") + + args = parser.parse_args() + + sniffed_data = read_file(args.sniffed_data) + parsed_sniff = parse_sniffed_packet(sniffed_data) + + emporia_data = read_file(args.emporia_csv) + parsed_emporia = parse_emporia(emporia_data) + + hex3_to_watts = defaultdict(list) + result = find_nearest_after(parsed_sniff, parsed_emporia) + for timestamp, packet, watts in result: + hex3 = packet[90:96] + hex3_to_watts[hex3].append((watts, packet)) + + for hex3, watts in hex3_to_watts.items(): + print('Hex3:', hex3, 'Calculated W:', hex_to_decimal(hex3), 'Pairing:', watts) + +if __name__ == "__main__": + main() \ No newline at end of file