Creating a commandline program to control Kasa Lights with python-kasa and typer

Sorry I've been away for a while. Started a new job so blog writing kind off fell off

So I bought a Stream Deck (sadly not a steam deck), and one of the cool things I can do from it is to have it fire off python scripts directly. 1

In my office I have a bunch of Kasa Lights that I wanted to be able to control directly from the Stream Deck, because I am lazy and I don't always have my phone with me.

Using python-kasa greatly simplifies making calls to the lights directly. I also wanted to have a nice command line interface for when I just want to activate directly from terminal, thats where typer comes in.

pip install python-kasa
pip install typer

The basic use of python-kasa is as follows

import asyncio
from kasa import Discover

asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

# Single device being used
dev = asyncio.run(Discover.discover_single(host="192.168.1.205"))
print(f"{dev}")

# Search for all devices in the 192.168.1 subnet
devices = asyncio.run(Discover.discover(target="192.168.1.255"))
for addr, dev in devices.items():
    asyncio.run(dev.update())
    print(f"{addr} >> {dev}")

In this case if we just wanted to control one device we can use:

dev=asyncio.run(Discover.discover_single(host='IP')
dev.update()
dev.turn_on

If you know what specific type of device you want to control you can use that class SmartPlug("ip"), in my case I don't since I have different devices and really only want to handle turning them off and on.

If we wanted to discover all of the devices we could do:

devices = asyncio.run(Discover.discover(target="192.168.1.255"))
for addr, dev in devices.items():
    asyncio.run(dev.update())
    print(f"{addr} >> {dev}")

This will iterate over the target range (in this case "192.168.1.XXX") to look for Kasa Devices.

Storing the IP's

I have terrible memory so I wanted to store the IP's of the devices so I wouldn't have to remember them and could just use an id to select the device. In order to do so I just whipped up a quick set of Create and Read functions to a sqlite db. We use contextlib and sqlalchmey and pydantic to help us out. The code is pretty straight forward we just create a session and use the @contextmanager decorator. Then create some pydantic validators and finally use sqlalchemy to handle the reads and writes. I didn't really worry too much about deleting the db, since I can just delete directly from the cmd if something goes wrong.

SQLALCHEMY_DATABASE_URL = "sqlite:///sql_app.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()


@contextmanager
def get_db():
    """Session Management for DB

    Yields:
        SQLAlchemy session: DB Session
    """
    db = SessionLocal()
    try:
        yield db
        db.commit()
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()


class Device(Base):
    """Model Representation of Device

    Args:
        Base ( SQLAlchemy ): SQL Alchemy Base Model
    """

    __tablename__ = "device_list"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(Integer, unique=True, index=True)
    ip = Column(String, unique=True)


class DeviceBase_Schm(BaseModel):
    """Validation for Records

    Args:
        BaseModel ( Pydantic Base Model): Pydantic Schema
    """

    name: str
    ip: str


class Device_Schm(DeviceBase_Schm):
    """Validation for Records

    Args:
        DeviceBase_Schm ( Pydantic Model): Pydantic Schema
    """

    id: int

    class Config:
        orm_mode = True

def create_device_db(db: Session, device_create: DeviceBase_Schm):
    """Insert validated record into DB

    Args:
        db (Session): Session passed as contex
        device_create (DeviceBase_Schm): Pydantic validated record

    Returns:
        _type_: _description_
    """
    db_device = Device(name=device_create.name, ip=device_create.ip)
    db.add(db_device)
    db.commit()
    db.refresh(db_device)
    return db_device


def list_all_devices(
    db: Session, skip: int = 0, limit: int = 100
) -> list[Device]:
    """Returns all device records in db

    Args:
        db (Session): Session Context Manager
        skip (int, optional): How many records to skip. Defaults to 0.
        limit (int, optional): Limit of Recoreds returned. Defaults to 100.

    Returns:
        List[SQLAlchemy Model]: Model Representation of all devices in db
    """
    return db.query(Device).offset(skip).limit(limit).all()


def get_device_ip(db: Session, id: int) -> Device:
    """Return IP of device from id

    Args:
        db (Session): DB session passed using context
        id (int): ID of object to get id of

    Returns:
        SQLAlchemy Model : SQLAlchemy Model Single Record
    """
    return db.query(Device).filter_by(id=id).first()

Controlling the Lights

Now that we have the lights stored in a db we can get the ip from the id very easly.

def get_one_device(id: int):
    """Acquire a device on its id

    Args:
        id (int): id of device to get
    """
    with get_db() as db:
        dev = get_device_ip(db, id)
        disp = Device_Schm.from_orm(dev)

Now we can concentrate on turning the lights off and on. To do so we first create a function that will take an ip, create a device and turn it on. We have to create this function separately so it will play nice with asyncio (Otherwise the thread will lock and you will have to wait for the light to respond).

async def aturn_on(ip):
    dev = await Discover.discover_single(host=ip)
    await dev.update()
    await dev.turn_on()
    return dev

Now we can fire off that thread with

asyncio.run(aturn_of("IP"))

Using Typer for a nice command line

Now we can go ahead and create our nice typer interface. Its very easy to create, all that really needs to be done is create an app (and since we are getting fancy a Console)

import typer
from rich.console import Console
from rich.progress import Progress, SpinnerColumn, TextColumn
from rich.table import Table

app = typer.Typer()
console = Console()

Now if we want to create a command we just need to use the @app.command() decorator

@app.command()
def get_all_devices():
    """Display a list of all devices to terminal"""
    with get_db() as db:
        devs = list_all_devices(db)
        table = Table("Name", "IP", "ID")
        for dev in devs:
            disp = Device_Schm.from_orm(dev)
            table.add_row(disp.name, str(disp.ip), str(disp.id))
        console.print(table)

So now we can setup a command to turn on a light based on the id, that will show a nice spinner when called.

@app.command()
def turn_on_device(id: int):
    """Turn on a device based on its id

    Args:
        id (int): id of device to turn off
    """
    with get_db() as db:
        dev = get_device_ip(db, id)
        disp = Device_Schm.from_orm(dev)
    console.log(disp.ip)
    with Progress(
        SpinnerColumn(),
        TextColumn("[progress.description]{task.description}"),
        transient=True,
    ) as progress:
        progress.add_task(description="Exec Command", total=None)
        res = asyncio.run(aturn_on(disp.ip))

Now I can just use Stream Decks - Advanced Launcher  to control my lights directly.

The full code can be found on my github.