File: device_configurations.py

package info (click to toggle)
pyvesync 3.2.1-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 1,824 kB
  • sloc: python: 15,994; makefile: 3
file content (90 lines) | stat: -rw-r--r-- 2,989 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
"""Get configuration details from the API."""

from __future__ import annotations

import logging
from pathlib import Path
from typing import Any

import aiofiles
import orjson

from pyvesync import VeSync
from pyvesync.models.vesync_models import RequestDeviceConfiguration

USERNAME = ''
PASSWORD = ''

vs_logger = logging.getLogger('pyvesync')
vs_logger.setLevel(logging.DEBUG)

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)


def parse_config(data: dict) -> dict[str, list[dict]]:
    """Parse the configuration data from the API into a nested dictionary."""
    result = {}
    # Navigate to productLineList
    data = orjson.loads(orjson.dumps(data))  # Ensure data is a dict
    config_list = data.get('result', {}).get('configList', [])
    for config in config_list:
        for item in config.get('items', []):
            item_value = item.get('itemValue')
            if isinstance(item_value, str):
                item_value = orjson.loads(item_value)

            product_lines = item_value.get('productLineList', [])
            for product_line in product_lines:
                for type_info in product_line.get('typeInfoList', []):
                    type_name = type_info.get('typeName')
                    if not type_name:
                        continue
                    models = []
                    for model_info in type_info.get('modelInfoList', []):
                        model = model_info.get('model')
                        if not model:
                            continue
                        # Add the full model_info dict under the model key
                        models.append({model: [model_info]})
                    if models:
                        result[type_name] = models
    return result


async def fetch_config(manager: VeSync) -> dict[Any, Any] | None:
    """Get full device configuration from the API."""
    endpoint = '/cloud/v1/app/getAppConfigurationV2'
    body = RequestDeviceConfiguration(
        accountID=manager.account_id,
        token=manager.token,
    )
    response, _ = await manager.async_call_api(
        api=endpoint,
        method='post',
        json_object=body,
    )
    return response


async def main() -> None:
    """Main function to fetch and display device configurations."""
    async with VeSync(USERNAME, PASSWORD, 'US') as manager:
        await manager.login()
        config = await fetch_config(manager)
        if not config:
            print('Failed to fetch configuration.')
            return
        parsed_config = parse_config(config)
        print(orjson.dumps(parsed_config, option=orjson.OPT_INDENT_2).decode('utf-8'))
        output_path = Path('models.json')
        async with aiofiles.open(output_path, 'w', encoding='utf-8') as file:
            await file.write(
                orjson.dumps(parsed_config, option=orjson.OPT_INDENT_2).decode('utf-8')
            )


if __name__ == '__main__':
    import asyncio

    asyncio.run(main())