Skip to content

Add type hints #24

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 11 commits into from
Nov 17, 2021
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