Changes for python3.11

This commit is contained in:
Martin Bauer 2024-03-06 17:15:07 +01:00
parent ef038a29a4
commit 11bf0505fb
4 changed files with 19 additions and 16 deletions

View File

@ -1,4 +1,4 @@
from dataclasses import dataclass from dataclasses import dataclass, field
import struct import struct
import colorsys import colorsys
@ -98,8 +98,8 @@ class EffectAlexaSwipeConfig:
bell_curve_width_in_leds: float = 3 bell_curve_width_in_leds: float = 3
start_position: float = 180 # in degrees start_position: float = 180 # in degrees
forward: bool = True forward: bool = True
primary_color: ColorRGBW = ColorRGBW(0, 0, 1, 0) primary_color: ColorRGBW = field(default_factory=lambda: ColorRGBW(0, 0, 1, 0))
secondary_color: ColorRGBW = ColorRGBW(0, 200 / 255, 1, 0) secondary_color: ColorRGBW = field(default_factory=lambda: ColorRGBW(0, 200 / 255, 1, 0))
def as_bytes(self) -> bytes: def as_bytes(self) -> bytes:
return struct.pack( return struct.pack(
@ -118,8 +118,8 @@ class EffectRandomTwoColorInterpolationConfig:
num_segments: int = 3 num_segments: int = 3
hue1_random: bool = False hue1_random: bool = False
hue2_random: bool = False hue2_random: bool = False
color1: ColorHSV = ColorHSV(240, 1, 1) color1: ColorHSV = field(default_factory=lambda: ColorHSV(240, 1, 1))
color2: ColorHSV = ColorHSV(192, 1, 1) color2: ColorHSV = field(default_factory=lambda: ColorHSV(192, 1, 1))
def as_bytes(self) -> bytes: def as_bytes(self) -> bytes:
c1 = ColorHSV.fromRGB(self.color1) if isinstance(self.color1, ColorRGBW) else self.color1 c1 = ColorHSV.fromRGB(self.color1) if isinstance(self.color1, ColorRGBW) else self.color1
@ -136,7 +136,7 @@ class EffectRandomTwoColorInterpolationConfig:
class EffectCircularConfig: class EffectCircularConfig:
speed: float = 360 # in degrees per second speed: float = 360 # in degrees per second
width: float = 180 # in degrees width: float = 180 # in degrees
color: ColorRGBW = ColorRGBW(0, 0, 1, 0) color: ColorRGBW = field(default_factory=lambda: ColorRGBW(0, 0, 1, 0))
def as_bytes(self) -> bytes: def as_bytes(self) -> bytes:
return struct.pack("<ff", self.speed, self.width) + self.color.as_bytes() return struct.pack("<ff", self.speed, self.width) + self.color.as_bytes()
@ -144,8 +144,8 @@ class EffectCircularConfig:
@dataclass @dataclass
class EffectSwipeAndChange: class EffectSwipeAndChange:
swipe: EffectAlexaSwipeConfig = EffectAlexaSwipeConfig() swipe: EffectAlexaSwipeConfig = field(default_factory=lambda: EffectAlexaSwipeConfig())
change: EffectRandomTwoColorInterpolationConfig = EffectRandomTwoColorInterpolationConfig() change: EffectRandomTwoColorInterpolationConfig = field(default_factory=lambda: EffectRandomTwoColorInterpolationConfig())
def as_bytes(self) -> bytes: def as_bytes(self) -> bytes:
return self.swipe.as_bytes() + self.change.as_bytes() return self.swipe.as_bytes() + self.change.as_bytes()
@ -164,4 +164,4 @@ class EffectReverseSwipe:
return struct.pack("<fff", self.swipeSpeed, self.bellCurveWidthInLeds, self.startPosition) return struct.pack("<fff", self.swipeSpeed, self.bellCurveWidthInLeds, self.startPosition)
def __repr__(self) -> str: def __repr__(self) -> str:
return f"Reverse swipe, speed {self.swipeSpeed}, width in leds {self.bellCurveWidthInLeds}, start position {self.startPosition}" return f"Reverse swipe, speed {self.swipeSpeed}, width in leds {self.bellCurveWidthInLeds}, start position {self.startPosition}"

View File

@ -18,6 +18,7 @@ import warnings
from pprint import pprint from pprint import pprint
from typing import Optional from typing import Optional
from mqtt_json import start_mqtt from mqtt_json import start_mqtt
import aiohttp
yaml = YAML(typ='safe') yaml = YAML(typ='safe')
@ -247,7 +248,7 @@ def main(config_path):
cfg = load_config(config_path) cfg = load_config(config_path)
loop = asyncio.get_event_loop() loop = asyncio.get_event_loop()
hass = HomeAssistantClient(cfg["general"]["hass_url"], cfg["general"]["hass_token"], loop) hass = HomeAssistantClient(cfg["general"]["hass_url"], cfg["general"]["hass_token"], loop=loop)
coro = serial_asyncio.create_serial_connection(loop, coro = serial_asyncio.create_serial_connection(loop,
MusicMouseProtocol, MusicMouseProtocol,
@ -256,7 +257,7 @@ def main(config_path):
transport, protocol = loop.run_until_complete(coro) transport, protocol = loop.run_until_complete(coro)
controller = Controller(protocol, hass, cfg) controller = Controller(protocol, hass, cfg)
mqtt_cfg = cfg["general"]["mqtt"] mqtt_cfg = cfg["general"]["mqtt"]
loop.create_task(start_mqtt(protocol, mqtt_cfg["server"], mqtt_cfg["user"],mqtt_cfg["password"] )) loop.create_task(start_mqtt(protocol, mqtt_cfg["server"], mqtt_cfg["user"], mqtt_cfg["password"] ))
loop.create_task(hass.connect()) loop.create_task(hass.connect())
return controller, loop return controller, loop

View File

@ -1,12 +1,12 @@
from led_cmds import ColorRGBW, EffectStaticConfig, EffectStaticDetailedConfig, EffectCircularConfig, EffectRandomTwoColorInterpolationConfig, EffectAlexaSwipeConfig, EffectSwipeAndChange from led_cmds import ColorRGBW, EffectStaticConfig, EffectStaticDetailedConfig, EffectCircularConfig, EffectRandomTwoColorInterpolationConfig, EffectAlexaSwipeConfig, EffectSwipeAndChange
import asyncio import asyncio
import asyncio_mqtt import aiomqtt
import json import json
from copy import deepcopy from copy import deepcopy
class ShelveLightMqtt: class ShelveLightMqtt:
def __init__(self, protocol, client: asyncio_mqtt.Client): def __init__(self, protocol, client: aiomqtt.Client):
self._protocol = protocol self._protocol = protocol
self._mqtt_client = client self._mqtt_client = client
@ -176,14 +176,14 @@ async def start_mqtt(music_mouse_protocol, server, username, password):
reconnect_interval = 10 # [seconds] reconnect_interval = 10 # [seconds]
while True: while True:
try: try:
async with asyncio_mqtt.Client(hostname=server, username=username, password=password) as client: async with aiomqtt.Client(hostname=server, username=username, password=password) as client:
shelve_light = ShelveLightMqtt(music_mouse_protocol, client) shelve_light = ShelveLightMqtt(music_mouse_protocol, client)
await shelve_light.init() await shelve_light.init()
async with client.filtered_messages("musicmouse_json/#") as messages: async with client.filtered_messages("musicmouse_json/#") as messages:
await client.subscribe("musicmouse_json/#") await client.subscribe("musicmouse_json/#")
async for message in messages: async for message in messages:
await shelve_light.handle_light_message(message) await shelve_light.handle_light_message(message)
except asyncio_mqtt.MqttError as error: except aiomqtt.MqttError as error:
print(f'Error "{error}". Reconnecting in {reconnect_interval} seconds') print(f'Error "{error}". Reconnecting in {reconnect_interval} seconds')
finally: finally:
await asyncio.sleep(reconnect_interval) await asyncio.sleep(reconnect_interval)

View File

@ -1,3 +1,5 @@
pyserial-asyncio==0.6 pyserial-asyncio==0.6
python-vlc==3.0.12118 python-vlc==3.0.20123
hass-client==0.1.2 hass-client==0.1.2
ruamel.yaml==0.18.6
aiomqtt==2.0.0