pumpe2.py

import time
from enum import Enum
from dataclasses import dataclass, field

from rich import get_console
from rich.live import Live
from rich.table import Table

from wled import set_level


console = get_console()


def get_status(step, height, status) -> Table:
    table = Table()
    table.add_row("Step", str(step))
    table.add_row("Height", f"{height:.2f}")
    table.add_row("Status", str(status))
    return table


class Direction(str, Enum):
    a_to_b = "A to B"
    b_to_a = "B to A"


@dataclass
class Height:
    min : int | float
    max : int | float
    step : int | float = field(default=1)

    def __post_init__(self):
        self._value = 50.0
        console.print("Height measurement initialized")
        console.print(f"Min-Height: {self.min} mm")
        console.print(f"Max-Height: {self.max} mm")
        console.print(f"Step: {self.step} mm")
        console.print(f"Value: {self.value:.2f} mm")

    @property
    def value(self) -> int | float:
        return self._value

    @value.setter
    def value(self, value):
        console.print("Setting new height to:", value)
        self._value = value

    @property
    def ok(self) -> bool:
        return self.min <= self.value <= self.max

    @property
    def min_ok(self) -> bool:
        return self.min < self.value

    @property
    def max_ok(self) -> bool:
        return self.max > self.value

    def update(self, direction: Direction | None) -> int | float:
        if direction is Direction.a_to_b:
            self._value -= self.step
        elif direction is Direction.b_to_a:
            self._value += self.step


@dataclass
class Pumpe:
    name : str

    def __post_init__(self):
         self._state = None
         console.print(f"Created {self.name}.")

    @property
    def state(self) -> Direction:
        return self._state

    def a_to_b(self):
        self._state = Direction.a_to_b
        # console.print(f"{self.name}: {self._state.value}")

    def b_to_a(self):
        self._state = Direction.b_to_a
        # console.print(f"{self.name}: {self._state.value}")

    def stop(self):
        self._state = None
        # console.print(f"{self.name}: Stop")



def main(live):
    delay = 0.005
    step_size = (140 / 14) * delay
    console.print(f"Simulation with step_size of {step_size:.2f} and a delay of {delay:.3f}s")

    height = Height(min=15.0, max=100.0, step=step_size)
    height.value = 100.0

    pumpe = Pumpe("Pupe 1")

    console.print("Starting main loop")
    step = 0
    timer = 1.0

    while True:
        time.sleep(delay)

        height.update(pumpe.state)
        # console.print(f"Height: {height.value:.2f}")
        # console.print("\nStep:", step)
        live.update(get_status(step, height.value, pumpe.state))
        set_level(height.value, min_value=0.0, max_value=140.0)

        #  0 [      start  ] -> Pumpe stoppen | ->  1

        #              Verzweigung
        #  1 [      max_ok ] -> Von b nach a  | -> 10
        #  1 [ not max_ok  ] -> Von a nach b  | -> 20

        #              von b nach a
        # 10 [ not max_ok  ] -> Pumpe stoppen | -> 11
        # 11 [ time 4 s    ] ->               | -> 20

        #              von a nach b
        # 20 [             ] -> Von a nach b  | -> 21
        # 21 [  not min_ok ] -> Pumpe stoppen | -> 30

        # 30 [             ] ->               | ->  0

        match step:
            case 0:
                step = 1
                pumpe.stop()
                input("Enter to start: ")
                console.clear()

            case 1 if height.max_ok:
                step = 10
                pumpe.b_to_a()

            case 1 if not height.max_ok:
                step = 20
                pumpe.a_to_b()

            case 10 if not height.max_ok:
                step = 11
                pumpe.stop()

            case 11:
                step = 20
                # console.print(f"Delay of {timer}s")
                time.sleep(timer)

            case 20:
                step = 21
                pumpe.a_to_b()

            case 21 if not height.min_ok:
                step = 30
                pumpe.stop()

            case 30:
                step = 0


if __name__ == "__main__":
    console.print("Starting simulation")
    try:
        with Live(refresh_per_second=60) as live:
            main(live)
    except KeyboardInterrupt:
        print()