# NikobusConnect
NikobusConnect is a Python library that provides an asynchronous interface for connecting to Nikobus home automation systems via IP or Serial connections. It allows you to control and monitor devices connected to a Nikobus system.
## Table of Contents
- [Features](#features)
- [Installation](#installation)
- [Requirements](#requirements)
- [Usage](#usage)
- [Connecting to Nikobus](#connecting-to-nikobus)
- [Sending Commands](#sending-commands)
- [Receiving Messages](#receiving-messages)
- [Protocol Functions](#protocol-functions)
- [Setting Output State](#setting-output-state)
- [Message Parsing](#message-parsing)
- [API Reference](#api-reference)
- [NikobusConnect](#nikobusconnect-class)
- [NikobusCommandHandler](#nikobuscommandhandler-class)
- [Protocol Functions](#protocol-functions-1)
- [Message Parsing](#message-parsing-function)
- [Examples](#examples)
- [Contributing](#contributing)
- [License](#license)
## Features
- **Asynchronous Communication:** Utilizes `asyncio` for non-blocking I/O operations.
- **Supports IP and Serial Connections:** Connect to Nikobus systems over TCP/IP or via serial ports.
- **Protocol Handling:** Construct and parse Nikobus protocol commands.
- **Command Handling:** Send commands, handle acknowledgments, and manage retries.
- **Message Parsing:** Parse messages received from the Nikobus system.
- **Modular Design:** Easy to integrate into applications like Home Assistant or custom Python scripts.
## Installation
Install the library using pip:
```bash
pip install nikobusconnect
```
## Requirements
- Python 3.7 or higher
- `pyserial-asyncio` package (automatically installed with pip)
## Usage
### Connecting to Nikobus
First, import the `NikobusConnect` class and establish a connection to your Nikobus system.
```python
import asyncio
from nikobusconnect import NikobusConnect
async def main():
connection_string = '192.168.1.100:8000'
nikobus = NikobusConnect(connection_string)
if await nikobus.connect():
print("Connected to Nikobus system")
await nikobus.close()
else:
print("Failed to connect to Nikobus system")
asyncio.run(main())
```
### Sending Commands
Use the `send` method to send commands to the Nikobus system.
Example command:
```python
command = "$1E12A3B400FF110000FFAA3D7BEE"
await nikobus.send(command)
```
### Receiving Messages
Use the `read` method to read data from the Nikobus system. Example received message:
```python
data = await nikobus.read()
message = data.decode('utf-8').strip()
print(f"Received message: {message}") # e.g., "$0515$1FA9C20A003F"
```
### Setting Output State
The `set_output_state` function allows setting the state for different Nikobus modules.
- **Parameters:**
- `address`: The module address as defined in Nikobus software, in a format such as "C9A5".
- `channel`: The channel to control, ranging from 1 to 6 (for shutter modules) and up to 12 for other modules.
- `value`: The state or intensity level.
- **Supported values for different modules:**
- **Switch Module:** `0x00` (Off) or `0x01` (On).
- **Dimmer Module:** Accepts any value between `0x00` (Off) and `0xFF` (Full On).
- **Shutter Module:** Supports:
- `0x00` to stop the cover,
- `0x01` to open, and
- `0x02` to close.
Example usage:
```python
await command_handler.set_output_state(address='C9A5', channel=1, value=0xFF) # Full brightness for dimmer
await command_handler.set_output_state(address='C9A6', channel=2, value=0x00) # Turn off switch
await command_handler.set_output_state(address='C9A7', channel=3, value=0x02) # Close shutter
```
### Message Parsing
Use the `parse_message` function to parse messages from the Nikobus system.
Example received message:
```python
from nikobusconnect import parse_message
message = '$0515$0EFF6C0E0060'
parsed = parse_message(message)
print(parsed)
```
## API Reference
### NikobusConnect Class
Class for managing the connection to the Nikobus system.
- **`NikobusConnect(connection_string: str)`**
Initialize the connection handler with the given connection string.
- **`async connect() -> bool`**
Connect to the Nikobus system.
- **`async send(command: str)`**
Send a command to the Nikobus system.
- **`async read()`**
Read data from the Nikobus system.
- **`async close()`**
Close the connection to the Nikobus system.
### NikobusCommandHandler Class
Class for handling commands to the Nikobus system.
- **`NikobusCommandHandler(nikobus_connection: NikobusConnect)`**
Initialize the command handler.
- **`async get_output_state(address: str, group: int) -> Optional[str]`**
Get the output state of a module.
- **`async set_output_state(address: str, channel: int, value: int)`**
Set the output state of a module.
### Protocol Functions
Functions for constructing and parsing Nikobus protocol commands.
- **`make_pc_link_command(func: int, addr: str, args: bytes = None) -> str`**
Construct a PC link command.
- **`calculate_group_number(channel: int) -> int`**
Calculate the group number of a channel.
### Message Parsing Function
- **`parse_message(message: str) -> dict`**
Parse a Nikobus message and return its components.
## Examples
### Full Example
```python
import asyncio
from nikobusconnect import NikobusConnect, NikobusCommandHandler, parse_message
async def main():
connection_string = '/dev/ttyUSB0'
nikobus = NikobusConnect(connection_string)
if await nikobus.connect():
print("Connected to Nikobus system")
command_handler = NikobusCommandHandler(nikobus)
await command_handler.set_output_state(address='C9A5', channel=1, value=0xFF)
state = await command_handler.get_output_state(address='C9A5', group=1)
print(f"Module state: {state}")
data = await nikobus.read()
message = data.decode('utf-8').strip()
parsed_message = parse_message(message)
print(f"Parsed message: {parsed_message}")
await nikobus.close()
else:
print("Failed to connect to Nikobus system")
asyncio.run(main())
```
## Contributing
Contributions are welcome!
## License
This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
---
**Note:** Replace placeholder values with actual values relevant to your Nikobus system.
For questions or support, please open an issue on the [GitHub repository](https://github.com/fdebrus/nikobusconnect).
Raw data
{
"_id": null,
"home_page": "https://github.com/fdebrus/nikobusconnect",
"name": "nikobusconnect",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.7",
"maintainer_email": null,
"keywords": null,
"author": "Frederic Debrus",
"author_email": "fdebrus@hotmail.com",
"download_url": "https://files.pythonhosted.org/packages/10/a2/07a7292cdce06c5b0cb2d68d0dda2e4336f0e8f9e2b9ceaa7757875572a1/nikobusconnect-0.1.2.tar.gz",
"platform": null,
"description": "\n# NikobusConnect\n\nNikobusConnect is a Python library that provides an asynchronous interface for connecting to Nikobus home automation systems via IP or Serial connections. It allows you to control and monitor devices connected to a Nikobus system.\n\n## Table of Contents\n\n- [Features](#features)\n- [Installation](#installation)\n- [Requirements](#requirements)\n- [Usage](#usage)\n - [Connecting to Nikobus](#connecting-to-nikobus)\n - [Sending Commands](#sending-commands)\n - [Receiving Messages](#receiving-messages)\n - [Protocol Functions](#protocol-functions)\n - [Setting Output State](#setting-output-state)\n - [Message Parsing](#message-parsing)\n- [API Reference](#api-reference)\n - [NikobusConnect](#nikobusconnect-class)\n - [NikobusCommandHandler](#nikobuscommandhandler-class)\n - [Protocol Functions](#protocol-functions-1)\n - [Message Parsing](#message-parsing-function)\n- [Examples](#examples)\n- [Contributing](#contributing)\n- [License](#license)\n\n## Features\n\n- **Asynchronous Communication:** Utilizes `asyncio` for non-blocking I/O operations.\n- **Supports IP and Serial Connections:** Connect to Nikobus systems over TCP/IP or via serial ports.\n- **Protocol Handling:** Construct and parse Nikobus protocol commands.\n- **Command Handling:** Send commands, handle acknowledgments, and manage retries.\n- **Message Parsing:** Parse messages received from the Nikobus system.\n- **Modular Design:** Easy to integrate into applications like Home Assistant or custom Python scripts.\n\n## Installation\n\nInstall the library using pip:\n\n```bash\npip install nikobusconnect\n```\n\n## Requirements\n\n- Python 3.7 or higher\n- `pyserial-asyncio` package (automatically installed with pip)\n\n## Usage\n\n### Connecting to Nikobus\n\nFirst, import the `NikobusConnect` class and establish a connection to your Nikobus system.\n\n```python\nimport asyncio\nfrom nikobusconnect import NikobusConnect\n\nasync def main():\n connection_string = '192.168.1.100:8000'\n nikobus = NikobusConnect(connection_string)\n\n if await nikobus.connect():\n print(\"Connected to Nikobus system\")\n await nikobus.close()\n else:\n print(\"Failed to connect to Nikobus system\")\n\nasyncio.run(main())\n```\n\n### Sending Commands\n\nUse the `send` method to send commands to the Nikobus system.\n\nExample command:\n\n```python\ncommand = \"$1E12A3B400FF110000FFAA3D7BEE\"\nawait nikobus.send(command)\n```\n\n### Receiving Messages\n\nUse the `read` method to read data from the Nikobus system. Example received message:\n\n```python\ndata = await nikobus.read()\nmessage = data.decode('utf-8').strip()\nprint(f\"Received message: {message}\") # e.g., \"$0515$1FA9C20A003F\"\n```\n\n### Setting Output State\n\nThe `set_output_state` function allows setting the state for different Nikobus modules.\n\n- **Parameters:**\n - `address`: The module address as defined in Nikobus software, in a format such as \"C9A5\".\n - `channel`: The channel to control, ranging from 1 to 6 (for shutter modules) and up to 12 for other modules.\n - `value`: The state or intensity level.\n\n- **Supported values for different modules:**\n - **Switch Module:** `0x00` (Off) or `0x01` (On).\n - **Dimmer Module:** Accepts any value between `0x00` (Off) and `0xFF` (Full On).\n - **Shutter Module:** Supports:\n - `0x00` to stop the cover,\n - `0x01` to open, and\n - `0x02` to close.\n\nExample usage:\n\n```python\nawait command_handler.set_output_state(address='C9A5', channel=1, value=0xFF) # Full brightness for dimmer\nawait command_handler.set_output_state(address='C9A6', channel=2, value=0x00) # Turn off switch\nawait command_handler.set_output_state(address='C9A7', channel=3, value=0x02) # Close shutter\n```\n\n### Message Parsing\n\nUse the `parse_message` function to parse messages from the Nikobus system.\n\nExample received message:\n\n```python\nfrom nikobusconnect import parse_message\n\nmessage = '$0515$0EFF6C0E0060'\nparsed = parse_message(message)\nprint(parsed)\n```\n\n## API Reference\n\n### NikobusConnect Class\n\nClass for managing the connection to the Nikobus system.\n\n- **`NikobusConnect(connection_string: str)`**\n\n Initialize the connection handler with the given connection string.\n\n- **`async connect() -> bool`**\n\n Connect to the Nikobus system.\n\n- **`async send(command: str)`**\n\n Send a command to the Nikobus system.\n\n- **`async read()`**\n\n Read data from the Nikobus system.\n\n- **`async close()`**\n\n Close the connection to the Nikobus system.\n\n### NikobusCommandHandler Class\n\nClass for handling commands to the Nikobus system.\n\n- **`NikobusCommandHandler(nikobus_connection: NikobusConnect)`**\n\n Initialize the command handler.\n\n- **`async get_output_state(address: str, group: int) -> Optional[str]`**\n\n Get the output state of a module.\n\n- **`async set_output_state(address: str, channel: int, value: int)`**\n\n Set the output state of a module.\n\n### Protocol Functions\n\nFunctions for constructing and parsing Nikobus protocol commands.\n\n- **`make_pc_link_command(func: int, addr: str, args: bytes = None) -> str`**\n\n Construct a PC link command.\n\n- **`calculate_group_number(channel: int) -> int`**\n\n Calculate the group number of a channel.\n\n### Message Parsing Function\n\n- **`parse_message(message: str) -> dict`**\n\n Parse a Nikobus message and return its components.\n\n## Examples\n\n### Full Example\n\n```python\nimport asyncio\nfrom nikobusconnect import NikobusConnect, NikobusCommandHandler, parse_message\n\nasync def main():\n connection_string = '/dev/ttyUSB0'\n nikobus = NikobusConnect(connection_string)\n\n if await nikobus.connect():\n print(\"Connected to Nikobus system\")\n\n command_handler = NikobusCommandHandler(nikobus)\n\n await command_handler.set_output_state(address='C9A5', channel=1, value=0xFF)\n\n state = await command_handler.get_output_state(address='C9A5', group=1)\n print(f\"Module state: {state}\")\n\n data = await nikobus.read()\n message = data.decode('utf-8').strip()\n parsed_message = parse_message(message)\n print(f\"Parsed message: {parsed_message}\")\n\n await nikobus.close()\n else:\n print(\"Failed to connect to Nikobus system\")\n\nasyncio.run(main())\n```\n\n## Contributing\n\nContributions are welcome! \n\n## License\n\nThis project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.\n\n---\n\n**Note:** Replace placeholder values with actual values relevant to your Nikobus system.\n\nFor questions or support, please open an issue on the [GitHub repository](https://github.com/fdebrus/nikobusconnect).\n",
"bugtrack_url": null,
"license": null,
"summary": "A Python library for connecting to Nikobus systems",
"version": "0.1.2",
"project_urls": {
"Homepage": "https://github.com/fdebrus/nikobusconnect"
},
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "71a8083a229da6479b2ad76babbed0edb7ab22678c88ae5e609b1a3f85b303e5",
"md5": "ce8260414730586beb4342cddce5d7c1",
"sha256": "0391d037842c7ac340fcd37e413568d11ae7ffac606d30ca9ccbce459c9e81a0"
},
"downloads": -1,
"filename": "nikobusconnect-0.1.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "ce8260414730586beb4342cddce5d7c1",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.7",
"size": 9085,
"upload_time": "2024-11-06T15:23:44",
"upload_time_iso_8601": "2024-11-06T15:23:44.318272Z",
"url": "https://files.pythonhosted.org/packages/71/a8/083a229da6479b2ad76babbed0edb7ab22678c88ae5e609b1a3f85b303e5/nikobusconnect-0.1.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "10a207a7292cdce06c5b0cb2d68d0dda2e4336f0e8f9e2b9ceaa7757875572a1",
"md5": "e0624375402eee82c03000683b052162",
"sha256": "f2a93ebe03e3ae9ddca61b4a6306c6cc26f094dc0654813babc9c8870e11494c"
},
"downloads": -1,
"filename": "nikobusconnect-0.1.2.tar.gz",
"has_sig": false,
"md5_digest": "e0624375402eee82c03000683b052162",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.7",
"size": 9518,
"upload_time": "2024-11-06T15:23:45",
"upload_time_iso_8601": "2024-11-06T15:23:45.781015Z",
"url": "https://files.pythonhosted.org/packages/10/a2/07a7292cdce06c5b0cb2d68d0dda2e4336f0e8f9e2b9ceaa7757875572a1/nikobusconnect-0.1.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-11-06 15:23:45",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "fdebrus",
"github_project": "nikobusconnect",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "nikobusconnect"
}
JSON
