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


# Single device being used
dev =""))

# Search for all devices in the 192.168.1 subnet
devices =""))
for addr, dev in devices.items():
    print(f"{addr} >> {dev}")

In this case if we just wanted to control one device we can use:'IP')
turning on a smart light

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 =""))
for addr, dev in devices.items():
    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()

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

        SQLAlchemy session: DB Session
    db = SessionLocal()
        yield db
    except Exception:

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

        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

        BaseModel ( Pydantic Base Model): Pydantic Schema

    name: str
    ip: str

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

        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

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

        _type_: _description_
    db_device = Device(, ip=device_create.ip)
    return db_device

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

        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.

        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

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

        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

        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"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

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(, str(disp.ip), str(
Command to display all 

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

def turn_on_device(id: int):
    """Turn on a device based on its id

        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)
    with Progress(
    ) as progress:
        progress.add_task(description="Exec Command", total=None)
        res =

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

The full code can be found on my github.