291 lines
7.0 KiB
Python
291 lines
7.0 KiB
Python
|
|
import os
|
|
import json
|
|
import numpy as np
|
|
from datetime import datetime, timedelta
|
|
import dateutil.parser as dup
|
|
from jay_diff import jay_merge_full
|
|
from matplotlib import pyplot as plot
|
|
import eval_records_settings as settings
|
|
|
|
|
|
def has_diff_entries(diff):
|
|
handled = False
|
|
if "update" in diff:
|
|
handled = True
|
|
if "update_add" in diff:
|
|
handled = True
|
|
if "add" in diff:
|
|
handled = True
|
|
if "delete" in diff:
|
|
handled = True
|
|
|
|
return handled
|
|
|
|
|
|
def convert(_filename: str) -> float:
|
|
name = _filename
|
|
if "records_" in _filename:
|
|
name = os.path.splitext(_filename)[0].split("_")[1]
|
|
date_str = name.split("T")[0]
|
|
time_str = name.split("T")[1]
|
|
if "-" in time_str:
|
|
time_str = time_str.replace("-", ":")
|
|
|
|
timestamp_str = f"{date_str}T{time_str}"
|
|
ts = dup.parse(timestamp_str)
|
|
return datetime_to_int(ts)
|
|
|
|
|
|
def datetime_to_int(_dt: datetime) -> float:
|
|
result = datetime.timestamp(_dt)
|
|
return result
|
|
|
|
|
|
class Endpoint:
|
|
def __init__(self, _name: str, _ep_spec: dict, _ts_format='%Y-%m-%dT%H:%M:%S.%f%z'):
|
|
self._name = _name
|
|
self._ep_path = _ep_spec["values"]
|
|
self._ep_path_keys = self.to_keys(self._ep_path)
|
|
self._timestamps = []
|
|
self._values = []
|
|
|
|
self._ts_format = _ts_format
|
|
self._ts_path_keys = self.to_keys(_ep_spec["timestamps"])
|
|
|
|
def timestamps_str(self) -> str:
|
|
return f"{self._ep_path}:{len(self._timestamps)}: {self._timestamps}"
|
|
|
|
@property
|
|
def name(self):
|
|
return self._name
|
|
|
|
@property
|
|
def timestamps(self) -> list[int]:
|
|
return self._timestamps
|
|
|
|
@property
|
|
def values(self) -> list[float]:
|
|
return self._values
|
|
|
|
@property
|
|
def path(self) -> str:
|
|
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._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
|
|
|
|
|
|
def list_folder_ymd(_path: str):
|
|
day_list = sorted(os.listdir(_path), key=lambda f: int(f))
|
|
return day_list
|
|
|
|
|
|
def check_sel(root_path, sel: list):
|
|
res = []
|
|
dir_entries = list_folder_ymd(root_path)
|
|
if sel is None:
|
|
res = dir_entries
|
|
else:
|
|
if isinstance(sel, tuple):
|
|
if len(sel) == 2:
|
|
res = [str(e) for e in range(sel[0], sel[1]+1, 1)]
|
|
else:
|
|
res = [str(e) for e in dir_entries if int(e) >= int(sel[0])]
|
|
|
|
return res
|
|
|
|
|
|
def process(end_points: list[Endpoint], base_path: str, user: str, vin: str, sel_years=None, sel_months=None, sel_days=None, sel_hours=None):
|
|
_path = os.path.join(base_path, user)
|
|
root_path = os.path.join(_path, vin)
|
|
|
|
for _year in check_sel(root_path, sel_years):
|
|
year_path = os.path.join(root_path, _year)
|
|
if not os.path.exists(year_path):
|
|
continue
|
|
for _month in check_sel(year_path, sel_months):
|
|
month_path = os.path.join(year_path, _month)
|
|
if not os.path.exists(month_path):
|
|
continue
|
|
for _day in check_sel(month_path, sel_days):
|
|
day_path = os.path.join(month_path, _day)
|
|
if not os.path.exists(day_path):
|
|
continue
|
|
|
|
# Sort out files with unknown nickname
|
|
file_list_sorted = sorted(os.listdir(day_path), key=lambda f: convert(f))
|
|
vehicle_diff = {}
|
|
last = {}
|
|
for filename in file_list_sorted:
|
|
file_path = os.path.join(day_path, filename)
|
|
with open(file_path, "r") as fp:
|
|
records = json.load(fp)
|
|
print(f"In file \"{filename}\": found {len(records):4d} records")
|
|
for record in records:
|
|
try:
|
|
vehicle_diff = record['data']
|
|
is_header = True
|
|
if 'is_delta' in record['info']:
|
|
if record['info']['is_delta']:
|
|
is_header = False
|
|
|
|
if is_header:
|
|
last = {}
|
|
|
|
except KeyError as e:
|
|
print(f"KeyError: {e}")
|
|
|
|
if not has_diff_entries(vehicle_diff):
|
|
vehicle_diff = {'update_add': vehicle_diff}
|
|
|
|
vehicle_data = jay_merge_full(last, vehicle_diff)
|
|
last = vehicle_data
|
|
try:
|
|
for ep in end_points:
|
|
ep.assign({'info': record['info'], 'data': vehicle_data})
|
|
except KeyError:
|
|
pass
|
|
|
|
for ep in end_points:
|
|
ep.remove_dups()
|
|
|
|
|
|
def find_ep_by_name(end_points: list[Endpoint], name: str):
|
|
for ep in end_points:
|
|
if name in ep.name:
|
|
return ep
|
|
|
|
return None
|
|
|
|
|
|
def fast_forward(iter_t: iter, iter_v: iter, now):
|
|
try:
|
|
while True:
|
|
t = next(iter_t)
|
|
v = next(iter_v)
|
|
if t >= now:
|
|
return t, v
|
|
except StopIteration:
|
|
return None
|
|
|
|
|
|
def resample(_ep: Endpoint, dt_from: datetime, dt_stop: datetime, dt_step: timedelta):
|
|
iter_t = iter(_ep.timestamps)
|
|
iter_v = iter(_ep.values)
|
|
|
|
t = []
|
|
v = []
|
|
_t_next = next(iter_t)
|
|
_v_next = next(iter_v)
|
|
_dt = dt_from
|
|
_ts_stop = datetime_to_int(dt_stop)
|
|
_ts = datetime_to_int(_dt)
|
|
fast_forward(iter_t, iter_v, _ts)
|
|
while _ts < _ts_stop:
|
|
print(f"Time : {_ts}")
|
|
print(f"Value: {_v_next}")
|
|
t.append(_ts)
|
|
v.append(_v_next)
|
|
|
|
try:
|
|
while _ts >= _t_next:
|
|
_t_next = next(iter_t)
|
|
_v_next = next(iter_v)
|
|
except StopIteration:
|
|
pass
|
|
|
|
_dt = _dt + dt_step
|
|
_ts = datetime.timestamp(_dt)
|
|
|
|
return t, v
|
|
|
|
|
|
if __name__ == '__main__':
|
|
end_points = []
|
|
for k in settings.eps.keys():
|
|
end_points.append(Endpoint(k, settings.eps[k]))
|
|
|
|
# Process data and store into endpoints
|
|
process(end_points, settings.BASE, settings.USER, settings.VIN, sel_days=settings.sel_days)
|
|
|
|
# get endpoints
|
|
ep_odo = find_ep_by_name(end_points, name="odo")
|
|
ep_soc = find_ep_by_name(end_points, name="soc")
|
|
ep_range = find_ep_by_name(end_points, name="range")
|
|
|
|
# define observation interval
|
|
dt_start = datetime(2024, 10, 15, 0, 0, 0)
|
|
dt_stop = datetime(2024, 10, 28, 23, 59, 59)
|
|
td_step = timedelta(seconds=5*60)
|
|
|
|
# resample values to equidistant time interval
|
|
t_odo, v_odo = resample(ep_odo, dt_start, dt_stop, td_step)
|
|
t_soc, v_soc = resample(ep_soc, dt_start, dt_stop, td_step)
|
|
t_range, v_range = resample(ep_range, dt_start, dt_stop, td_step)
|
|
|
|
# calc speed
|
|
speed = [0]
|
|
for i in range(1, len(t_odo)):
|
|
dt = float(t_odo[i] - t_odo[i-1])
|
|
dv = v_odo[i] - v_odo[i-1]
|
|
speed.append(dv/dt*60*60)
|
|
|
|
# Convert x-axis
|
|
dt_odo_h = (np.array(t_odo) - t_odo[0])/3600
|
|
dt_soc_h = (np.array(t_soc) - t_soc[0])/3600
|
|
dt_range_h = (np.array(t_range) - t_range[0])/3600
|
|
|
|
# plot data
|
|
plot.subplot(4, 1, 1)
|
|
plot.plot(dt_odo_h, np.array(v_odo) - v_odo[0])
|
|
plot.title("Kilometerstand")
|
|
plot.grid()
|
|
plot.subplot(4, 1, 2)
|
|
plot.plot(dt_odo_h, speed)
|
|
plot.title("Speed")
|
|
plot.grid()
|
|
plot.subplot(4, 1, 3)
|
|
plot.plot(dt_soc_h, v_soc)
|
|
plot.title("Akkustand")
|
|
plot.grid()
|
|
plot.subplot(4, 1, 4)
|
|
plot.plot(dt_range_h, v_range)
|
|
plot.title("Range")
|
|
plot.grid()
|
|
|
|
plot.show()
|