128 lines
3.1 KiB
Python
128 lines
3.1 KiB
Python
import json
|
|
import os
|
|
from datetime import datetime
|
|
|
|
BASE = "results"
|
|
VIN = 'WVWZZZE1ZMP010760'
|
|
YEAR = 2024
|
|
MONTH = 10
|
|
DAY = "18"
|
|
|
|
|
|
class Endpoint:
|
|
def __init__(self, _ep_path: str, _ts_path: str, _ts_format='%Y-%m-%dT%H:%M:%S.%f%z'):
|
|
self._ep_path = _ep_path
|
|
self._ep_path_keys = self.to_keys(_ep_path)
|
|
self._timestamps = []
|
|
self._values = []
|
|
|
|
self._ts_format = _ts_format
|
|
self._ts_path_keys = self.to_keys(_ts_path)
|
|
|
|
def __str__(self):
|
|
return f"{self._ep_path}:{len(self._values)}: {self._values}"
|
|
|
|
def timestamps_str(self):
|
|
return f"{self._ep_path}:{len(self._timestamps)}: {self._timestamps}"
|
|
|
|
@property
|
|
def path(self):
|
|
return self._ep_path
|
|
|
|
def assign(self, _record: dict):
|
|
value = self.deref_multi(_record, self._ep_path_keys)
|
|
self._values += [value]
|
|
|
|
# Extract timestamp
|
|
capture_time = self.deref_multi(_record, self._ts_path_keys)
|
|
timestamp_int = convert(capture_time, self._ts_format)
|
|
self._timestamps += [timestamp_int]
|
|
|
|
def remove_dups(self):
|
|
new_dict = {}
|
|
index = 0
|
|
unique_indices = []
|
|
for ts in self._timestamps:
|
|
if ts not in new_dict:
|
|
new_dict[ts] = index
|
|
unique_indices.append(index)
|
|
index += 1
|
|
|
|
t_list = self._timestamps
|
|
v_list = self._values
|
|
self._timestamps = []
|
|
self._values = {}
|
|
|
|
self._timestamps = [t_list[i] for i in unique_indices]
|
|
self._values = [v_list[i] for i in unique_indices]
|
|
|
|
@staticmethod
|
|
def to_keys(key_path: str):
|
|
keys = key_path.split('/')
|
|
return keys
|
|
|
|
@staticmethod
|
|
def deref_multi(data, keys):
|
|
return Endpoint.deref_multi(data[keys[0]], keys[1:]) if keys else data
|
|
|
|
|
|
end_points = [
|
|
Endpoint('data/measurements/odometerStatus/value/odometer', 'info/timestamp'),
|
|
Endpoint('data/measurements/fuelLevelStatus/value/currentSOC_pct', 'info/timestamp'),
|
|
Endpoint('data/vehicleLights/lightsStatus/value/lights', 'info/timestamp'),
|
|
Endpoint('data/fuelStatus/rangeStatus/value/primaryEngine', 'info/timestamp')
|
|
]
|
|
|
|
|
|
def convert(timestamp_str: str, pattern: str = "%Y-%m-%dT%H:%M:%S.%f%z"):
|
|
ts = datetime.strptime(timestamp_str, pattern)
|
|
result = ts.year
|
|
result = 100*result + ts.month
|
|
result = 100*result + ts.day
|
|
result = 100*result + ts.hour
|
|
result = 100*result + ts.minute
|
|
result = 100*result + ts.second
|
|
result = 1000000*result + ts.microsecond
|
|
|
|
return result
|
|
|
|
|
|
path = os.path.join(BASE, VIN)
|
|
path = os.path.join(path, str(YEAR))
|
|
path = os.path.join(path, str(MONTH))
|
|
|
|
day_list = os.listdir(path)
|
|
print("Days in '", path, "' :")
|
|
# prints all files
|
|
print(day_list)
|
|
|
|
month_path = path
|
|
record_list = []
|
|
for day in day_list:
|
|
if DAY is not None:
|
|
if day not in DAY:
|
|
continue
|
|
day_path = os.path.join(month_path, day)
|
|
file_list = os.listdir(day_path)
|
|
print("Files in '", day_path, "' :")
|
|
# prints all files
|
|
print(file_list)
|
|
|
|
for filename in file_list:
|
|
file_path = os.path.join(day_path, filename)
|
|
with open(file_path, "r") as fp:
|
|
records = json.load(fp)
|
|
for record in records:
|
|
try:
|
|
for end_point in end_points:
|
|
end_point.assign(record)
|
|
except KeyError:
|
|
pass
|
|
|
|
for end_point in end_points:
|
|
end_point.remove_dups()
|
|
print(end_point)
|
|
print(end_point.timestamps_str())
|
|
|
|
d = convert('2024-10-17T18:37:17.067Z')
|