Source code for leads.dt.odometer

from threading import Lock as _Lock
from typing import override as _override

from leads.dt.device import Device



[docs]
class Odometer(Device):
    def __init__(self) -> None:
        super().__init__()
        self._mileage: float = 0


[docs]
    @_override
    def write(self, payload: float) -> None:
        self._mileage += payload



[docs]
    @_override
    def read(self) -> float:
        return self._mileage





[docs]
class ConcurrentOdometer(Odometer):
    def __init__(self) -> None:
        super().__init__()
        self._lock: _Lock = _Lock()


[docs]
    @_override
    def write(self, payload: float) -> None:
        self._lock.acquire()
        try:
            super().write(payload)
        finally:
            self._lock.release()