capy v0.5.4

import capy

reponse = {
    "ville": "Paris",
    "temperature": "18.5",
    "pluie": "non",
    "jours": [{"nom": "lundi", "max": 21}, {"nom": "mardi", "max": 19}],
}

print(capy.get(reponse, "ville"))                 # Paris
print(capy.get(reponse, "jours.1.nom"))           # mardi  (0 = le 1er)
print(capy.get(reponse, "jours.-1.max"))          # 19     (-1 = le dernier)
print(capy.get(reponse, "vent", default=0))       # 0      (la clé n'existe pas)

print(capy.to_float(reponse, "temperature"))      # 18.5
print(capy.to_bool(reponse, "pluie"))             # False
print(capy.to_str({"ok": True}))                  # {"ok": true}  (dict -> texte JSON)

# Si c'est impossible : ValueError avec un message qui dit quoi corriger
for essai in (lambda: capy.get(reponse, "jours.5"), lambda: capy.to_int(reponse, "ville")):
    try:
        essai()
    except ValueError as e:
        print("Erreur :", e)

import capy

# 1. Personne n'écoute sur ce port -> ConnectionError
try:
    capy.call("localhost", 4999, "bonjour")
except ConnectionError as e:
    print("Pas de réponse :", e)

# 2. Les erreurs côté serveur reviennent dans la réponse
#    (handle() traite une demande sans passer par le réseau : pratique pour tester)
api = capy.Server(4000)


@api.action("diviser")
def diviser(a, b):
    return {"resultat": a / b}


print(api.handle('{"action": "diviser", "a": 6, "b": 3}'))  # {'resultat': 2.0}
print(api.handle('{"action": "diviser", "a": 6, "b": 0}'))  # {'erreur': 'division by zero'}
print(api.handle('{"action": "multiplier"}'))               # {'erreur': 'action inconnue'}
print(api.handle("pas du json"))                            # {'erreur': 'JSON invalide'}

import threading
import time

import capy

api = capy.Server(0)  # 0 : un port libre au hasard
compteur = 0
verrou = threading.Lock()


@api.action("plus_un")
def plus_un():
    global compteur
    with verrou:  # Un seul appel à la fois dans ce bloc
        valeur = compteur
        time.sleep(0.01)  # Simule un petit travail (lire un fichier, calculer…)
        compteur = valeur + 1
    return {"compteur": compteur}


threading.Thread(target=api.run, daemon=True).start()
while not api.listening:  # Attendre que le serveur soit prêt
    time.sleep(0.05)

clients = [threading.Thread(target=capy.call, args=("localhost", api.port, "plus_un")) for _ in range(20)]
for client in clients:
    client.start()
for client in clients:
    client.join()

print("Compteur :", compteur)  # Toujours 20 (sans le verrou : souvent moins !)

import threading
import time

import capy

api = capy.Server(0)


@api.action("carre")
def carre(x):
    return {"resultat": capy.to_float(x) ** 2}


threading.Thread(target=api.run, daemon=True).start()  # daemon : s'arrête avec le programme
while not api.listening:
    time.sleep(0.05)

print("Port choisi :", api.port)
print(capy.call("localhost", api.port, "carre", x=7))     # {'resultat': 49.0}
print(capy.call("localhost", api.port, "carre", x="3"))   # {'resultat': 9.0} ("3" converti)
print(capy.call("localhost", api.port, "carre", x="abc")) # {'erreur': "capy.to_float : impossible de transformer 'abc' (str) en float"}
print(api.handle('{"action": "carre", "x": 5}'))          # {'resultat': 25.0}, sans réseau

import capy.wifi

serveurs = capy.wifi.scan(timeout=1)

if not serveurs:
    print("Aucun serveur capy.wifi trouvé (même wifi ? wifi public qui isole les appareils ?)")

for nom, (ip, port) in sorted(serveurs.items()):
    ici = " (cet appareil)" if ip == "127.0.0.1" else ""
    print(f"{nom:<15} {ip}:{port}{ici}")
# maison          192.168.1.42:51234
# salon           127.0.0.1:50110 (cet appareil)

sudo tee /etc/systemd/system/capy-relay.service > /dev/null <<'SERVICE'
[Unit]
Description=Relais capy.network
After=network-online.target
Wants=network-online.target

[Service]
ExecStart=/usr/bin/python3 -u -m capy.network 3879 MA_CLE_SECRETE
WorkingDirectory=/root
Restart=always
RestartSec=3

[Install]
WantedBy=multi-user.target
SERVICE

sudo systemctl daemon-reload
sudo systemctl enable --now capy-relay   # Lancer maintenant + à chaque démarrage
systemctl status capy-relay              # Il doit être "active (running)"
journalctl -u capy-relay -f              # Voir son journal en direct (Ctrl+C pour sortir)

import capy.domo

home = capy.domo.Home()  # capy.domo.json à côté du script (créé au save())

nouveaux = home.scan()   # ~2 secondes
for appareil in nouveaux:
    print(f"+ {appareil.name:<18} {appareil.brand:<14} {appareil.host or appareil.id}")
for conseil in home.hints:  # Ex : « Pont Hue trouvé à 192.168.1.10 : appaire-le avec … »
    print("!", conseil)

home.save()
print(home)  # <capy.domo.Home 4 appareils, 0 groupes, 0 scènes (…/capy.domo.json)>

import time

import capy.domo

home = capy.domo.Home()
lampe = home["bureau"]  # Le nom donné dans capy.domo.json

lampe.on()
lampe.brightness(30)            # %
time.sleep(1)
lampe.color("orange")           # "orange", "#FF7800" ou (255, 120, 0)
time.sleep(1)
lampe.white("chaud")            # Ou 2700 (kelvins)
time.sleep(1)
lampe.set(brightness=100, color="bleu")

print(lampe.state())  # {'on': True, 'brightness': 100, 'color': '#0000FF', 'kelvin': None, 'reachable': True}
lampe.off()

import capy.domo

home = capy.domo.Home()

home.set_group("salon", ["lampe-du-salon", "bureau"])
home.scenes["film"] = {
    "salon": {"brightness": 15, "color": "orange"},
    "cuisine": {"on": False},
}
home.save()

home["salon"].white("chaud")   # Les deux lampes en même temps
home.scene("film")             # Tout le monde prend son réglage
home["tout"].off()             # « tout » : tous les appareils

home.save_scene("soiree")      # L'état actuel devient une scène
home.save()

import datetime as dt
import sys
import time

import capy.domo

DUREE = 20 * 60  # Le lever de soleil dure 20 minutes
lampe = capy.domo.Home()["chambre"]

heure, minute = (int(x) for x in (sys.argv[1] if len(sys.argv) > 1 else "7:30").split(":"))
reveil = dt.datetime.now().replace(hour=heure, minute=minute, second=0, microsecond=0)
if reveil < dt.datetime.now():
    reveil += dt.timedelta(days=1)
debut = reveil - dt.timedelta(seconds=DUREE)
print(f"Lever de soleil de {debut:%H:%M} à {reveil:%H:%M} (Ctrl+C pour annuler)")
time.sleep(max((debut - dt.datetime.now()).total_seconds(), 0))

for etape in range(1, 61):  # 60 étapes : toutes les 20 secondes
    avance = etape / 60
    if avance < 0.4:
        lampe.set(brightness=1 + avance * 50, color=(255, round(80 + avance * 200), 0))  # Rouge-orangé qui monte
    else:
        lampe.set(brightness=avance * 100, kelvin=round(2200 + avance * 3800))         # Puis blanc de plus en plus froid
    time.sleep(DUREE / 60)
print("Debout ! ☀️")