First implementation of data chart.
This commit is contained in:
parent
fb8883c185
commit
fc25be87e9
87
src/corchestrate/chart.py
Normal file
87
src/corchestrate/chart.py
Normal file
|
|
@ -0,0 +1,87 @@
|
|||
from datetime import datetime
|
||||
from plotly.subplots import make_subplots
|
||||
|
||||
from corchestrate.chart_data import ChartData
|
||||
from corchestrate.indexed_input import IndexedInput
|
||||
from corchestrate.measurement import Measurement
|
||||
|
||||
|
||||
def plot_over_time(input_data: list[ChartData]):
|
||||
indexed_inputs = [
|
||||
IndexedInput(data.name, data.measurements) for data in input_data
|
||||
]
|
||||
|
||||
datetime_indices = []
|
||||
|
||||
fig = make_subplots(5, 1)
|
||||
|
||||
while True:
|
||||
min_datetime: datetime | None = None
|
||||
# Find min_datetime
|
||||
for indexed_input in indexed_inputs:
|
||||
current_measurement = indexed_input.current()
|
||||
|
||||
if current_measurement == None:
|
||||
continue
|
||||
|
||||
if min_datetime == None:
|
||||
min_datetime = current_measurement.measured_at
|
||||
continue
|
||||
|
||||
if current_measurement.measured_at < min_datetime:
|
||||
min_datetime = current_measurement.measured_at
|
||||
|
||||
# If no valid measurement is available, break the loop
|
||||
if min_datetime == None:
|
||||
break
|
||||
|
||||
datetime_indices.append(min_datetime)
|
||||
|
||||
# Extract data at minimum datetime
|
||||
for indexed_input in indexed_inputs:
|
||||
current_measurement = indexed_input.current()
|
||||
if current_measurement is None:
|
||||
continue
|
||||
|
||||
if current_measurement.measured_at == min_datetime:
|
||||
indexed_input.extracted_heart_rate.append(
|
||||
current_measurement.heart_rate)
|
||||
indexed_input.extracted_speed.append(current_measurement.speed)
|
||||
indexed_input.extracted_cadence.append(
|
||||
current_measurement.cadence)
|
||||
indexed_input.extracted_power.append(current_measurement.power)
|
||||
indexed_input.extracted_temperature.append(
|
||||
current_measurement.temperature)
|
||||
# Advance to next index
|
||||
indexed_input.next()
|
||||
|
||||
# Plot
|
||||
for indexed_input in indexed_inputs:
|
||||
fig.add_scatter(x=datetime_indices,
|
||||
y=indexed_input.extracted_heart_rate,
|
||||
row=1,
|
||||
col=1,
|
||||
name=f"Heart rate - {indexed_input.name}")
|
||||
fig.add_scatter(x=datetime_indices,
|
||||
y=indexed_input.extracted_speed,
|
||||
row=2,
|
||||
col=1,
|
||||
name=f"Speed - {indexed_input.name}")
|
||||
fig.add_scatter(x=datetime_indices,
|
||||
y=indexed_input.extracted_cadence,
|
||||
row=3,
|
||||
col=1,
|
||||
name=f"Cadence - {indexed_input.name}")
|
||||
fig.add_scatter(x=datetime_indices,
|
||||
y=indexed_input.extracted_power,
|
||||
row=4,
|
||||
col=1,
|
||||
name=f"Power - {indexed_input.name}")
|
||||
fig.add_scatter(x=datetime_indices,
|
||||
y=indexed_input.extracted_temperature,
|
||||
row=5,
|
||||
col=1,
|
||||
name=f"Temperature - {indexed_input.name}")
|
||||
|
||||
fig.update_layout(height=3000)
|
||||
fig.show()
|
||||
9
src/corchestrate/chart_data.py
Normal file
9
src/corchestrate/chart_data.py
Normal file
|
|
@ -0,0 +1,9 @@
|
|||
from dataclasses import dataclass
|
||||
|
||||
from corchestrate.measurement import Measurement
|
||||
|
||||
|
||||
@dataclass
|
||||
class ChartData:
|
||||
name: str
|
||||
measurements: list[Measurement]
|
||||
29
src/corchestrate/indexed_input.py
Normal file
29
src/corchestrate/indexed_input.py
Normal file
|
|
@ -0,0 +1,29 @@
|
|||
from numpy import double
|
||||
from corchestrate.measurement import Measurement
|
||||
|
||||
|
||||
class IndexedInput:
|
||||
|
||||
def __init__(self, name: str, measurements: list[Measurement]) -> None:
|
||||
self.index: int = 0
|
||||
self.name = name
|
||||
self.measurements: list[Measurement] = measurements
|
||||
|
||||
self.extracted_heart_rate: list[float | None] = []
|
||||
self.extracted_speed: list[float | None] = []
|
||||
self.extracted_cadence: list[float | None] = []
|
||||
self.extracted_power: list[float | None] = []
|
||||
self.extracted_temperature: list[float | None] = []
|
||||
|
||||
def is_out_of_range(self) -> bool:
|
||||
return len(self.measurements) <= self.index
|
||||
|
||||
def current(self):
|
||||
if self.is_out_of_range():
|
||||
return None
|
||||
|
||||
return self.measurements[self.index]
|
||||
|
||||
def next(self):
|
||||
self.index += 1
|
||||
self.current()
|
||||
Loading…
Reference in a new issue