Coverage for luckyloop/client.py : 100%

#!/usr/bin/env python3 

# -*- coding: utf-8 -*- 

from typing import Dict, Any 

import socket 

class LuckyClient: 

"""Allows to remote control the LuckyLoop 

args 

---- 

host: str 

defaults to 127.0.0.1, but should be set in production to the ip-adress of the LuckyLoop device 

port: int 

defaults to 1219 because LuckyServer has the acronym LS and L is 12th character in alphabet and S is the 19th. 

.. note:: 

For safety reasons, parameters like target frequency and phase are not set during initialization of the client, as this could interrupt an ongoing process on LuckyLoop when an user tries to reconnect. All parameters have be set explicitly after initialization. 

Example 

------- 

.. code-block:: python 

luckyloop = LuckyClient() 

luckyloop.frequency = 20 sets the target frequency to 20 

luckyloop.phase = 2 # sets the target phase to 2 

luckyloop.trigger() # triggers immediatly 

""" 

buffer_size = 8192 

def __init__(self, host: str = "127.0.0.1", port: int = 1219): 

self.host = host 

self.port = port 

self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 

self.enquire() 

@property 

def phase(self) -> int: 

"get the current phase set on LuckyLoop" 

status = self.enquire() 

return status["target_phase"] 

@phase.setter 

def phase(self, target_phase: int): 

"""set the target_phase on LuckyLoop 

args 

---- 

target_phase 

an integer between 0 and 360°  

""" 

if target_phase < 0 or target_phase > 360: 

raise ValueError("Target phase should be between 0 and 360") 

if type(target_phase) != int: 

raise ValueError("Target phase must be an integer") 

condition = { 

"threshold": "=", 

"target_phase": target_phase, 

"target_frequency": "=", 

"trigger": "=", 

} 

self.set_condition(condition) 

@property 

def threshold(self) -> int: 

"get the current threshold set on LuckyLoop" 

status = self.enquire() 

return status["threshold"] 

@threshold.setter 

def threshold(self, threshold: int): 

"""set the target_phase on LuckyLoop 

args 

---- 

target_phase 

an integer between 0 and 360°  

""" 

if type(threshold) != int or threshold < 0: 

raise ValueError("threshold must be an integer larger or equal to 0") 

condition = { 

"threshold": threshold, 

"target_phase": "=", 

"target_frequency": "=", 

"trigger": "=", 

} 

self.set_condition(condition) 

@property 

def frequency(self) -> int: 

"get the current frequency set on LuckyLoop" 

status = self.enquire() 

return status["target_frequency"] 

@frequency.setter 

def frequency(self, target_frequency: int): 

"""set the target_frequency on LuckyLoop 

LuckyLoop calculates phase and amplitude continually for a specific frequency. Change this target_frequency 

args 

---- 

target_frequency 

an integer of at least 1 Hz 

""" 

if target_frequency <= 0: 

raise ValueError("Target frequency must be higher than 0 Hz") 

if type(target_frequency) != int: 

raise ValueError("Target frequency must be an integer") 

condition = { 

"threshold": "=", 

"target_phase": "=", 

"target_frequency": target_frequency, 

"trigger": "=", 

} 

self.set_condition(condition) 

def trigger(self): 

"""trigger LuckyLoop 

Immediatly pulls the specified TRIGOUT to HIGH. LuckyLoop handles from then on, and by default it pulls it LOW after one cycle. 

""" 

condition = { 

"threshold": "=", 

"target_phase": "=", 

"target_frequency": "=", 

"trigger": 1, 

} 

return self.set_condition(condition) 

def set_condition(self, condition: Dict[str, Any]) -> Dict[str, Any]: 

return self.query(self.encode(condition)) 

def shutdown(self): 

"shuts LuckyLoop down gracefully" 

message = "[shutdown:]\0".encode("ascii") 

self.query(message) 

def enquire(self) -> Dict[str, int]: 

"enquire about the current status and version of LuckyLoop" 

message = "[send_status:]\0".encode("ascii") 

return self.query(message) 

def query(self, message: bytes) -> Dict[str, Any]: 

"send and receive a message to LuckyLoop" 

self.connect() 

self.socket.sendall(message) 

status = self.socket.recv(self.buffer_size) 

self.close() 

return self.decode(status) 

def connect(self): 

"connect with LuckyLoop" 

self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 

try: 

self.socket.connect((self.host, self.port)) 

self.socket.settimeout(1) 

except ConnectionRefusedError: 

raise ConnectionError( 

"It is not possible to reach LuckyLoop. Restart the device and make sure that IP and port are correct." 

) 

def close(self): 

"close the connection to LuckyLoop" 

# self.socket.shutdown(1) 

self.socket.close() 

@staticmethod 

def encode(condition: Dict[str, Any]) -> bytearray: 

"serialize a status dictionary for sending" 

message = f"[set_condition:{condition['threshold']},{condition['target_phase']},{condition['target_frequency']},{condition['trigger']}]".encode( 

"ascii" 

) 

return message 

@staticmethod 

def decode(status: bytearray) -> Dict[str, int]: 

"deserialize a message from LuckyServer and return it as a dictionary" 

items = status.decode("ascii").replace(":", "").strip().split(" ") 

cond = dict() 

for k, v in zip(items[0::2], items[1::2]): 

cond[k.lower()] = int(v) 

return cond