| import threading |
| from reachy_mini import ReachyMini, ReachyMiniApp |
| from reachy_mini.utils import create_head_pose |
| import numpy as np |
| import time |
| from pydantic import BaseModel |
|
|
|
|
| frequency = 10 |
| base_sleep_time = 1 / 100 |
|
|
|
|
| class Takina(ReachyMiniApp): |
| |
| |
| custom_app_url: str | None = "http://0.0.0.0:8042" |
| |
| |
| request_media_backend: str | None = None |
|
|
| def run(self, reachy_mini: ReachyMini, stop_event: threading.Event): |
| t0 = time.time() |
|
|
| antennas_enabled = True |
|
|
| |
| |
| class AntennaState(BaseModel): |
| enabled: bool |
|
|
| @self.settings_app.post("/antennas") |
| def update_antennas_state(state: AntennaState): |
| nonlocal antennas_enabled |
| antennas_enabled = state.enabled |
| return {"antennas_enabled": antennas_enabled} |
|
|
| |
| |
| def get_antennas_deg(): |
| if antennas_enabled: |
| amp_deg = 25.0 |
| a = amp_deg * np.sin(2.0 * np.pi * 0.5 * t) |
| return np.array([a, -a]) |
| return np.array([0.0, 0.0]) |
| |
| def init_pos(): |
| reachy_mini.set_target( |
| head = create_head_pose(roll=0, pitch=0, yaw=0, x=0, y=0, z=0, degrees=True, mm=True), |
| antennas = np.array([0.0, 0.0]), |
| ) |
| |
| def progA(): |
| y = 30.0 * np.sin(2.0 * np.pi * 0.4 * t) |
| pitch_deg = 10.0 + 10.0 * np.sin(2.0 * np.pi * 1.6 * t + np.pi) |
| return create_head_pose(pitch=pitch_deg, y=y, degrees=True, mm=True) |
| |
| def progB(): |
| x = 30.0 * np.cos(2.0 * np.pi * 0.5 * t + np.pi/2) |
| y = 30.0 * np.sin(2.0 * np.pi * 0.5 * t + np.pi/2) |
| return create_head_pose(x=x, y=y, mm=True) |
| |
| init_pos() |
|
|
| |
| while not stop_event.is_set(): |
| t = time.time() - t0 |
|
|
| |
| antennas_rad = np.deg2rad(get_antennas_deg()) |
| |
| if (t < 7.5): |
| head_pose = progA() |
| elif(t < (7.5 + 6)): |
| head_pose = progB() |
| else: |
| break |
| |
| |
| |
| reachy_mini.set_target( |
| head=head_pose, |
| antennas=antennas_rad, |
| ) |
| |
| duration_loop = (time.time() - t0) - t |
| sleep_time = base_sleep_time - duration_loop |
| if sleep_time < 0: |
| sleep_time = 0 |
| time.sleep(sleep_time) |
| |
| init_pos() |
|
|
|
|
| if __name__ == "__main__": |
| app = Takina() |
| try: |
| app.wrapped_run() |
| except KeyboardInterrupt: |
| app.stop() |