Skip to content
Merged
2 changes: 1 addition & 1 deletion .pre-commit-config.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ repos:
name: pylint (library code)
types: [python]
args:
- --disable=consider-using-f-string
- --disable=consider-using-f-string,duplicate-code
exclude: "^(docs/|examples/|tests/|setup.py$)"
- id: pylint
name: pylint (example code)
Expand Down
30 changes: 18 additions & 12 deletions adafruit_funhouse/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,12 @@
from adafruit_funhouse.graphics import Graphics
from adafruit_funhouse.peripherals import Peripherals

try:
from typing import Optional, Dict, Union, Callable, Sequence, List
from adafruit_dotstar import DotStar
except ImportError:
pass

__version__ = "0.0.0-auto.0"
__repo__ = "https://github.com/adafruit/Adafruit_CircuitPython_FunHouse.git"

Expand Down Expand Up @@ -65,17 +71,17 @@ class FunHouse(PortalBase):
def __init__(
self,
*,
url=None,
headers=None,
json_path=None,
regexp_path=None,
default_bg=0,
status_dotstar=None,
json_transform=None,
rotation=270,
scale=1,
debug=False,
):
url: Optional[str] = None,
headers: Dict[str, str] = None,
json_path: Optional[Union[List[str], List[List[str]]]] = None,
regexp_path: Optional[Sequence[str]] = None,
default_bg: int = 0,
status_dotstar: Optional[DotStar] = None,
json_transform: Optional[Union[Callable, List[Callable]]] = None,
rotation: int = 270,
scale: int = 1,
debug: bool = False,
) -> None:

network = Network(
status_dotstar=status_dotstar,
Expand Down Expand Up @@ -105,7 +111,7 @@ def __init__(

gc.collect()

def enter_light_sleep(self, sleep_time):
def enter_light_sleep(self, sleep_time: float) -> None:
"""
Enter light sleep and resume the program after a certain period of time.

Expand Down
9 changes: 8 additions & 1 deletion adafruit_funhouse/graphics.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,14 @@ class Graphics(GraphicsBase):
"""

# pylint: disable=too-many-instance-attributes, too-many-locals, too-many-branches, too-many-statements, too-few-public-methods
def __init__(self, *, default_bg=0, rotation=270, scale=1, debug=False):
def __init__(
self,
*,
default_bg: int = 0,
rotation: int = 270,
scale: int = 1,
debug: bool = False
) -> None:
self._debug = debug
self.display = board.DISPLAY
self.display.rotation = rotation
Expand Down
69 changes: 42 additions & 27 deletions adafruit_funhouse/network.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,12 @@
from adafruit_portalbase.network import NetworkBase
from adafruit_portalbase.wifi_esp32s2 import WiFi

try:
from typing import Optional, Union, Callable
from adafruit_dotstar import DotStar
except ImportError:
pass

__version__ = "0.0.0-auto.0"
__repo__ = "https://github.com/adafruit/Adafruit_CircuitPython_FunHouse.git"

Expand All @@ -54,18 +60,18 @@ class Network(NetworkBase):
def __init__(
self,
*,
status_dotstar=None,
extract_values=True,
debug=False,
):
status_dotstar: Optional[DotStar] = None,
extract_values: bool = True,
debug: bool = False,
) -> None:
super().__init__(
WiFi(status_led=status_dotstar),
extract_values=extract_values,
debug=debug,
)
self._mqtt_client = None

def init_io_mqtt(self):
def init_io_mqtt(self) -> IO_MQTT:
"""Initialize MQTT for Adafruit IO"""
try:
aio_username = self._secrets["aio_username"]
Expand All @@ -80,12 +86,12 @@ def init_io_mqtt(self):
# pylint: disable=too-many-arguments
def init_mqtt(
self,
broker,
port=8883,
username=None,
password=None,
use_io=False,
):
broker: str,
port: int = 8883,
username: str = None,
password: str = None,
use_io: bool = False,
) -> Union[MQTT.MQTT, IO_MQTT]:
"""Initialize MQTT"""
self.connect()
self._mqtt_client = MQTT.MQTT(
Expand All @@ -103,12 +109,14 @@ def init_mqtt(

# pylint: enable=too-many-arguments

def _get_mqtt_client(self):
def _get_mqtt_client(self) -> Union[MQTT.MQTT, IO_MQTT]:
if self._mqtt_client is not None:
return self._mqtt_client
raise RuntimeError("Please initialize MQTT before using")

def mqtt_loop(self, *args, suppress_mqtt_errors=True, **kwargs):
def mqtt_loop(
self, *args: int, suppress_mqtt_errors: bool = True, **kwargs: int
) -> None:
"""Run the MQTT Loop"""
self._get_mqtt_client()
if suppress_mqtt_errors:
Expand All @@ -123,7 +131,12 @@ def mqtt_loop(self, *args, suppress_mqtt_errors=True, **kwargs):
if self._mqtt_client is not None:
self._mqtt_client.loop(*args, **kwargs)

def mqtt_publish(self, *args, suppress_mqtt_errors=True, **kwargs):
def mqtt_publish(
self,
*args: Union[str, int, float],
suppress_mqtt_errors: bool = True,
**kwargs: Union[str, int, float]
) -> None:
"""Publish to MQTT"""
self._get_mqtt_client()
if suppress_mqtt_errors:
Expand All @@ -136,14 +149,16 @@ def mqtt_publish(self, *args, suppress_mqtt_errors=True, **kwargs):
if self._mqtt_client is not None:
self._mqtt_client.publish(*args, **kwargs)

def mqtt_connect(self, *args, **kwargs):
def mqtt_connect(
self, *args: Union[bool, str, int], **kwargs: Union[bool, str, int]
) -> None:
"""Connect to MQTT"""
self._get_mqtt_client()
if self._mqtt_client is not None:
self._mqtt_client.connect(*args, **kwargs)

@property
def on_mqtt_connect(self):
def on_mqtt_connect(self) -> Optional[Callable]:
"""
Get or Set the MQTT Connect Handler

Expand All @@ -153,12 +168,12 @@ def on_mqtt_connect(self):
return None

@on_mqtt_connect.setter
def on_mqtt_connect(self, value):
def on_mqtt_connect(self, value: Callable) -> None:
self._get_mqtt_client()
self._mqtt_client.on_connect = value

@property
def on_mqtt_disconnect(self):
def on_mqtt_disconnect(self) -> Optional[Callable]:
"""
Get or Set the MQTT Disconnect Handler

Expand All @@ -168,11 +183,11 @@ def on_mqtt_disconnect(self):
return None

@on_mqtt_disconnect.setter
def on_mqtt_disconnect(self, value):
def on_mqtt_disconnect(self, value: Callable) -> None:
self._get_mqtt_client().on_disconnect = value

@property
def on_mqtt_subscribe(self):
def on_mqtt_subscribe(self) -> Optional[Callable]:
"""
Get or Set the MQTT Subscribe Handler

Expand All @@ -182,11 +197,11 @@ def on_mqtt_subscribe(self):
return None

@on_mqtt_subscribe.setter
def on_mqtt_subscribe(self, value):
def on_mqtt_subscribe(self, value: Callable) -> None:
self._get_mqtt_client().on_subscribe = value

@property
def on_mqtt_unsubscribe(self):
def on_mqtt_unsubscribe(self) -> Optional[Callable]:
"""
Get or Set the MQTT Unsubscribe Handler

Expand All @@ -196,11 +211,11 @@ def on_mqtt_unsubscribe(self):
return None

@on_mqtt_unsubscribe.setter
def on_mqtt_unsubscribe(self, value):
def on_mqtt_unsubscribe(self, value: Callable) -> None:
self._get_mqtt_client().on_unsubscribe = value

@property
def on_mqtt_message(self):
def on_mqtt_message(self) -> Optional[Callable]:
"""
Get or Set the MQTT Message Handler

Expand All @@ -210,17 +225,17 @@ def on_mqtt_message(self):
return None

@on_mqtt_message.setter
def on_mqtt_message(self, value):
def on_mqtt_message(self, value: Callable) -> None:
self._get_mqtt_client().on_message = value

@property
def enabled(self):
def enabled(self) -> bool:
"""
Get or Set whether the WiFi is enabled

"""
return self._wifi.enabled

@enabled.setter
def enabled(self, value):
def enabled(self, value: bool) -> None:
self._wifi.enabled = bool(value)
Loading