thingnario logger 1.0

.gitignore

*.pyc
/.venv/

main.py

import argparse
import json

from sqlalchemy import create_engine, engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column

OWNER_KEYS = ('americas', 'chailease', 'demo', 'plus', 'thailand', 'thingnario', 'tsc')


class Base(DeclarativeBase):
    pass


class SolarInverter(Base):
    __tablename__ = 'SolarInverter'

    id:           Mapped[int] = mapped_column(primary_key=True)
    plantNo:      Mapped[str]
    invModelName: Mapped[str]


class DeviceLoggers(Base):
    __tablename__ = 'DeviceLogger'

    id:              Mapped[int] = mapped_column(primary_key=True)
    plantNo:         Mapped[str]
    firmwareVersion: Mapped[str]


def main():
    args = get_args()
    for owner_key, plant_no in query_all(args.url, args.model):
        print(owner_key, plant_no)


def get_args():
    args = argparse.ArgumentParser()
    args.add_argument(
        '--url',
        required=True,
        type=engine.make_url,
    )
    args.add_argument(
        '--model',
        nargs='+',
        required=True,
    )
    return args.parse_args()


def query_all(url: engine.URL, models: list[str]):
    for owner_key in OWNER_KEYS:
        for plant_no in query(url, owner_key, models):
            yield owner_key, plant_no


def query(url: engine.URL, owner_key: str, models: list[str]):
    engine = create_engine(url.set(database=f'solar_PV_{owner_key}'))
    with Session(engine) as session:
        statement = select(SolarInverter.plantNo).where(SolarInverter.invModelName.in_(models))
        plant_nos = set(row.plantNo for row in session.execute(statement).all())
        loggers = (
            (set(json.loads(row.plantNo)), row.firmwareVersion)
            for row in session.query(DeviceLoggers).all())

    for logger_plant_nos, version in loggers:
        if plant_nos & logger_plant_nos and is_old_version(version):
            yield from logger_plant_nos


def is_old_version(version: str | None):
    return version is None or version.startswith(('1.', '2.'))


if __name__ == '__main__':
    main()

requirements.txt

PyMySQL==1.1.1
SQLAlchemy==2.0.36

setup.cfg

[flake8]
max_line_length = 100

[isort]
line_length = 100