Python code to LUA

[bednarekluc]

Hi there,

I'm looking to convert this python code to LUA.
I have no clue how to start.
Can this be done?
Any help would be greatly appreciated.
The full code is attached to this post.

These are the Classes : 

Code:

# Copyright 2021, 2022 Milan Meulemans.
#
# This file is part of aioaseko.
#
# aioaseko is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# aioaseko is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with aioaseko.  If not, see <https://www.gnu.org/licenses/>.

"""aioAseko web API account."""
from __future__ import annotations

from dataclasses import dataclass

from aiohttp import ClientError, ClientResponse, ClientSession

from .exceptions import APIUnavailable, InvalidAuthCredentials
from .unit import Unit


class WebAccount:
    """Aseko web account."""

    def __init__(self, session: ClientSession, username: str, password: str) -> None:
        """Init Aseko account."""
        self._session = session
        self._username = username
        self._password = password

    async def _request(
        self, method: str, path: str, data: dict | None = None
    ) -> ClientResponse:
        """Make a request to the Aseko web API."""
        resp = await self._session.request(
            method, f"https://pool.aseko.com/api/{path}", data=data
        )
        if resp.status == 401:
            raise InvalidAuthCredentials
        try:
            resp.raise_for_status()
        except ClientError:
            raise APIUnavailable
        return resp

    async def login(self) -> AccountInfo:
        """Login to the Aseko web API."""
        resp = await self._request(
            "post",
            "login",
            {
                "username": self._username,
                "password": self._password,
                "agree": "on",
            },
        )
        data = await resp.json()
        return AccountInfo(data["email"], data["userId"], data.get("language"))

    async def get_units(self) -> list[Unit]:
        """Get units."""
        resp = await self._request("get", "units")
        data = await resp.json()
        return [
            Unit(
                self,
                int(item["serialNumber"]),
                item["type"],
                item.get("name"),
                item.get("notes"),
                item["timezone"],
                item["isOnline"],
                item["dateLastData"],
                item["hasError"],
            )
            for item in data["items"]
        ]


@dataclass(frozen=True)
class AccountInfo:
    """Aseko account info."""

    email: str
    user_id: str
    language: str | None

And this Is the code to run the classes :

Code:

from aiohttp import ClientSession
from asyncio import run

import aioaseko

async def main():
    async with ClientSession() as session:
        account = aioaseko.MobileAccount(session, "aioAseko@example.com", "passw0rd")
        try:
            await account.login()
        except aioaseko.InvalidAuthCredentials:
            print("The username or password you entered is wrong.")
            return
        units = await account.get_units()
        for unit in units:
            print(unit.name)
            await unit.get_state()
            print(f"Water flow: {unit.water_flow}")
            for variable in unit.variables:
                print(variable.name, variable.current_value, variable.unit)
        await account.logout()
run(main())

[Batko]

I'm not sure what it is, gpt chat saw it.

Code:

I'll assume that you want me to rewrite the Python code for Aseko web API into a Lua script that can interact with the KNX protocol. Since the original code interacts with a web API, there will be some significant changes to the logic and implementation to make it work with the KNX protocol. Here is a possible implementation:

Code:

-- Import required libraries
local knx = require("knx")
local json = require("json")

-- Define Aseko KNX address
local aseko_address = "0/0/1"

-- Define KNX group addresses for login and unit data
local login_address = "0/0/2"
local units_address = "0/0/3"

-- Define Aseko account credentials
local username = "your_username"
local password = "your_password"

-- Define function to handle incoming KNX telegrams
local function handle_telegram(address, data)
    if address == login_address then
        -- Parse login data from KNX telegram
        local login_data = json.decode(data)

        -- Login to Aseko web API
        -- ... implementation ...

        -- Send unit data to KNX
        -- ... implementation ...
    elseif address == units_address then
        -- Parse unit data from KNX telegram
        local unit_data = json.decode(data)

        -- Send unit command to Aseko web API
        -- ... implementation ...
    end
end

-- Create a KNX device and connect to the KNX bus
local device = knx.create_device("192.168.1.1")
device:connect()

-- Register the handle_telegram function to be called when a KNX telegram is received
device:listen(handle_telegram)

-- Main loop to keep the program running
while true do
    -- Wait for incoming KNX telegrams
    device:step()
end

[bednarekluc]

The important part is missing calling the API.

[admin]

You've already asked this question some time ago
Send you Aseko login/password via PM. Otherwise making a working example is too complicated.

[bednarekluc]

You've already asked this question some time ago
Send you Aseko login/password via PM. Otherwise making a working example is too complicated.

I have sent you the details in a PM.
Username and password and also what I did find out about the API

[admin]

Use this. Make sure you have the latest firmware as requests in such format won't work on older versions.
Change unitid, username, password as needed. unitid is the ID that you see when opening the status page (https://pool.aseko.com/XYZ/data/actual). Sometimes the login fails with gateway timeout, just run the script again if it happens.

Code:

json = require('json')
http = require('socket.http')

unitid = '1234'
username = 'user@example.com'
password = 'password'

body = json.encode({
  username = username,
  password = password,
  agree = 'on'
})

resp, code, hdrs, stat = http.request({
  url = 'https://pool.aseko.com/api/login',
  method = 'POST',
  headers = {
    ['Content-Type'] = 'application/json',
    ['Content-Length'] = #body,
  },
  body = body
})

setcookie = hdrs['set-cookie'] or ''
cookie = setcookie:split(';')[1]

if not cookie then
  log('login failed', resp)
  return
end

resp, code, hdrs, stat = http.request({
  url = 'https://pool.aseko.com/api/units/' .. unitid,
  method = 'GET',
  headers = {
    ['Cookie'] = cookie
  }
})

data = json.pdecode(resp)
if type(data) ~= 'table' then
  log('data request failed', resp)
  return
end

values = {}

for _, var in ipairs(data.variables) do
  values[ var.type ] = var.currentValue
end

grp.checkupdate('1/1/1', values.rx)
grp.checkupdate('1/1/2', values.waterTemp)
grp.checkupdate('1/1/3', values.ph)

[bednarekluc]

Use this. Make sure you have the latest firmware as requests in such format won't work on older versions.
Change unitid, username, password as needed. unitid is the ID that you see when opening the status page (https://pool.aseko.com/XYZ/data/actual). Sometimes the login fails with gateway timeout, just run the script again if it happens.

Code:

json = require('json')
http = require('socket.http')

unitid = '1234'
username = 'user@example.com'
password = 'password'

body = json.encode({
  username = username,
  password = password,
  agree = 'on'
})

resp, code, hdrs, stat = http.request({
  url = 'https://pool.aseko.com/api/login',
  method = 'POST',
  headers = {
    ['Content-Type'] = 'application/json',
    ['Content-Length'] = #body,
  },
  body = body
})

setcookie = hdrs['set-cookie'] or ''
cookie = setcookie:split(';')[1]

if not cookie then
  log('login failed', resp)
  return
end

resp, code, hdrs, stat = http.request({
  url = 'https://pool.aseko.com/api/units/' .. unitid,
  method = 'GET',
  headers = {
    ['Cookie'] = cookie
  }
})

data = json.pdecode(resp)
if type(data) ~= 'table' then
  log('data request failed', resp)
  return
end

values = {}

for _, var in ipairs(data.variables) do
  values[ var.type ] = var.currentValue
end

grp.checkupdate('1/1/1', values.rx)
grp.checkupdate('1/1/2', values.waterTemp)
grp.checkupdate('1/1/3', values.ph)

Works like a charm.
I put this script as a resident script and after a while I get this error 

Code:

string: {"status":401,"error":"You've been signed out","message":"Your session has expired. Please sign in again with your credentials."}

[admin]

Use a scheduled script instead of resident.

[bednarekluc]

Indeed no more error.
Thanks a lot.