FEAT : Premiers tests

.gitignore

@@ -0,0 +1,2 @@
+venv/
+matter-server-data/

commissionner.py

@@ -0,0 +1,26 @@
+#!python
+import aiohttp
+import asyncio
+from matter_server.client import MatterClient
+from matter_server.client.models.node import MatterNode
+from matter_server.common.models import MatterNodeData
+
+async def commissionner(ws_server_url):
+    # Créer une session aiohttp
+    async with aiohttp.ClientSession() as session:
+        # Initialiser le client Matter avec la session aiohttp
+        async with MatterClient(ws_server_url, session) as client:
+        # client = MatterClient(ws_server_url, session)
+        # client.connect()
+
+          try:
+            mnd:MatterNodeData = await client.commission_with_code("16489714152", True) # TODO coroutine car sinon ne rend pas la main
+            print("ICI !!!!!!!!!!!!!!!!!!!!!!")
+          except Exception as e:
+              print(f"Erreur lors de la récupération des appareils: {e}")
+
+# Remplacez par l'adresse IP de votre python-matter-server
+ws_server_url = "ws://localhost:5580/ws"
+
+# Exécuter la fonction asynchrone
+asyncio.run(commissionner(ws_server_url))

compose.yml

@@ -0,0 +1,12 @@
+version: '3.8'
+
+services:
+  matter-server:
+    image: ghcr.io/home-assistant-libs/python-matter-server:stable
+    container_name: matter-server
+    restart: unless-stopped
+    network_mode: host
+    volumes:
+      - ./matter-server-data:/data
+    environment:
+      - DEBUG=1

lister.py

@@ -0,0 +1,42 @@
+#!python
+import aiohttp
+import asyncio
+from matter_server.client import MatterClient
+from matter_server.client.models.node import MatterNode
+
+async def list_devices(ws_server_url):
+
+  # Créer une session aiohttp
+  async with aiohttp.ClientSession() as session:
+    
+    # Initialiser le client Matter avec la session aiohttp
+    async with MatterClient(ws_server_url, session) as client:
+      client.logger.setLevel("DEBUG")
+      await client.connect()
+
+      print(f"Fabric ID ---> {client.server_info}")
+      # print(f"---------------> {}")
+      
+      print("✅ Connecté à PMS. Attente de synchronisation des nœuds...")
+      # Boucle d'attente simple pendant 5 secondes max
+      # for i in range(30):
+      #     print(f"... {i}")
+      #     if client.get_nodes():
+      #         break
+      #     await asyncio.sleep(1)
+
+      try:
+        # fabrics:list[MatterFabricData] = client.get_matter_fabrics(node_id=)
+        # for fabric in fabrics:
+        #   print(f"Fabric ID: {fabric.fabric_id()}")
+        nodes:list[MatterNode] = client.get_nodes()
+        for node in nodes:
+          print(f"Device ID: {node.node_id()}, Name: {node.name()}")                  
+      except Exception as e:
+        print(f"Erreur lors de la récupération des appareils: {e}")
+
+# Remplacez par l'adresse IP de votre python-matter-server
+ws_server_url = "ws://localhost:5580/ws"
+
+# Exécuter la fonction asynchrone
+asyncio.run(list_devices(ws_server_url))

lister2.py

@@ -0,0 +1,35 @@
+import asyncio
+import aiohttp
+from matter_server.client import MatterClient
+
+async def main():
+    # Crée une session HTTP asynchrone
+    async with aiohttp.ClientSession() as session:
+        client = MatterClient(
+            aiohttp_session=session,
+            ws_server_url="ws://localhost:5580/ws"
+        )
+
+        ready_event = asyncio.Event()
+        await client.connect()
+
+        @client.on("ready")
+        async def on_ready():
+            print("✅ PMS est prêt.")
+            ready_event.set()
+
+        @client.on("node_added")
+        async def on_node_added(node):
+            print(f"🆕 Node ajouté : {node.node_id}, {node.product_name}")
+
+        # await client.connect()
+        await ready_event.wait()
+
+        print("📦 Nœuds déjà connus :")
+        for node in client.nodes:
+            print(f" - ID: {node.node_id}, Produit: {node.product_name}")
+
+        await asyncio.sleep(5)  # Attendre pour voir les nouveaux events
+        await client.disconnect()
+
+asyncio.run(main())

notes.md

@@ -0,0 +1,4 @@
+python3 -m venv venv
+pip install -r requirements.txt
+
+1131-540-8993

onOff.py

@@ -0,0 +1,19 @@
+#!python
+from matter_server.client import MatterClient
+
+def turn_on_shelly_via_matter_client(server_ip, device_id):
+    # Initialiser le client Matter
+    client = MatterClient(server_ip)
+
+    try:
+        # Envoyer la commande pour allumer l'appareil
+        response = client.send_command(device_id, "on")
+        print("Shelly Switch allumé avec succès via matter_server.client.")
+    except Exception as e:
+        print(f"Erreur lors de l'allumage du Shelly Switch: {e}")
+
+# Remplacez par l'adresse IP de votre python-matter-server et l'ID de l'appareil
+server_ip = "localhost"
+device_id = "your_device_id"
+
+turn_on_shelly_via_matter_client(server_ip, device_id)

requirements.txt

@@ -0,0 +1,18 @@
+aenum==3.1.16
+aiohappyeyeballs==2.6.1
+aiohttp==3.11.18
+aiorun==2025.1.1
+aiosignal==1.3.2
+async-timeout==5.0.1
+attrs==25.3.0
+coloredlogs==15.0.1
+dacite==1.9.2
+frozenlist==1.6.0
+home-assistant-chip-clusters==2025.4.0
+humanfriendly==10.0
+idna==3.10
+multidict==6.4.3
+orjson==3.10.18
+propcache==0.3.1
+python-matter-server==8.0.0
+yarl==1.20.0

tetst.py

@@ -0,0 +1,25 @@
+import json
+
+# Import the CHIP clusters
+from chip.clusters import Objects as clusters
+
+# Import the ability to turn objects into dictionaries, and vice-versa
+from matter_server.common.helpers.util import dataclass_from_dict,dataclass_to_dict
+
+command = clusters.OnOff.Commands.On()
+payload = dataclass_to_dict(command)
+
+
+message = {
+    "message_id": "example",
+    "command": "device_command",
+    "args": {
+        "endpoint_id": 1,
+        "node_id": 1,
+        "payload": payload,
+        "cluster_id": command.cluster_id,
+        "command_name": "On"
+    }
+}
+
+print(json.dumps(message, indent=2))