diff --git a/pyftms/__init__.py b/pyftms/__init__.py index 58f068f0ae..10174b6cda 100644 --- a/pyftms/__init__.py +++ b/pyftms/__init__.py @@ -29,7 +29,7 @@ ) from .client.backends import FtmsEvents from .client.machines import CrossTrainer, IndoorBike, Rower, Treadmill -from .models import IndoorBikeSimulationParameters, ResultCode +from .models import IndoorBikeSimulationParameters, ResultCode, TrainingStatusCode __all__ = [ "get_client", @@ -59,4 +59,5 @@ "SettingRange", "ResultCode", "PropertiesManager", + "TrainingStatusCode", ] diff --git a/pyftms/client/backends/updater.py b/pyftms/client/backends/updater.py index 2a69ae1389..4fb02cd8c6 100644 --- a/pyftms/client/backends/updater.py +++ b/pyftms/client/backends/updater.py @@ -2,22 +2,22 @@ # SPDX-License-Identifier: Apache-2.0 import logging -from typing import Any, Generic, cast +from typing import Any, cast from bleak import BleakClient from bleak.backends.characteristic import BleakGATTCharacteristic -from ...models import RealtimeDataT +from ...models import RealtimeData from ...serializer import ModelSerializer, get_serializer from .event import FtmsCallback, UpdateEvent, UpdateEventData _LOGGER = logging.getLogger(__name__) -class DataUpdater(Generic[RealtimeDataT]): - _serializer: ModelSerializer[RealtimeDataT] +class DataUpdater: + _serializer: ModelSerializer[RealtimeData] - def __init__(self, model: type[RealtimeDataT], callback: FtmsCallback) -> None: + def __init__(self, model: type[RealtimeData], callback: FtmsCallback) -> None: self._cb = callback self._serializer = get_serializer(model) self._prev: dict[str, Any] = {} diff --git a/pyftms/client/client.py b/pyftms/client/client.py index 0a5b1162c7..cab5f225d6 100644 --- a/pyftms/client/client.py +++ b/pyftms/client/client.py @@ -3,8 +3,9 @@ import logging from abc import ABC +from functools import cached_property from types import MappingProxyType -from typing import Any, ClassVar, Generic, TypeVar +from typing import Any, ClassVar from bleak import BleakClient from bleak.backends.device import BLEDevice @@ -38,10 +39,8 @@ _LOGGER = logging.getLogger(__name__) -RealtimeDataT = TypeVar("RealtimeDataT", bound=RealtimeData) - -class FitnessMachine(ABC, Generic[RealtimeDataT], PropertiesManager): +class FitnessMachine(ABC, PropertiesManager): """ Base FTMS client. @@ -51,7 +50,7 @@ class FitnessMachine(ABC, Generic[RealtimeDataT], PropertiesManager): _machine_type: ClassVar[MachineType] """Machine type.""" - _data_model: type[RealtimeDataT] + _data_model: type[RealtimeData] """Model of real-time training data.""" _data_uuid: ClassVar[str] @@ -59,7 +58,7 @@ class FitnessMachine(ABC, Generic[RealtimeDataT], PropertiesManager): _cli: BleakClientWithServiceCache | None = None - _data_updater: DataUpdater[RealtimeDataT] + _data_updater: DataUpdater # Static device info @@ -140,7 +139,7 @@ def machine_type(self) -> MachineType: """Machine type.""" return self._machine_type - @property + @cached_property def supported_properties(self) -> tuple[str, ...]: """ Properties that supported by this machine. @@ -151,12 +150,12 @@ def supported_properties(self) -> tuple[str, ...]: """ return self._get_supported_properties(self._m_features) - @property + @cached_property def available_properties(self) -> tuple[str, ...]: """All properties that *MAY BE* supported by this machine type.""" return self._get_supported_properties(MachineFeatures(~0)) - @property + @cached_property def supported_settings(self) -> tuple[str, ...]: """Supported settings.""" return ControlModel._get_features(self._m_settings) @@ -228,15 +227,19 @@ async def _write_command(self, code: ControlCode | None = None, *args, **kwargs) ) async def reset(self) -> ResultCode: + """Initiates the procedure to reset the controllable settings of a fitness machine.""" return await self._write_command(ControlCode.RESET) - async def start(self) -> ResultCode: + async def start_resume(self) -> ResultCode: + """Initiate the procedure to start or resume a training session.""" return await self._write_command(ControlCode.START_RESUME) async def stop(self) -> ResultCode: + """Initiate the procedure to stop a training session.""" return await self._write_command(stop_pause=StopPauseCode.STOP) async def pause(self) -> ResultCode: + """Initiate the procedure to pause a training session.""" return await self._write_command(stop_pause=StopPauseCode.PAUSE) async def set_setting(self, setting_id: str, *args: Any) -> ResultCode: @@ -246,6 +249,9 @@ async def set_setting(self, setting_id: str, *args: Any) -> ResultCode: **Methods for setting specific parameters.** """ + if setting_id not in self.supported_settings: + return ResultCode.NOT_SUPPORTED + if not args: raise ValueError("No data to pass.") @@ -255,63 +261,120 @@ async def set_setting(self, setting_id: str, *args: Any) -> ResultCode: return await self._write_command(code=None, **{setting_id: args}) async def set_target_speed(self, value: float) -> ResultCode: - """""" - return await self._write_command(target_speed=value) + """ + Sets target speed. + + Units: `km/h`. + """ + return await self.set_setting("target_speed", value) async def set_target_inclination(self, value: float) -> ResultCode: - """""" - return await self._write_command(target_inclination=value) + """ + Sets target inclination. + + Units: `%`. + """ + return await self.set_setting("target_inclination", value) async def set_target_resistance(self, value: float) -> ResultCode: - """""" - return await self._write_command(target_resistance=value) + """ + Sets target resistance level. + + Units: `unitless`. + """ + return await self.set_setting("target_resistance", value) async def set_target_power(self, value: int) -> ResultCode: - """""" - return await self._write_command(target_power=value) + """ + Sets target power. + + Units: `Watts`. + """ + return await self.set_setting("target_power", value) async def set_target_heart_rate(self, value: int) -> ResultCode: - """""" - return await self._write_command(target_heart_rate=value) + """ + Sets target heart rate. + + Units: `bpm`. + """ + return await self.set_setting("target_heart_rate", value) async def set_target_energy(self, value: int) -> ResultCode: - """""" - return await self._write_command(target_energy=value) + """ + Sets target expended energy. + + Units: `Kcals`. + """ + return await self.set_setting("target_energy", value) async def set_target_steps(self, value: int) -> ResultCode: - """""" - return await self._write_command(target_steps=value) + """ + Sets targeted number of steps. + + Units: `steps`. + """ + return await self.set_setting("target_steps", value) async def set_target_strides(self, value: int) -> ResultCode: - """""" - return await self._write_command(target_strides=value) + """ + Sets targeted number of strides. + + Units: `strides`. + """ + return await self.set_setting("target_strides", value) async def set_target_distance(self, value: int) -> ResultCode: - """""" - return await self._write_command(target_distance=value) + """ + Sets targeted distance. + + Units: `meters`. + """ + return await self.set_setting("target_distance", value) async def set_target_time(self, *value: int) -> ResultCode: - """""" - return await self._write_command(code=None, target_time=value) + """ + Set targeted training time. + + Units: `seconds`. + """ + return await self.set_setting("target_time", *value) async def set_bike_simulation_params( - self, p: IndoorBikeSimulationParameters + self, + value: IndoorBikeSimulationParameters, ) -> ResultCode: - """""" - return await self._write_command(indoor_bike_simulation=p) + """Set indoor bike simulation parameters.""" + return await self.set_setting("indoor_bike_simulation", value) async def set_wheel_circumference(self, value: float) -> ResultCode: - """""" - return await self._write_command(wheel_circumference=value) + """ + Set wheel circumference. + + Units: `mm`. + """ + return await self.set_setting("wheel_circumference", value) async def spin_down_start(self) -> ResultCode: - """""" - return await self._write_command(spin_down_control=SpinDownControlCode.START) + """ + Start Spin-Down. + + It can be sent either in response to a request to start Spin-Down, or separately. + """ + return await self.set_setting("spin_down_control", SpinDownControlCode.START) async def spin_down_ignore(self) -> ResultCode: - """""" - return await self._write_command(spin_down_control=SpinDownControlCode.IGNORE) + """ + Ignore Spin-Down. + + It can be sent in response to a request to start Spin-Down. + """ + return await self.set_setting("spin_down_control", SpinDownControlCode.IGNORE) async def set_target_cadence(self, value: float) -> ResultCode: - """""" - return await self._write_command(target_cadence=value) + """ + Set targeted cadence. + + Units: `rpm`. + """ + return await self.set_setting("target_cadence", value) diff --git a/pyftms/client/machines/cross_trainer.py b/pyftms/client/machines/cross_trainer.py index 3903fa197c..6d02537b37 100644 --- a/pyftms/client/machines/cross_trainer.py +++ b/pyftms/client/machines/cross_trainer.py @@ -9,7 +9,7 @@ from ..properties import MachineType -class CrossTrainer(FitnessMachine[CrossTrainerData]): +class CrossTrainer(FitnessMachine): """Cross Trainer (Elliptical Trainer)""" _machine_type: ClassVar[MachineType] = MachineType.CROSS_TRAINER diff --git a/pyftms/models/__init__.py b/pyftms/models/__init__.py index 6e32b4a6e3..c53136adaf 100644 --- a/pyftms/models/__init__.py +++ b/pyftms/models/__init__.py @@ -8,7 +8,6 @@ CrossTrainerData, IndoorBikeData, RealtimeData, - RealtimeDataT, RowerData, TreadmillData, ) @@ -40,5 +39,4 @@ "TrainingStatusFlags", "TrainingStatusModel", "RealtimeData", - "RealtimeDataT", ] diff --git a/pyftms/models/realtime_data/__init__.py b/pyftms/models/realtime_data/__init__.py index 3d1ffbbfd1..64633b5bf0 100644 --- a/pyftms/models/realtime_data/__init__.py +++ b/pyftms/models/realtime_data/__init__.py @@ -1,22 +1,16 @@ # Copyright 2024, Sergey Dudanov # SPDX-License-Identifier: Apache-2.0 -import typing - from .common import RealtimeData from .cross_trainer import CrossTrainerData from .indoor_bike import IndoorBikeData from .rower import RowerData from .treadmill import TreadmillData -RealtimeDataT = typing.TypeVar("RealtimeDataT", bound=RealtimeData) - - __all__ = [ "CrossTrainerData", "IndoorBikeData", "RowerData", "TreadmillData", "RealtimeData", - "RealtimeDataT", ]