velux kfl-200

[domotiqa]

looks like Velux changed the API of the KLF200

https://velcdn.azureedge.net/~/media/com...er3-16.pdf

looks harder

an example

Code:

## ======================================================
## Script pour demarrer une scene de la box Velux KLF-200
## ------------------------------------------------------
## Version 0.1.1 - Last update : 03.11.18 by Christian R.
## ======================================================

import ssl, socket, time, struct
from time import sleep

# ===============================================================================
# Variables
# ===============================================================================

SCENE_ID        = 1

KLF200_ADDRESS  = "192.168.0.12"
PASSWORD        = "Velux123"
PORT            = 51200
LoopDelay       = 1

# ===============================================================================
# slip
# ===============================================================================

END     = b"\xC0"    # SLIP escape character as per RFC 1055
ESC     = b"\xDB"    # SLIP escape character as per RFC 1055
ESC_END = b"\xDC"    # SLIP escape character as per RFC 1055
ESC_ESC = b"\xDD"    # SLIP escape character as per RFC 1055


def slip_pack(inputFrame):    
   data = inputFrame
   data = data.replace(ESC, ESC + ESC_ESC)
   data = data.replace(END, ESC + ESC_END)
   return END + data + END

def slip_unpack(inputFrame):
   data = inputFrame
   if(data[0:1]==END and data[-1:]==END):
       data = data.replace(ESC + ESC_END, END)
       data = data.replace(ESC + ESC_ESC, ESC)
       return data[1:-1]
   else:
       print("Error: No SLIP frame!\n")
       return inputFrame # error -> return input

# ===============================================================================
# toolbox
# ===============================================================================

def getIndex(sourceDict, value):
   return (k for k, v in sourceDict.items() if v == value).__next__()

def toHex(s):
   return ":".join("{:02x}".format(c) for c in s)

# ===============================================================================
# ActivateScene
# ===============================================================================

def process_connection(conn):
   conn.settimeout(10.0) # 10 sec
   
   print("Send valid password")
   conn.write(bytes(ST_GW_PASSWORD_ENTER_REQ(PASSWORD)))
   print("Received: ", toHex(slip_unpack(conn.recv())), "\n")
   
   time.sleep(LoopDelay)

   print("Activate Scene with ID = ", SCENE_ID)
   conn.write(bytes(ST_GW_ACTIVATE_SCENE_REQ(bSceneID=SCENE_ID)))
   print("Received: ", toHex(slip_unpack(conn.recv())))
   

def main():
   sock = socket.socket(socket.AF_INET)
   sock.settimeout(10.0)

   context = ssl.SSLContext(ssl.PROTOCOL_TLS)
   context.check_hostname = False
   #accept self-signed certificate
   context.verify_mode = ssl.CERT_NONE

   conn = context.wrap_socket(sock, server_hostname=KLF200_ADDRESS)

   try:
       conn.connect((KLF200_ADDRESS, PORT))
       process_connection(conn)
   except BaseException as e:
       raise(e)
   finally:
       conn.close()

# ===============================================================================
# klf200api
# ===============================================================================

GW_ACTIVATE_SCENE_REQ                       =  0x0412
GW_ACTIVATE_SCENE_CFM                       =  0x0413

GW_PASSWORD_ENTER_REQ                       =  0x3000
GW_PASSWORD_ENTER_CFM                       =  0x3001

dictPriorityLevel = {
   0: 'Human Protection',
   1: 'Environment Protection',
   2: 'User Level 1',
   3: 'User Level 2',
   4: 'Comfort Level 1',
   5: 'Comfort Level 2',
   6: 'Comfort Level 3',
   7: 'Comfort Level 4',
}

dictCommandOriginator = {
   0x00: "LOCAL_USER",  # // User pressing button locally on actuator
   0x01: "USER",  # // User Remote control causing action on actuator
   0x02: "RAIN",  # // Sensor
   0x03: "TIMER",  # // Sensor
   0x04: "SECURITY",  # // SCD controlling actuator
   0x05: "UPS",  # // UPS unit
   0x06: "SFC",  # // Smart Function Controller
   0x07: "LSC",  # // Lifestyle Scenario Controller
   0x08: "SAAC",  # // Stand Alone Automatic Controls
   0x09: "WIND",  # // Wind detection
   0x10: "MYSELF",  # // Used when an actuator decides to move by itself
   0xFE: "AUTOMATIC_CYCLE",  # // Used in context with automatic cycle;
   0xFF: "EMERGENCY"  # // Used in context with emergency or security commands,
   # // -this command originator should never be disabled
}

dictVelocity = {
   0: 'DEFAULT',
   1: 'SILENT',
   2: 'FAST',
   255: 'VELOCITY_NOT_AVAILABLE', #Only used in status reply
}

# ===============================================================================
   
class ST_GW_FRAME:
   def __init__(self, Command):
       self.DataLength    = 0
       self.Command       = Command
       self.binary_output = b"";

   def __bytes__(self):
       self.binary_output = struct.pack("BB", 0, self.DataLength + 3)
       self.binary_output += struct.pack(">H", self.Command)
       self.binary_output += self.pack_data()
       self.binary_output += struct.pack("B", self.calc_crc())
       return slip_pack(self.binary_output)
   
   def calc_crc(self):
       crc = 0
       for sym in self.binary_output:
           crc = crc ^ int(sym)
       return crc

   def pack_data(self):
       return b""


class ST_GW_ACTIVATE_SCENE_REQ (ST_GW_FRAME):
   def __init__(self,
                wSessionID         = 0x1234,
                CommandOriginator  = 'USER',
                PriorityLevel      = 'User Level 2',
                bSceneID           = 0,
                Velocity           = 'DEFAULT'):
       ST_GW_FRAME.__init__(self, GW_ACTIVATE_SCENE_REQ)
       self.DataLength         = 6
       self.wSessionID         = wSessionID
       self.bCommandOriginator = getIndex(dictCommandOriginator, CommandOriginator)
       self.bPriorityLevel     = getIndex(dictPriorityLevel, PriorityLevel)
       self.bSceneID           = bSceneID
       self.bVelocity          = getIndex(dictVelocity, Velocity)
       

   def pack_data(self):
       ret = struct.pack(">H", self.wSessionID)
       ret += bytes([self.bCommandOriginator])
       ret += bytes([self.bPriorityLevel])
       ret += bytes([self.bSceneID])
       ret += bytes([self.bVelocity])
       return ret


class ST_GW_PASSWORD_ENTER_REQ (ST_GW_FRAME):
   def __init__(self, Password):
       ST_GW_FRAME.__init__(self, GW_PASSWORD_ENTER_REQ)
       self.DataLength = 32
       self.Password   = Password

   def pack_data(self):
       binary_data = bytes(self.Password,encoding='ascii')
       binary_len = len(binary_data)
       ret = binary_data[:self.DataLength if binary_len > self.DataLength else binary_len]
       while binary_len < self.DataLength:
           ret += b'\x00'
           binary_len = binary_len + 1
       return ret

# ===============================================================================
# Start script
# ===============================================================================
   
main()
print("Finished")