Quick start

We (developers) all know how important is to get things working fast. In this section, you can find some quick-reference recipies that can be used for start working straight forward.

Note

Immediately getting hands dirty with code is OK for quick demos. However, you should really read the entire documentation carefully in order to avoid gotchas. After all, someone has taken the effort of writing, why would you not read it?

Listing devices

Listing Meross devices

import asyncio
import os

from meross_iot.http_api import MerossHttpClient
from meross_iot.manager import MerossManager

EMAIL = os.environ.get('MEROSS_EMAIL') or "YOUR_MEROSS_CLOUD_EMAIL"
PASSWORD = os.environ.get('MEROSS_PASSWORD') or "YOUR_MEROSS_CLOUD_PASSWORD"


async def main():
    # Setup the HTTP client API from user-password
    http_api_client = await MerossHttpClient.async_from_user_password(email=EMAIL, password=PASSWORD, api_base_url="https://iot.meross.com")

    # Setup and start the device manager
    manager = MerossManager(http_client=http_api_client)
    await manager.async_init()

    # Discover devices.
    await manager.async_device_discovery()
    meross_devices = manager.find_devices()

    # Print them
    print("I've found the following devices:")
    for dev in meross_devices:
        print(f"- {dev.name} ({dev.type}): {dev.online_status}")

    # Close the manager and logout from http_api
    manager.close()
    await http_api_client.async_logout()

if __name__ == '__main__':
    if os.name == 'nt':
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
    loop.stop()

In order to discover meross devices, you need to setup a MerossManager. That class is the one in charge of handling mqtt and http calls to respective endpoints. It also keeps track of discovered devices and allows you to discover new ones.

Once the manager has been created and started via async_init(), you can look for devices via the the async_device_discovery() method. This method will query the HTTP API for getting the UUIDs (alongside other info) of the Meross devices registered with your account. To see the latest discovered devices cached by the MerossManager you can use the find_device(). You should not invoke async_device_discovery() too frequently as doing so might trigger some alert/banning from Meross API.

The find_device() method supports some filtering arguments to search for specific devices. Have a look at the following method signature for more details.

Controlling switches

Toggling smart switches

import asyncio
import os
import os
from meross_iot.http_api import MerossHttpClient
from meross_iot.manager import MerossManager

EMAIL = os.environ.get('MEROSS_EMAIL') or "YOUR_MEROSS_CLOUD_EMAIL"
PASSWORD = os.environ.get('MEROSS_PASSWORD') or "YOUR_MEROSS_CLOUD_PASSWORD"


async def main():
    # Setup the HTTP client API from user-password
    http_api_client = await MerossHttpClient.async_from_user_password(email=EMAIL, password=PASSWORD, api_base_url="https://iot.meross.com")

    # Setup and start the device manager
    manager = MerossManager(http_client=http_api_client)
    await manager.async_init()

    # Retrieve all the MSS310 devices that are registered on this account
    await manager.async_device_discovery()
    plugs = manager.find_devices(device_type="mss310")

    if len(plugs) < 1:
        print("No MSS310 plugs found...")
    else:
        # Turn it on channel 0
        # Note that channel argument is optional for MSS310 as they only have one channel
        dev = plugs[0]

        # Update device status: this is needed only the very first time we play with this device (or if the
        #  connection goes down)
        await dev.async_update()

        print(f"Turning on {dev.name}...")
        await dev.async_turn_on(channel=0)
        print("Waiting a bit before turing it off")
        await asyncio.sleep(5)
        print(f"Turing off {dev.name}")
        await dev.async_turn_off(channel=0)

    # Close the manager and logout from http_api
    manager.close()
    await http_api_client.async_logout()

if __name__ == '__main__':
    if os.name == 'nt':
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
    loop.stop()

Meross devices that supports toggling and on/off commands implement either the meross_iot.controller.mixins.toggle.ToggleMixin or meross_iot.controller.mixins.toggle.ToggleXMixin. Both classes expose a number of methods to control these devices, such async_turn_off() and async_turn_on().

Refer to the ToggleXMixn for a panoramic around those.

Controlling bulbs

Operating smart bulbs

import asyncio
import os
from random import randint

from meross_iot.http_api import MerossHttpClient
from meross_iot.manager import MerossManager
from meross_iot.model.enums import OnlineStatus

EMAIL = os.environ.get('MEROSS_EMAIL') or "YOUR_MEROSS_CLOUD_EMAIL"
PASSWORD = os.environ.get('MEROSS_PASSWORD') or "YOUR_MEROSS_CLOUD_PASSWORD"


async def main():
    # Setup the HTTP client API from user-password
    http_api_client = await MerossHttpClient.async_from_user_password(email=EMAIL, password=PASSWORD, api_base_url="https://iot.meross.com")

    # Setup and start the device manager
    manager = MerossManager(http_client=http_api_client)
    await manager.async_init()

    # Retrieve the MSL120 devices that are registered on this account
    await manager.async_device_discovery()
    plugs = manager.find_devices(device_type="msl120", online_status=OnlineStatus.ONLINE)

    if len(plugs) < 1:
        print("No online msl120 smart bulbs found...")
    else:
        # Let's play with RGB colors. Note that not all light devices will support
        # rgb capabilities. For this reason, we first need to check for rgb before issuing
        # color commands.
        dev = plugs[0]

        # Update device status: this is needed only the very first time we play with this device (or if the
        #  connection goes down)
        await dev.async_update()
        if not dev.get_supports_rgb():
            print("Unfortunately, this device does not support RGB...")
        else:
            # Check the current RGB color
            current_color = dev.get_rgb_color()
            print(f"Currently, device {dev.name} is set to color (RGB) = {current_color}")
            # Randomly chose a new color
            rgb = randint(0, 255), randint(0, 255), randint(0, 255)
            print(f"Chosen random color (R,G,B): {rgb}")
            await dev.async_set_light_color(rgb=rgb)
            print("Color changed!")

    # Close the manager and logout from http_api
    manager.close()
    await http_api_client.async_logout()

if __name__ == '__main__':
    if os.name == 'nt':
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
    loop.stop()

Smart bulbs implement the meross_iot.controller.mixins.light.LightMixin interface, which handle RGB color settings, as well as luminance and color temperature. More details on this class are available here

Controlling garage door openers

Meross garage door openers are somehow basic: in most cases they only simulate the button-press of the garage door. The door state is instead monitored with a specific sensor, mounted on the garage door. Such sensor is not able to tell you the exact state of the garage door. In fact, it only tells you if the door is closed or not closed (i.e. empty).

When you operate the Meross Garage Opener, it sends the signal to the garage motor, which starts opening/closing. Then, once the door closes, the sensor reports “closing state”, and the door is marked as closed. However, when opening the door, things are quite different: as soon as the motor is operated, the sensor quickly reports “door opened” state as the magnet proximity sensor immediately changes state, even if the door is not completely opened.

Warning

Operating garage door is dangerous and might not be safe. You use this capability at your own risk, absolving the author of this library of all responsibility.

Operating door opener

import asyncio
import os

from meross_iot.controller.mixins.garage import GarageOpenerMixin
from meross_iot.http_api import MerossHttpClient
from meross_iot.manager import MerossManager

EMAIL = os.environ.get('MEROSS_EMAIL') or "YOUR_MEROSS_CLOUD_EMAIL"
PASSWORD = os.environ.get('MEROSS_PASSWORD') or "YOUR_MEROSS_CLOUD_PASSWORD"


async def main():
    # Setup the HTTP client API from user-password
    http_api_client = await MerossHttpClient.async_from_user_password(email=EMAIL, password=PASSWORD, api_base_url="https://iot.meross.com")

    # Setup and start the device manager
    manager = MerossManager(http_client=http_api_client)
    await manager.async_init()

    # Retrieve all the devices that implement the garage-door opening mixin
    await manager.async_device_discovery()
    openers = manager.find_devices(device_class=GarageOpenerMixin, device_type="msg100")

    if len(openers) < 1:
        print("No garage opener found...")
    else:
        dev = openers[0]

        # Update device status: this is needed only the very first time we play with this device (or if the
        #  connection goes down)
        await dev.async_update()

        # Check current door status
        open_status = dev.get_is_open()
        if open_status:
            print(f"Door {dev.name} is open")
        else:
            print(f"Door {dev.name} is closed")

        # To open the door, uncomment the following:
        #print(f"Opening door {dev.name}...")
        #await dev.async_open(channel=0)
        #print("Door opened!")
        #
        # Wait a bit before closing it again
        #await asyncio.sleep(5)
        #
        #print(f"Closing door {dev.name}...")
        #await dev.async_close()
        # print(f"Door closed!")

    # Close the manager and logout from http_api
    manager.close()
    await http_api_client.async_logout()

if __name__ == '__main__':
    if os.name == 'nt':
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
    loop.stop()

Garage door functionality is implemented by the meross_iot.controller.mixins.garage.GarageOpenerMixin. Have a look at here for more details.

Reading sensors

Meross devices might be equipped with sensors. Some devices (like temperature and humidity sensor) are readonly, configuring themselves as proper sensor devices. Others, as the MSS310, the Thermostat valve or the garage openers are instead actuators that offer some data reading capabilities. For this reason, there is no “general” sensor mixin class; on the contrary, you should rely on the capabilities offered by other mixins.

The following example will show you how to read power consumption data from a MSS310 plug, which implements both the meross_iot.controller.mixins.electricity.ElectricityMixin and the meross_iot.controller.mixins.consumption.ConsumptionXMixin.

Reading power consumption

import asyncio
import os

from meross_iot.controller.mixins.electricity import ElectricityMixin
from meross_iot.http_api import MerossHttpClient
from meross_iot.manager import MerossManager

EMAIL = os.environ.get('MEROSS_EMAIL') or "YOUR_MEROSS_CLOUD_EMAIL"
PASSWORD = os.environ.get('MEROSS_PASSWORD') or "YOUR_MEROSS_CLOUD_PASSWORD"


async def main():
    # Setup the HTTP client API from user-password
    http_api_client = await MerossHttpClient.async_from_user_password(email=EMAIL, password=PASSWORD, api_base_url="https://iot.meross.com")

    # Setup and start the device manager
    manager = MerossManager(http_client=http_api_client)
    await manager.async_init()

    # Retrieve all the devices that implement the electricity mixin
    await manager.async_device_discovery()
    devs = manager.find_devices(device_class=ElectricityMixin)

    if len(devs) < 1:
        print("No electricity-capable device found...")
    else:
        dev = devs[0]

        # Update device status: this is needed only the very first time we play with this device (or if the
        #  connection goes down)
        await dev.async_update()

        # Read the electricity power/voltage/current
        instant_consumption = await dev.async_get_instant_metrics()
        print(f"Current consumption data: {instant_consumption}")

    # Close the manager and logout from http_api
    manager.close()
    await http_api_client.async_logout()

if __name__ == '__main__':
    if os.name == 'nt':
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
    loop.stop()

In this case, the core of the script is the method async_get_instant_metrics() which reads the current electricity data from the device. More details on the electricity mixin are available here

For reading data from the MS100 temperature and humidity sensor, you can rely on the following snippet.

Reading from sensor

import asyncio
import os

from meross_iot.http_api import MerossHttpClient
from meross_iot.manager import MerossManager

EMAIL = os.environ.get('MEROSS_EMAIL') or "YOUR_MEROSS_CLOUD_EMAIL"
PASSWORD = os.environ.get('MEROSS_PASSWORD') or "YOUR_MEROSS_CLOUD_PASSWORD"


async def main():
    # Setup the HTTP client API from user-password
    http_api_client = await MerossHttpClient.async_from_user_password(email=EMAIL, password=PASSWORD, api_base_url="https://iot.meross.com")

    # Setup and start the device manager
    manager = MerossManager(http_client=http_api_client)
    await manager.async_init()

    # Retrieve all the MS100 devices that are registered on this account
    await manager.async_device_discovery()
    sensors = manager.find_devices(device_type="ms100")

    if len(sensors) < 1:
        print("No MS100 plugs found...")
    else:
        dev = sensors[0]

        # Manually force and update to retrieve the latest temperature sensed from
        # the device. This ensures we get the most recent data and not a cached value
        await dev.async_update()

        # Access read cached data
        temp = dev.last_sampled_temperature
        humid = dev.last_sampled_humidity
        time = dev.last_sampled_time

        print(f"Current sampled data on {time.isoformat()}; Temperature={temp}°C, Humidity={humid}%")
    # Close the manager and logout from http_api
    manager.close()
    await http_api_client.async_logout()

if __name__ == '__main__':
    if os.name == 'nt':
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
    loop.stop()

More details on the specific methods offered by the Ms100 sensor device are documented within the Ms100Sensor class.

Controlling Thermostat

The Meross thermostat valve is operated via meross_iot.controller.subdevice.Mts100v3Valve class.

Operating the smart valve

import asyncio
import os
from random import randint

from meross_iot.http_api import MerossHttpClient
from meross_iot.manager import MerossManager

EMAIL = os.environ.get('MEROSS_EMAIL') or "YOUR_MEROSS_CLOUD_EMAIL"
PASSWORD = os.environ.get('MEROSS_PASSWORD') or "YOUR_MEROSS_CLOUD_PASSWORD"


async def main():
    # Setup the HTTP client API from user-password
    http_api_client = await MerossHttpClient.async_from_user_password(email=EMAIL, password=PASSWORD, api_base_url="https://iot.meross.com")

    # Setup and start the device manager
    manager = MerossManager(http_client=http_api_client)
    await manager.async_init()

    # Retrieve all the mts100v3 devices that are registered on this account
    await manager.async_device_discovery()
    sensors = manager.find_devices(device_type="mts100v3")

    if len(sensors) < 1:
        print("No mts100v3 plugs found...")
    else:
        dev = sensors[0]

        # Manually force and update to retrieve the latest temperature sensed from
        # the device (this ensures we get the most recent value rather than a cached one)
        await dev.async_update()

        # Access read cached data
        on_off = dev.is_on()

        # Turn on the device if it's not on
        if not on_off:
            print(f"Device {dev.name} is off, turning it on...")
            await dev.async_turn_on()

        temp = await dev.async_get_temperature()
        print(f"Current ambient temperature = {temp} °C, "
              f"Target Temperature = {dev.target_temperature}, "
              f"mode = {dev.mode},"
              f"heating = {dev.is_heating}")

        # Randomly choose a temperature between min and max
        new_temp = randint(dev.min_supported_temperature, dev.max_supported_temperature)
        print(f"Setting target temperature to {new_temp}")
        await dev.async_set_target_temperature(new_temp)

    # Close the manager and logout from http_api
    manager.close()
    await http_api_client.async_logout()

if __name__ == '__main__':
    if os.name == 'nt':
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
    loop.stop()

For more information about all the properties and methods exposed by the valve class, have a look at Valve reference