Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Zigpy application examples #1087

Open
sergorl opened this issue Oct 31, 2022 · 8 comments
Open

Zigpy application examples #1087

sergorl opened this issue Oct 31, 2022 · 8 comments
Labels
documentation Improvements or additions to documentation

Comments

@sergorl
Copy link

sergorl commented Oct 31, 2022

Hi, guys. Are there some examples like?:

  1. Device enumeration: list all connected devices to my zigbee-adapter
  2. Data reading from devices
  3. Device command sending
@13g10n
Copy link

13g10n commented Nov 2, 2022

+1 here

It will be extra useful to get some examples of basic flow when using as separate library (without HA and/or even quirks).

@Hedda
Copy link
Contributor

Hedda commented Nov 17, 2022

Unfortunatly, no one has written application examples or documentation than what is in zigpy wiki and CONTRIBUTING.md file:

https://github.com/zigpy/zigpy/wiki

https://github.com/zigpy/zigpy/blob/dev/CONTRIBUTING.md

Currently, best zigpy use reference is instead Home Assistant's zha component integration as Zigbee gateway implementation:

https://github.com/home-assistant/core/tree/dev/homeassistant/components/zha

Perhaps some independent developers are willing to extend/improve the zigpy API documentation in CONTRIBUTING.md via PRs?

https://github.com/zigpy/zigpy/blob/dev/CONTRIBUTING.md

zigpy also have the zigpy wiki which is a wiki so anyone can add or change to contribute informaton and examples:

https://github.com/zigpy/zigpy/wiki

Anyway, zigpy libraries are for low-level Zigbee use so you really need to read up on ZCL (Zigbee Cluster Library) and the Zigbee protocol as a primer in order to get deeper knowledge about how Zigbee works in order to develop new applications with zigpy.

Currently the zigpy API is only documented here -> https://github.com/zigpy/zigpy/blob/dev/CONTRIBUTING.md#the-zigpy-api

Unfortunately, no one has yet written a guide on how to make an application with zigpy or provided examples for all use cases.

This is a free and open source project that is maintained by volunteers in their spare time, so you can actually help here too!

If you post a direct specific and narrow question then I am sure that some zigpy developer(s) can try to help give a code example.

Note! Be aware that many tools from zigpy radio libraries are deprecated as have instead been moved to zigpy-cli instead:

https://github.com/zigpy/zigpy-cli

Perhaps with help from independent developers like yourself, zigpy-cli could someday become a high-level tool, but needs to be extended with more commands (so submit patches to zigpy-cli to extend its functionality) -> https://github.com/zigpy/zigpy-cli

Also, you need to go to application and command examples previously posted as replies to direct questions -> #1087

Today there are only larger/full Zigbee gateway implementations such as ZHA integration component for Home Assistant, the Zigbee Plugin for Domoticz, and the Zigbee Plugin for Jeedom (competing open-source home automation software, where in the case of the Zigbee Plugin for Jeedom you, unfortunately, need to buy and install Jeedom and that plugin to get its code).

Also check out zha-toolkit -> https://github.com/mdeweerd/zha-toolkit/ it however depends on the zha component integration services -> https://www.home-assistant.io/integrations/zha#services

As for zigpy API documentation there is unfortunately only some basic info available here:

https://github.com/zigpy/zigpy/blob/dev/CONTRIBUTING.md

(And the ZHA Device Handlers /Quirks documentation -> https://github.com/zigpy/zha-device-handlers/edit/dev/README.md ).

Also see this pull request "Documenting zigpy API a kind of developer guide" that was only partially copied to above document:

#477

and

#470

Maybe you could improve API documation and try to submit another pull request to update API docs in CONTRIBUTING.md or post info in wiki? -> https://github.com/zigpy/zigpy/wiki

(Perhaps for additional inspiration also look at https://github.com/Koenkk/zigbee-herdsman#api-documentation ?).

There no other docs but also check through issues and discussions as some pointers have been given to others who asked, as ex:

#709

zigpy/zigpy-znp#108

zigpy/zigpy-znp#77

zigpy/zigpy-znp#152

zigpy/zigpy-znp#25

zigpy/zigpy-znp#166

zigpy/zigpy-znp#192

#452

#469

#470

#471

#715

#7

zigpy/zigpy-znp#166

#595

#709

#865

zigpy/bellows#431

as well as existing articles in the zigpy wiki:

https://github.com/zigpy/zigpy/wiki

Other than using the ZHA integration component code for Home Assistant as reference might also want to look at the code of the Zigbee Plugin for Domoticz and the Zigbee Plugin for Jeedom (competing open-source home automation software) which both partially implements zigpy libraries as dependencies:

https://github.com/zigbeefordomoticz/Domoticz-Zigbee/ (dev https://github.com/zigbeefordomoticz/Domoticz-Zigbee/tree/dev).

Again, also check out zha-toolkit -> https://github.com/mdeweerd/zha-toolkit/ it however depends on the zha component integration services -> https://www.home-assistant.io/integrations/zha#services

PS: zigpy developers also generally recommend read primer on Zigbee / ZCL is also needed as prior knowledge and reference:

PPS: In addition, you can also reference third-party and manufacturer-specific documentation:

@Hedda
Copy link
Contributor

Hedda commented Nov 17, 2022

Oh, by the way, other than zigpy-cli be sure to also check out the code for zha-toolkit for low-level Zigbee commands using zigpy:

https://github.com/zigpy/zigpy-cli

https://github.com/mdeweerd/zha-toolkit/

FYI, dmulcahey also began working on a web socket client–server model design but it looks like he at least paused that it for now:

#1207

https://github.com/zigpy/zha-websocket-server/

https://github.com/zigpy/zhaws-addon

https://github.com/dmulcahey/home-assistant/tree/dm/zha-ws

https://github.com/dmulcahey/home-assistant/tree/dm/zha-ws/homeassistant/components/zhaws

Believe needs both zhaws-addon + zhaws from fork of Home Assistant core with ZHA WS implementation as custom component.

@Hedda
Copy link
Contributor

Hedda commented May 19, 2023

Again, perhaps some independent developers be willing to extend/improve the zigpy API documentation in CONTRIBUTING.md?

https://github.com/zigpy/zigpy/blob/dev/CONTRIBUTING.md#the-zigpy-api

Also see this related suggestion/idea of adding some kind of Automatic API Documentation Generation for zigpy Python libraries:

#1105

If need inspiration suggest to check out matter.js and com.zsmartsystems.zigbee as a few examples of summarizing API concepts:

https://github.com/project-chip/matter.js/blob/main/packages/matter.js/API.md

https://github.com/zsmartsystems/com.zsmartsystems.zigbee/blob/master/README.md

@garrethcain
Copy link

I'd be willing to write some stand-alone examples (not HA related) and even get documentation started if I could get some help to ramp up my understanding...

@Hedda
Copy link
Contributor

Hedda commented Jan 12, 2024

I'd be willing to write some stand-alone examples (not HA related) and even get documentation started if I could get some help to ramp up my understanding...

@garrethcain FYI, probably not up to date but @pipiche38 did start to write a zigpy_api.md a few years ago, check out:

#477

#471

IMHO best approach to begin today might be to just start writing one or more short stand-alone examples as a small "draft" pull request in a new section for the CONTRIBUTING.md file as then you can get feedback to that pull reuqest from zigpy developers:

https://github.com/zigpy/zigpy/blob/dev/CONTRIBUTING.md

Then once you get started with a format that is acceptable to existing zigpy developers then you could submit new pull requests.

@Hedda
Copy link
Contributor

Hedda commented Aug 14, 2024

This specific issue/tracker and the application-level questions in it should now maybe/probably be moved to the repository for the new zha (zha gateway) library:

The zigpy library is a low-level library, while the new zha library is a high-level library that is event based and has a simpler API.

Readme is still pending:

Packages of tagged versions are also released via the "zha" project on PyPI

This new zha library still depends on the zigpy project + other libraries from the zigpy organization on GitHub:

The new zha library is meant to be used by application-level implementations such as the Zigbee Home Automation integration) in Home Assistant, howevers others (like yourselves) could potentially also use it to create stand-alone Zigbee Gateway applications or externally by other types of Zigbee host applications.

That new zha library is actually parts of Home Assistant's ZHA component that have been split/broken-out as an stand-alone library, which is now what the ZHA integration in Home Assistant now depends on to use as an external Zigbee Gateway library so that code there could potentially be used as a reference implementation:

@TheJulianJES TheJulianJES added the documentation Improvements or additions to documentation label Nov 25, 2024
@DataBeaver
Copy link

I wrote this during my adventures of getting zigpy to do stuff. Perhaps it could be useful as an example program. I can create a PR if you give some guidance of where the file should go. Feedback on how to improve it is also welcome.

import asyncio
import logging
import zigpy

# Change bellows to a different module if your radio does not use a
# Silicon Labs chip
from bellows.zigbee.application import ControllerApplication

# Some shortcut names for convenience
from zigpy.zcl.clusters.general import OnOff as OnOffCluster
from zigpy.zcl.clusters.general import LevelControl as LevelControlCluster

# Change these to suit your system
database_path = "zigbee.db"
device_path = "/dev/ttyUSB1"

# Note: You will need to get your zigbee devices into pairing mode in order
# for them to join the network.  The method for that varies by device.  The
# Zigbee2MQTT device database has useful information, for example:
#   https://www.zigbee2mqtt.io/devices/LED2201G8.html

class MainListener:
	"""Monitors new devices and adds them to the light switch"""

	def __init__(self, switch):
		self.switch = switch

	def device_initialized(self, dev):
		"""Called when a device is fully initialized and all of its endpoints
		and clusters are available for inspection."""

		self.switch.process_device(dev)

class OnOffListener:
	"""Listens to on/off commands from button devices and forwards them to
	the light switch object"""

	def __init__(self, switch):
		self.switch = switch

	def cluster_command(self, tsn, cmd, args):
		cmd_defs = OnOffCluster.ServerCommandDefs
		if cmd==cmd_defs.off.id:
			asyncio.create_task(self.switch.turn_off())
		elif cmd==cmd_defs.on.id:
			asyncio.create_task(self.switch.turn_on())

class LevelListener:
	"""Listens to level control commands from control devices and forwards them
	to the light switch object"""

	def __init__(self, switch):
		self.switch = switch

	def cluster_command(self, tsn, cmd, args):
		cmd_defs = LevelControlCluster.ServerCommandDefs
		if cmd==cmd_defs.move.id or cmd==cmd_defs.move_with_on_off.id:
			asyncio.create_task(self.switch.adjust_level(args.move_mode, args.rate, cmd==cmd_defs.move_with_on_off.id))
		elif cmd==cmd_defs.stop.id or cmd==cmd_defs.stop_with_on_off.id:
			asyncio.create_task(self.switch.stop_level(cmd==cmd_defs.stop_with_on_off.id))

class LightSwitch:
	"""Controls lights (or other devices with on/off and level controls) based
	on commands from control devices"""

	def __init__(self):
		self.onoff_clusters = []
		self.onoff_listener = OnOffListener(self)

		self.level_clusters = []
		self.level_listener = LevelListener(self)

		# 254 is the maximum level value allowed by the Zigbee spec
		self.current_level = 254

	def process_device(self, dev):
		"""Enumerates the endpoints and clusters on a device to see if it has
		anything interesting for us"""

		print("Got device {} {} {}".format(dev.name, dev.manufacturer, dev.model))
		for ep in dev.endpoints.values():
			# The endpoints dict also contains the ZDO object which has a
			# completely different interface
			if type(ep)!=zigpy.endpoint.Endpoint:
				continue

			# In clusters are inputs for the device so we can send commands
			# to them
			for clus in ep.in_clusters.values():
				if clus.cluster_id==OnOffCluster.cluster_id:
					print("  Found on/off control")
					self.onoff_clusters.append(clus)
				elif clus.cluster_id==LevelControlCluster.cluster_id:
					print("  Found level control")
					self.level_clusters.append(clus)
					asyncio.create_task(self.read_level(clus))

			# Out clusters are outputs from the device and provide events we
			# can listen to
			for clus in ep.out_clusters.values():
				if clus.cluster_id==OnOffCluster.cluster_id:
					print("  Listening to on/off events")
					clus.add_listener(self.onoff_listener)
				elif clus.cluster_id==LevelControlCluster.cluster_id:
					print("  Listening to level events")
					clus.add_listener(self.level_listener)

	async def read_level(self, clus):
		success, failure = await clus.read_attributes(["current_level"])
		level = success.get("current_level")
		if level is not None:
			self.current_level = level
			print("Current level is {}".format(self.current_level))

	async def turn_off(self):
		print("Turning off")
		for clus in self.onoff_clusters:
			await clus.off()

	async def turn_on(self):
		print("Turning on")
		for clus in self.onoff_clusters:
			await clus.on()

	async def adjust_level(self, mode, rate, with_onoff):
		print("Adjusting level {} with rate {}".format(mode._name_, rate))
		for clus in self.level_clusters:
			if with_onoff:
				await clus.move_with_on_off(mode, rate)
			else:
				await clus.move(mode, rate)

	async def stop_level(self, with_onoff):
		print("Stopping level adjustment")
		for clus in self.level_clusters:
			if with_onoff:
				await clus.stop_with_on_off()
			else:
				await clus.stop()
		for clus in self.level_clusters:
			await self.read_level(clus)

async def main():
	# Create the controller application.  This will automatically connect to the
	# radio and create a network.  Known devices are stored in the database so
	# they can be used on subsequence runs of the program.
	app = await ControllerApplication.new(config={
			"database_path": database_path,
			"device": {
				"path": device_path
			}
		}, auto_form=True)
	try:
		# Create and add the main listener
		switch = LightSwitch()
		app.add_listener(MainListener(switch))

		# Immediately process any existing devices
		for dev in app.devices.values():
			switch.process_device(dev)

		# Allow new devices to join the network for a minute
		async def permit_join():
			print("Allowing devices to join")
			await app.permit(60)
			await asyncio.sleep(60)
			print("Join period ended")
		asyncio.create_task(permit_join())

		# Run the event loop forever by awaiting a future which will never get
		# a result
		await asyncio.get_running_loop().create_future()
	except asyncio.exceptions.CancelledError:
		# Pressing Ctrl-C will cause a CancelledError to be raised
		pass

	# If the app is not shutdown, the process will hang waiting for a background
	# thread to finish
	await app.shutdown()

# Get some basic information from zigpy printed to the console.  Changing the
# level to DEBUG can be useful to discover what kinds of events your control
# devices send but will produce a lot of spam.
logging.basicConfig(level=logging.INFO)

# Start the application
asyncio.run(main())

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
documentation Improvements or additions to documentation
Projects
None yet
Development

No branches or pull requests

6 participants