Compare commits
No commits in common. "main" and "old" have entirely different histories.
22 changed files with 331 additions and 3915 deletions
4
.gitignore
vendored
4
.gitignore
vendored
|
|
@ -1,4 +1,2 @@
|
|||
__pycache__
|
||||
/env
|
||||
/out
|
||||
config.toml
|
||||
/config.toml
|
||||
|
|
|
|||
|
|
@ -1,15 +0,0 @@
|
|||
repos:
|
||||
- repo: https://github.com/PyCQA/isort
|
||||
rev: 5.12.0
|
||||
hooks:
|
||||
- id: isort
|
||||
args: ["--profile", "black"]
|
||||
- repo: https://github.com/psf/black
|
||||
rev: 23.3.0
|
||||
hooks:
|
||||
- id: black
|
||||
- repo: https://github.com/PyCQA/flake8
|
||||
rev: "6.0.0"
|
||||
hooks:
|
||||
- id: flake8
|
||||
args: ["--max-line-length=88", "--extend-ignore=E203"]
|
||||
|
|
@ -1,2 +0,0 @@
|
|||
# Oin
|
||||
Use Raspberry Pi with Sense Hat as interface for Home Assistant Thermostat.
|
||||
|
|
@ -1,14 +0,0 @@
|
|||
[logging]
|
||||
level = "INFO"
|
||||
|
||||
[homeassistant]
|
||||
entity = "climate.chaudiere"
|
||||
secondary_entities = [
|
||||
"sensor.esptic_tempo",
|
||||
"sensor.rte_tempo_prochaine_couleur",
|
||||
]
|
||||
|
||||
[homeassistant.mqtt]
|
||||
username = "oin"
|
||||
password = ""
|
||||
host = "homeassistant.local"
|
||||
23
oin/__main__.py
Normal file
23
oin/__main__.py
Normal file
|
|
@ -0,0 +1,23 @@
|
|||
import tomllib
|
||||
from datetime import date, datetime, timedelta, timezone
|
||||
import json
|
||||
from threading import Timer
|
||||
from sense_hat import SenseHat
|
||||
from signal import pause
|
||||
from urllib import request, parse
|
||||
|
||||
from .tempo import TempoAPI
|
||||
from .logger import SenseLogger
|
||||
from .info import InfoPanel
|
||||
|
||||
with open("config.toml", "rb") as config_file:
|
||||
config = tomllib.load(config_file)
|
||||
|
||||
sense = SenseHat()
|
||||
tempo_api = TempoAPI(config.get("rte_api_key"))
|
||||
|
||||
|
||||
sense_logger = SenseLogger(sense, "dbname=tsdb user=edpibu host=/run/postgresql")
|
||||
info_panel = InfoPanel(sense, tempo_api)
|
||||
|
||||
sense.show_message(">")
|
||||
65
oin/info/__init__.py
Normal file
65
oin/info/__init__.py
Normal file
|
|
@ -0,0 +1,65 @@
|
|||
from threading import Thread, Timer
|
||||
from datetime import date
|
||||
|
||||
from .utils import num_to_pixels
|
||||
from ..tempo import convert_color
|
||||
|
||||
|
||||
class InfoPanel:
|
||||
def __init__(self, sense, tempo_api, rotation=0):
|
||||
self.sense = sense
|
||||
self.stick = self.sense.stick
|
||||
self.rotation = rotation
|
||||
self.sense.set_rotation(self.rotation)
|
||||
|
||||
self.tempo_api = tempo_api
|
||||
|
||||
self.start_stick_info()
|
||||
self.clear_timer = Timer(5, self.sense.clear)
|
||||
|
||||
def start_stick_info(self):
|
||||
self.thread = Thread(target=self.stick_info)
|
||||
self.thread.start()
|
||||
|
||||
def start_clear_timer(self):
|
||||
self.clear_timer.cancel()
|
||||
self.clear_timer = Timer(5, self.sense.clear)
|
||||
self.clear_timer.start()
|
||||
|
||||
def stick_info(self):
|
||||
event = self.stick.wait_for_event(emptybuffer=True)
|
||||
self.start_stick_info()
|
||||
|
||||
match event.direction:
|
||||
case "up":
|
||||
self.show_trash()
|
||||
case "right":
|
||||
self.show_temp()
|
||||
case "left":
|
||||
self.show_humidity()
|
||||
case "down":
|
||||
self.show_tempo()
|
||||
case "middle":
|
||||
self.sense.clear()
|
||||
|
||||
self.start_clear_timer()
|
||||
|
||||
def show_trash(self):
|
||||
today = date.today().isocalendar()
|
||||
if (today.week % 2) ^ (today.weekday >= 4):
|
||||
self.sense.set_pixels(64 * [[255, 255, 0]])
|
||||
else:
|
||||
grid_line = 4 * [[0, 127, 0], [0, 15, 0]]
|
||||
self.sense.set_pixels(4 * (grid_line + grid_line[::-1]))
|
||||
|
||||
def show_temp(self):
|
||||
self.sense.set_pixels(
|
||||
num_to_pixels(self.sense.temperature, color=[255, 0, 255])
|
||||
)
|
||||
|
||||
def show_humidity(self):
|
||||
self.sense.set_pixels(num_to_pixels(self.sense.humidity, color=[0, 255, 255]))
|
||||
|
||||
def show_tempo(self):
|
||||
tempo_colors = [convert_color(c) for c in self.tempo_api.tempo]
|
||||
self.sense.set_pixels([tempo_colors[1]] * 48 + [tempo_colors[0]] * 16)
|
||||
133
oin/info/utils.py
Normal file
133
oin/info/utils.py
Normal file
|
|
@ -0,0 +1,133 @@
|
|||
from itertools import chain
|
||||
|
||||
|
||||
def digit_to_pixels(n, color=[255, 255, 255], bg_color=[0, 0, 0]):
|
||||
X = color
|
||||
O = bg_color
|
||||
match n:
|
||||
case "0":
|
||||
return [
|
||||
[O, O, O, O],
|
||||
[O, X, X, O],
|
||||
[X, O, O, X],
|
||||
[X, O, O, X],
|
||||
[X, O, O, X],
|
||||
[X, O, O, X],
|
||||
[O, X, X, O],
|
||||
[O, O, O, O],
|
||||
]
|
||||
case "1":
|
||||
return [
|
||||
[O, O, O, O],
|
||||
[O, O, O, X],
|
||||
[O, O, X, X],
|
||||
[O, X, O, X],
|
||||
[O, O, O, X],
|
||||
[O, O, O, X],
|
||||
[O, O, O, X],
|
||||
[O, O, O, O],
|
||||
]
|
||||
case "2":
|
||||
return [
|
||||
[O, O, O, O],
|
||||
[O, X, X, O],
|
||||
[X, O, O, X],
|
||||
[O, O, O, X],
|
||||
[O, O, X, O],
|
||||
[O, X, O, O],
|
||||
[X, X, X, X],
|
||||
[O, O, O, O],
|
||||
]
|
||||
case "3":
|
||||
return [
|
||||
[O, O, O, O],
|
||||
[X, X, X, X],
|
||||
[O, O, O, X],
|
||||
[O, O, X, O],
|
||||
[O, O, O, X],
|
||||
[X, O, O, X],
|
||||
[O, X, X, O],
|
||||
[O, O, O, O],
|
||||
]
|
||||
case "4":
|
||||
return [
|
||||
[O, O, O, O],
|
||||
[O, O, O, X],
|
||||
[O, O, X, X],
|
||||
[O, X, O, X],
|
||||
[X, O, X, X],
|
||||
[X, X, X, X],
|
||||
[O, O, O, X],
|
||||
[O, O, O, O],
|
||||
]
|
||||
case "5":
|
||||
return [
|
||||
[O, O, O, O],
|
||||
[X, X, X, X],
|
||||
[X, O, O, O],
|
||||
[X, X, X, O],
|
||||
[O, O, O, X],
|
||||
[X, O, O, X],
|
||||
[O, X, X, O],
|
||||
[O, O, O, O],
|
||||
]
|
||||
case "6":
|
||||
return [
|
||||
[O, O, O, O],
|
||||
[O, X, X, X],
|
||||
[X, O, O, O],
|
||||
[X, X, X, O],
|
||||
[X, O, O, X],
|
||||
[X, O, O, X],
|
||||
[O, X, X, O],
|
||||
[O, O, O, O],
|
||||
]
|
||||
case "7":
|
||||
return [
|
||||
[O, O, O, O],
|
||||
[X, X, X, X],
|
||||
[O, O, O, X],
|
||||
[O, O, X, O],
|
||||
[O, X, O, O],
|
||||
[X, O, O, O],
|
||||
[X, O, O, O],
|
||||
[O, O, O, O],
|
||||
]
|
||||
case "8":
|
||||
return [
|
||||
[O, O, O, O],
|
||||
[O, X, X, O],
|
||||
[X, O, O, X],
|
||||
[O, X, X, O],
|
||||
[X, O, O, X],
|
||||
[X, O, O, X],
|
||||
[O, X, X, O],
|
||||
[O, O, O, O],
|
||||
]
|
||||
case "9":
|
||||
return [
|
||||
[O, O, O, O],
|
||||
[O, X, X, O],
|
||||
[X, O, O, X],
|
||||
[X, O, O, X],
|
||||
[O, X, X, X],
|
||||
[O, O, O, X],
|
||||
[O, X, X, O],
|
||||
[O, O, O, O],
|
||||
]
|
||||
case _:
|
||||
return 8 * [4 * [O]]
|
||||
|
||||
|
||||
def num_to_pixels(n, color=[255, 255, 255], bg_color=[0, 0, 0]):
|
||||
digits = f"{n:02.0f}"
|
||||
if len(digits) > 2:
|
||||
digits = "XX"
|
||||
|
||||
return list(
|
||||
chain.from_iterable(
|
||||
chain.from_iterable(
|
||||
zip(*[digit_to_pixels(i, color, bg_color) for i in digits])
|
||||
)
|
||||
)
|
||||
)
|
||||
40
oin/logger/__init__.py
Normal file
40
oin/logger/__init__.py
Normal file
|
|
@ -0,0 +1,40 @@
|
|||
from datetime import date, datetime, timedelta, timezone
|
||||
from threading import Timer
|
||||
import psycopg
|
||||
from psycopg_pool import ConnectionPool
|
||||
|
||||
|
||||
class SenseLogger:
|
||||
def __init__(self, sense, conninfo):
|
||||
self.sense = sense
|
||||
self.database_pool = ConnectionPool(conninfo)
|
||||
|
||||
self.timer = Timer(0, self.log)
|
||||
self.timer.start()
|
||||
|
||||
def log(self):
|
||||
self.timer = Timer(30, self.log)
|
||||
self.timer.start()
|
||||
|
||||
dt = datetime.now()
|
||||
|
||||
temp = self.sense.temperature
|
||||
pres = self.sense.pressure
|
||||
humi = self.sense.humidity
|
||||
|
||||
print(dt)
|
||||
INS = "INSERT INTO sensor_data (time, sensor, value) VALUES (%s, %s, %s);"
|
||||
|
||||
conn = self.database_pool.getconn()
|
||||
try:
|
||||
with conn.cursor() as cursor:
|
||||
cursor.execute(INS, (dt, "temp", temp))
|
||||
cursor.execute(INS, (dt, "pres", pres))
|
||||
cursor.execute(INS, (dt, "humi", humi))
|
||||
|
||||
conn.commit()
|
||||
except:
|
||||
conn.rollback()
|
||||
raise
|
||||
finally:
|
||||
self.database_pool.putconn(conn)
|
||||
69
oin/tempo/__init__.py
Normal file
69
oin/tempo/__init__.py
Normal file
|
|
@ -0,0 +1,69 @@
|
|||
from datetime import date, datetime, timedelta, timezone
|
||||
from threading import Timer
|
||||
from urllib import request, parse
|
||||
import json
|
||||
|
||||
|
||||
class TempoAPI:
|
||||
def __init__(self, api_key):
|
||||
self.rte_api_key = api_key
|
||||
self.update_rte_access_token()
|
||||
|
||||
def update_rte_access_token(self):
|
||||
with request.urlopen(
|
||||
request.Request(
|
||||
"https://digital.iservices.rte-france.com/token/oauth/",
|
||||
headers={
|
||||
"Authorization": f"Basic {self.rte_api_key}",
|
||||
"Content-Type": "application/x-www-form-urlencoded",
|
||||
},
|
||||
method="POST",
|
||||
)
|
||||
) as res:
|
||||
data = json.load(res)
|
||||
|
||||
self.rte_access_token = data.get("access_token")
|
||||
self.renewal_timer = Timer(data.get("expires_in"), self.update_rte_access_token)
|
||||
self.renewal_timer.start()
|
||||
|
||||
@property
|
||||
def tempo(self):
|
||||
today = datetime.combine(date.today(), datetime.min.time()).astimezone()
|
||||
url = parse.urljoin(
|
||||
"https://digital.iservices.rte-france.com/open_api/tempo_like_supply_contract/v1/tempo_like_calendars",
|
||||
"?"
|
||||
+ parse.urlencode(
|
||||
{
|
||||
"start_date": today.isoformat(),
|
||||
"end_date": (today + timedelta(days=2)).isoformat(),
|
||||
},
|
||||
safe=":",
|
||||
),
|
||||
)
|
||||
with request.urlopen(
|
||||
request.Request(
|
||||
url,
|
||||
headers={
|
||||
"Authorization": f"Bearer {self.rte_access_token}",
|
||||
"Accept": "application/json",
|
||||
},
|
||||
)
|
||||
) as res:
|
||||
api_data = json.load(res)
|
||||
|
||||
return [
|
||||
val.get("value")
|
||||
for val in api_data.get("tempo_like_calendars").get("values")
|
||||
]
|
||||
|
||||
|
||||
def convert_color(color):
|
||||
match color:
|
||||
case "BLUE":
|
||||
return [0, 0, 255]
|
||||
case "WHITE":
|
||||
return [255, 255, 255]
|
||||
case "RED":
|
||||
return [255, 0, 0]
|
||||
case _:
|
||||
return [255, 255, 0]
|
||||
|
|
@ -1,41 +0,0 @@
|
|||
import logging
|
||||
import sys
|
||||
import tomllib
|
||||
|
||||
from .mqtt import HAClient
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
config_path = "config.toml"
|
||||
|
||||
|
||||
def main() -> int:
|
||||
with open(config_path, "rb") as config_file:
|
||||
config = tomllib.load(config_file)
|
||||
|
||||
logging.basicConfig(**config.get("logging", dict()))
|
||||
|
||||
ha_config = config.get("homeassistant")
|
||||
if ha_config is None:
|
||||
logger.error(f"Missing home assistant config in <{config_path}>")
|
||||
logger.error(f"\t{config}")
|
||||
return 1
|
||||
|
||||
client = HAClient(
|
||||
ha_config.get("entity"),
|
||||
ha_config.get("secondary_entities"),
|
||||
mqtt_config=ha_config.get("mqtt"),
|
||||
)
|
||||
|
||||
code = client.connect()
|
||||
if code != 0:
|
||||
return 1
|
||||
|
||||
code = client.loop()
|
||||
if code != 0:
|
||||
return 1
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
|
|
@ -1,203 +0,0 @@
|
|||
import json
|
||||
import logging
|
||||
from collections.abc import Callable
|
||||
from typing import Any, TypedDict
|
||||
|
||||
import paho.mqtt.client as mqtt
|
||||
from paho.mqtt.enums import CallbackAPIVersion, MQTTErrorCode
|
||||
|
||||
from .screen import Screen
|
||||
from .select import Selector
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class HAOptions(TypedDict):
|
||||
dev: dict[str, str]
|
||||
o: dict[str, str]
|
||||
availability_topic: str
|
||||
state_topic: str
|
||||
cmps: dict[str, dict[str, str]]
|
||||
|
||||
|
||||
class HAClient:
|
||||
def __init__(
|
||||
self,
|
||||
entity: str,
|
||||
secondary_entities: list[str] = [],
|
||||
mqtt_config: dict[str, str] = dict(),
|
||||
) -> None:
|
||||
self.entity = entity
|
||||
self.secondary_entities = secondary_entities
|
||||
self.config = mqtt_config
|
||||
|
||||
self.state_topic = "oin/state"
|
||||
self.availability_topic = "oin/availability"
|
||||
|
||||
self.client = mqtt.Client(CallbackAPIVersion.VERSION2)
|
||||
username = self.config.get("username", None)
|
||||
logger.debug(f"Setting up MQTT with user <{username}>")
|
||||
self.client.username_pw_set(
|
||||
username=username,
|
||||
password=self.config.get("password", None),
|
||||
)
|
||||
|
||||
self.screen = Screen()
|
||||
self.selector = Selector(self.send_data)
|
||||
|
||||
@property
|
||||
def ha_options(self) -> HAOptions:
|
||||
return {
|
||||
"dev": {
|
||||
"ids": "oin",
|
||||
"name": "Oin",
|
||||
},
|
||||
"o": {
|
||||
"name": "Oin",
|
||||
},
|
||||
"availability_topic": self.availability_topic,
|
||||
"state_topic": self.state_topic,
|
||||
"cmps": self.selector.ha_options,
|
||||
}
|
||||
|
||||
def connect(self) -> int:
|
||||
self.client.will_set(self.availability_topic, "offline", retain=True)
|
||||
|
||||
host = self.config.get("host")
|
||||
port = self.config.get("port", 1883)
|
||||
|
||||
if host is None:
|
||||
logger.error("Host not found in config.")
|
||||
logger.error(f"\t{self.config}")
|
||||
return 1
|
||||
if not isinstance(port, int):
|
||||
logger.warning(f"Invalid port config : <{port}> ; using port 1883.")
|
||||
port = 1883
|
||||
|
||||
logger.debug(f"Connecting to <{host}> on port <{port}>.")
|
||||
code = self.client.connect(host, port)
|
||||
if code != 0:
|
||||
logger.error(f"Could not connect to host <{host}> on port <{port}>.")
|
||||
return 1
|
||||
|
||||
code = self.subscribe(entity_topic(self.entity), self.state_update)
|
||||
if code != 0:
|
||||
return 1
|
||||
code = self.subscribe(
|
||||
[entity_topic(entity) for entity in self.secondary_entities],
|
||||
self.secondary_state_update,
|
||||
)
|
||||
if code != 0:
|
||||
return 1
|
||||
|
||||
m_info = self.publish_json(
|
||||
"homeassistant/device/oin/config", self.ha_options, retain=True
|
||||
)
|
||||
m_info.wait_for_publish(60)
|
||||
if not m_info.is_published():
|
||||
logger.error("Config message timed out")
|
||||
return 1
|
||||
|
||||
m_info = self.publish(self.availability_topic, "online", retain=True)
|
||||
m_info.wait_for_publish(60)
|
||||
if not m_info.is_published():
|
||||
logger.error("Availability message timed out")
|
||||
return 1
|
||||
|
||||
logger.info("Connected to Home Assistant.")
|
||||
return 0
|
||||
|
||||
def publish(
|
||||
self, topic: str, data: str, retain: bool = False
|
||||
) -> mqtt.MQTTMessageInfo:
|
||||
logger.debug(f"Sending message on topic <{topic}>.")
|
||||
return self.client.publish(topic, data, retain=retain)
|
||||
|
||||
def publish_json(
|
||||
self, topic: str, data: Any, retain: bool = False
|
||||
) -> mqtt.MQTTMessageInfo:
|
||||
return self.publish(topic, json.dumps(data), retain)
|
||||
|
||||
def subscribe(
|
||||
self,
|
||||
topic: str | list[str],
|
||||
callback: Callable[[mqtt.Client, Any, mqtt.MQTTMessage], None],
|
||||
) -> MQTTErrorCode:
|
||||
logger.debug(f"Subscribing to <{topic}>.")
|
||||
|
||||
match topic:
|
||||
case str():
|
||||
self.client.message_callback_add(topic, callback)
|
||||
code, _ = self.client.subscribe(topic)
|
||||
case list():
|
||||
for top in topic:
|
||||
self.client.message_callback_add(top, callback)
|
||||
code, _ = self.client.subscribe([(top, 0) for top in topic])
|
||||
|
||||
if code != 0:
|
||||
logger.error(f"Failed subscribing to topic <{topic}> with code <{code}>.")
|
||||
return code
|
||||
|
||||
def loop(self) -> MQTTErrorCode:
|
||||
logger.info("Starting MQTT client loop.")
|
||||
code = self.client.loop_forever(retry_first_connection=True)
|
||||
|
||||
if code != 0:
|
||||
logger.error("MQTT client loop failed with code <{code}>.")
|
||||
else:
|
||||
logger.info("MQTT client loop successfully exited")
|
||||
return code
|
||||
|
||||
def state_update(
|
||||
self, client: mqtt.Client, userdata: Any, message: mqtt.MQTTMessage
|
||||
) -> None:
|
||||
data = message.payload.decode()
|
||||
logger.debug(f"Message received on topic <{message.topic}>: {data}.")
|
||||
|
||||
subtopic = message.topic.rsplit("/", maxsplit=1)[1]
|
||||
|
||||
match subtopic:
|
||||
case "current_temperature":
|
||||
self.screen.value = json.loads(data)
|
||||
case "temperature":
|
||||
if (value := json.loads(data)) != self.selector.temperature:
|
||||
self.screen.tmp_value = value
|
||||
self.selector.temperature = value
|
||||
case "hvac_action":
|
||||
self.screen.mode = json.loads(data)
|
||||
case "preset_modes":
|
||||
if (value := json.loads(data)) != self.selector.preset_modes:
|
||||
self.selector.preset_modes = value
|
||||
case "preset_mode":
|
||||
if (value := json.loads(data)) != self.selector.mode:
|
||||
self.selector.mode = value
|
||||
case "state":
|
||||
match data:
|
||||
case "heat":
|
||||
self.selector.switch = True
|
||||
case "off":
|
||||
self.selector.switch = False
|
||||
case other:
|
||||
logger.warning(f"Unknown state received: <{other}>.")
|
||||
case _:
|
||||
pass
|
||||
|
||||
def secondary_state_update(
|
||||
self, client: mqtt.Client, userdata: Any, message: mqtt.MQTTMessage
|
||||
) -> None:
|
||||
data = message.payload.decode()
|
||||
logger.debug(f"Message received on topic <{message.topic}>: {data}.")
|
||||
|
||||
_, grp, ent, subtopic = message.topic.split("/")
|
||||
idx = self.secondary_entities.index(f"{grp}.{ent}")
|
||||
|
||||
if subtopic == "state":
|
||||
self.screen.secondary |= {idx: data}
|
||||
|
||||
def send_data(self, data: Any) -> mqtt.MQTTMessageInfo:
|
||||
return self.publish_json(self.state_topic, data)
|
||||
|
||||
|
||||
def entity_topic(entity: str, subtopic: str = "#") -> str:
|
||||
topic = entity.replace(".", "/")
|
||||
return f"homeassistant/{topic}/{subtopic}"
|
||||
|
|
@ -1,191 +0,0 @@
|
|||
import logging
|
||||
import math
|
||||
from threading import Thread, Timer
|
||||
|
||||
import bdfparser
|
||||
from sense_hat.sense_hat import SenseHat
|
||||
from sense_hat.stick import InputEvent
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
COLORS = {
|
||||
"Bleu": (0, 0, 255),
|
||||
"Blanc": (255, 255, 255),
|
||||
"Rouge": (255, 0, 0),
|
||||
"Verte": (0, 127, 31),
|
||||
"Jaune": (255, 255, 0),
|
||||
"heat": (255, 0, 0),
|
||||
"heating": (255, 0, 0),
|
||||
"idle": (127, 0, 255),
|
||||
"off": (127, 127, 127),
|
||||
"on_setting": (255, 255, 0),
|
||||
"off_setting": (255, 255, 255),
|
||||
None: (0, 0, 0),
|
||||
}
|
||||
|
||||
|
||||
class Screen:
|
||||
def __init__(self) -> None:
|
||||
self.sense = SenseHat()
|
||||
self._value = ""
|
||||
self._tmp = False
|
||||
self._tmp_value = ""
|
||||
self._mode = ""
|
||||
self.font = bdfparser.Font("src/tom-thumb.bdf")
|
||||
self._secondary: dict[int, str] = dict()
|
||||
self._secondary_pixels: list[tuple[int, int, int]] = [(0, 0, 0)] * 8
|
||||
|
||||
self.timer = Timer(0, self.set_pixels)
|
||||
self.auto_dim = AutoDim(self.sense)
|
||||
self.auto_dim.start()
|
||||
|
||||
self._held = False
|
||||
self.sense.stick.direction_middle = self.stick_click
|
||||
self.sense.stick.direction_any = self.auto_dim.undim
|
||||
|
||||
@property
|
||||
def value(self) -> None | str:
|
||||
return self._value
|
||||
|
||||
@value.setter
|
||||
def value(self, value: float) -> None:
|
||||
logger.debug(f"Updated value: <{value}>")
|
||||
|
||||
self._value = format_value(value)
|
||||
if not self._tmp:
|
||||
self.set_pixels()
|
||||
|
||||
@property
|
||||
def color(self) -> tuple[int, int, int]:
|
||||
return COLORS.get(self.mode, (0, 0, 0))
|
||||
|
||||
@property
|
||||
def mode(self) -> str:
|
||||
return self._mode
|
||||
|
||||
@mode.setter
|
||||
def mode(self, value: str) -> None:
|
||||
self._mode = value
|
||||
if not self._tmp:
|
||||
self.set_pixels()
|
||||
|
||||
@property
|
||||
def tmp_value(self) -> None | str:
|
||||
return self._tmp_value
|
||||
|
||||
@tmp_value.setter
|
||||
def tmp_value(self, value: float) -> None:
|
||||
logger.debug(f"Show value: <{value}>")
|
||||
|
||||
self.timer.cancel()
|
||||
self._tmp_value = format_value(value)
|
||||
self.show_tmp()
|
||||
|
||||
def show_tmp(self) -> None:
|
||||
self._tmp = True
|
||||
self.set_pixels(
|
||||
self.tmp_value,
|
||||
color=COLORS.get("off_setting" if self.mode == "off" else "on_setting"),
|
||||
)
|
||||
self.timer = Timer(3, self.set_pixels)
|
||||
self.timer.start()
|
||||
|
||||
def set_pixels(
|
||||
self,
|
||||
value: str | None = None,
|
||||
color: tuple[int, int, int] | None = None,
|
||||
bg_color: tuple[int, int, int] = (0, 0, 0),
|
||||
) -> None:
|
||||
if value is None:
|
||||
value = self.value
|
||||
self._tmp = False
|
||||
if color is None:
|
||||
color = self.color
|
||||
|
||||
if value:
|
||||
pixels = [color if x else bg_color for x in self.data_from_value(value)]
|
||||
else:
|
||||
pixels = 48 * [(0, 0, 0)]
|
||||
pixels += self.secondary_pixels
|
||||
|
||||
self.sense.set_pixels(pixels)
|
||||
|
||||
@property
|
||||
def secondary(self) -> dict[int, str]:
|
||||
return self._secondary
|
||||
|
||||
@secondary.setter
|
||||
def secondary(self, value: dict[int, str]) -> None:
|
||||
self._secondary = value
|
||||
|
||||
for idx in range(2):
|
||||
self._secondary_pixels[4 * idx : 4 * (idx + 1)] = [
|
||||
COLORS.get(value.get(idx, None), (0, 0, 0))
|
||||
] * 4
|
||||
|
||||
if not self._tmp:
|
||||
self.set_pixels()
|
||||
|
||||
@property
|
||||
def secondary_pixels(self) -> list[tuple[int, int, int]]:
|
||||
return self._secondary_pixels
|
||||
|
||||
def stick_click(self, event: InputEvent) -> None:
|
||||
match (event.action, self._held):
|
||||
case ("held", False):
|
||||
self._held = True
|
||||
case ("released", True):
|
||||
self._held = False
|
||||
case ("released", False):
|
||||
self.show_tmp()
|
||||
case _:
|
||||
pass
|
||||
|
||||
def data_from_value(self, value: str) -> list[int]:
|
||||
return self.font.draw(value, mode=0).crop(8, 7).todata(3)
|
||||
|
||||
|
||||
class AutoDim(Thread):
|
||||
def __init__(self, sense: SenseHat):
|
||||
super().__init__()
|
||||
|
||||
self.daemon = True
|
||||
self.sense = sense
|
||||
self.dim = False
|
||||
self.switching = False
|
||||
self.sense.gamma_reset()
|
||||
|
||||
def run(self) -> None:
|
||||
while True:
|
||||
self.auto_dim()
|
||||
|
||||
def auto_dim(self) -> None:
|
||||
accel_z = self.sense.get_accelerometer_raw()["z"]
|
||||
|
||||
if not self.switching and accel_z < 0.2:
|
||||
self.switching = True
|
||||
self.dim = not self.dim
|
||||
|
||||
elif self.switching and accel_z > 0.9:
|
||||
self.switching = False
|
||||
|
||||
@property
|
||||
def dim(self) -> bool:
|
||||
return self._dim
|
||||
|
||||
@dim.setter
|
||||
def dim(self, value: bool) -> None:
|
||||
if value:
|
||||
self.sense.gamma = [0] * 32
|
||||
else:
|
||||
self.sense.gamma_reset()
|
||||
self._dim = value
|
||||
|
||||
def undim(self) -> None:
|
||||
self.dim = False
|
||||
|
||||
|
||||
def format_value(value: float) -> str:
|
||||
v = math.trunc(value)
|
||||
d = "." if (value - v) >= 0.5 else ""
|
||||
return f"{v}{d}"
|
||||
|
|
@ -1,100 +0,0 @@
|
|||
import logging
|
||||
import math
|
||||
from collections.abc import Callable
|
||||
from typing import Any
|
||||
|
||||
from sense_hat.sense_hat import SenseHat
|
||||
from sense_hat.stick import ACTION_HELD, ACTION_RELEASED, InputEvent
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Selector:
|
||||
default_data = {"temperature": None, "mode": None, "switch": None}
|
||||
|
||||
def __init__(self, send_data: Callable[[dict[str, Any]], Any]):
|
||||
self.stick = SenseHat().stick
|
||||
self.temperature: float | None = None
|
||||
self.mode: str | None = None
|
||||
self.switch: bool | None = None
|
||||
self.preset_modes: list[str] = []
|
||||
self.send_data = send_data
|
||||
self.switch_held = False
|
||||
|
||||
self.stick.direction_middle = self.toggle
|
||||
|
||||
self.stick.direction_up = self.increase_temperature
|
||||
self.stick.direction_down = self.decrease_temperature
|
||||
|
||||
self.stick.direction_right = self.next_mode
|
||||
self.stick.direction_left = self.prev_mode
|
||||
|
||||
@property
|
||||
def ha_options(self) -> dict[str, dict[str, str]]:
|
||||
return {
|
||||
"temperature": {
|
||||
"p": "sensor",
|
||||
"value_template": "{{ value_json.temperature }}",
|
||||
"unique_id": "oin_temp",
|
||||
"device_class": "temperature",
|
||||
"entity_category": "diagnostic",
|
||||
"icon": "mdi:thermostat",
|
||||
"unit_of_measurement": "°C",
|
||||
},
|
||||
"mode": {
|
||||
"p": "sensor",
|
||||
"name": "Mode",
|
||||
"value_template": "{{ value_json.mode }}",
|
||||
"unique_id": "oin_mode",
|
||||
"entity_category": "diagnostic",
|
||||
"icon": "mdi:thermostat-auto",
|
||||
},
|
||||
"switch": {
|
||||
"p": "sensor",
|
||||
"name": "Switch",
|
||||
"value_template": "{{ value_json.switch }}",
|
||||
"unique_id": "oin_switch",
|
||||
"entity_category": "diagnostic",
|
||||
"icon": "mdi:thermostat-auto",
|
||||
},
|
||||
}
|
||||
|
||||
def increase_temperature(self, event: InputEvent) -> None:
|
||||
if event.action != ACTION_RELEASED and self.temperature is not None:
|
||||
self.callback({"temperature": math.floor(self.temperature * 2) / 2 + 0.5})
|
||||
|
||||
def decrease_temperature(self, event: InputEvent) -> None:
|
||||
if event.action != ACTION_RELEASED and self.temperature is not None:
|
||||
self.callback({"temperature": math.ceil(self.temperature * 2) / 2 - 0.5})
|
||||
|
||||
def next_mode(self, event: InputEvent) -> None:
|
||||
if event.action != ACTION_RELEASED and self.mode is not None:
|
||||
self.callback(
|
||||
{
|
||||
"mode": self.preset_modes[
|
||||
(self.preset_modes.index(self.mode) + 1)
|
||||
% len(self.preset_modes)
|
||||
]
|
||||
}
|
||||
)
|
||||
|
||||
def prev_mode(self, event: InputEvent) -> None:
|
||||
if event.action != ACTION_RELEASED and self.mode is not None:
|
||||
self.callback(
|
||||
{
|
||||
"mode": self.preset_modes[
|
||||
(self.preset_modes.index(self.mode) - 1)
|
||||
% len(self.preset_modes)
|
||||
]
|
||||
}
|
||||
)
|
||||
|
||||
def toggle(self, event: InputEvent) -> None:
|
||||
if not self.switch_held and event.action == ACTION_HELD:
|
||||
self.switch_held = True
|
||||
self.callback({"switch": "off" if self.switch else "heat"})
|
||||
elif self.switch_held and event.action == ACTION_RELEASED:
|
||||
self.switch_held = False
|
||||
|
||||
def callback(self, data: dict[str, Any]) -> None:
|
||||
self.send_data(self.default_data | data)
|
||||
|
|
@ -1,5 +0,0 @@
|
|||
{
|
||||
"include": ["oin_thermostat"],
|
||||
"strict": ["oin_thermostat"],
|
||||
"venvPath": "env"
|
||||
}
|
||||
2353
src/tom-thumb.bdf
2353
src/tom-thumb.bdf
File diff suppressed because it is too large
Load diff
|
|
@ -1,12 +0,0 @@
|
|||
"""
|
||||
This type stub file was generated by pyright.
|
||||
"""
|
||||
|
||||
from .bdfparser import *
|
||||
|
||||
"""
|
||||
BDF (Glyph Bitmap Distribution Format) Bitmap Font File Parser in Python
|
||||
Copyright (c) 2017-2021 Tom CHEN (tomchen.org), MIT License
|
||||
https://font.tomchen.org/bdfparser_py/
|
||||
"""
|
||||
__version__ = ...
|
||||
|
|
@ -1,345 +0,0 @@
|
|||
"""
|
||||
This type stub file was generated by pyright.
|
||||
"""
|
||||
from typing import Any
|
||||
|
||||
def format_warning(message, category, filename, lineno, file=..., line=...): ...
|
||||
|
||||
class Font:
|
||||
"""
|
||||
`Font` object
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/font
|
||||
"""
|
||||
|
||||
__PATTERN_VVECTOR_DELIMITER = ...
|
||||
__META_TITLES = ...
|
||||
__EMPTY_GLYPH = ...
|
||||
def __init__(self, *argv) -> None:
|
||||
"""
|
||||
Initialize a `Font` object. Load the BDF font file if a file path string or a file object is present.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/font#font
|
||||
"""
|
||||
...
|
||||
def load_file_path(self, file_path): # -> Self:
|
||||
"""
|
||||
Load the BDF font file in the file path.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/font#load_file_path
|
||||
"""
|
||||
...
|
||||
def load_file_obj(self, file_obj): # -> Self:
|
||||
"""
|
||||
Load the BDF font file object.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/font#load_file_obj
|
||||
"""
|
||||
...
|
||||
def length(self): # -> int:
|
||||
"""
|
||||
Returns how many glyphs actually exist in the font.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/font#length
|
||||
"""
|
||||
...
|
||||
def __len__(self): # -> int:
|
||||
"""
|
||||
Same as `.length()`
|
||||
"""
|
||||
...
|
||||
def itercps(self, order=..., r=...): # -> filter[Any] | Iterator[Any]:
|
||||
"""
|
||||
Almost identical to `.iterglyphs()`, except it returns an `iterator` of glyph codepoints instead of an `iterator` of `Glyph` objects.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/font#itercps
|
||||
"""
|
||||
...
|
||||
def iterglyphs(self, order=..., r=...): # -> Generator[Glyph | None, Any, None]:
|
||||
"""
|
||||
Returns an iterator of all the glyphs (as `Glyph` objects) in the font (default) or in the specified codepoint range in the font, sorted by the specified order (or by the ascending codepoint order by default).
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/font#iterglyphs
|
||||
"""
|
||||
...
|
||||
def glyphbycp(self, codepoint): # -> Glyph | None:
|
||||
"""
|
||||
Get a glyph (as Glyph object) by its codepoint.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/font#glyphbycp
|
||||
"""
|
||||
...
|
||||
def glyph(self, character): # -> Glyph | None:
|
||||
"""
|
||||
Get a glyph (as `Glyph` object) by its character.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/font#glyph
|
||||
"""
|
||||
...
|
||||
def lacksglyphs(self, string): # -> list[Any] | None:
|
||||
"""
|
||||
Check if there is any missing glyph and gets these glyphs' character.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/font#lacksglyphs
|
||||
"""
|
||||
...
|
||||
def drawcps(
|
||||
self,
|
||||
cps,
|
||||
linelimit=...,
|
||||
mode=...,
|
||||
direction=...,
|
||||
usecurrentglyphspacing=...,
|
||||
missing=...,
|
||||
):
|
||||
"""
|
||||
Draw the glyphs of the specified codepoints, to a `Bitmap` object.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/font#drawcps
|
||||
"""
|
||||
...
|
||||
def draw(
|
||||
self,
|
||||
string: str,
|
||||
linelimit: int = ...,
|
||||
mode: int = ...,
|
||||
direction: str = ...,
|
||||
usecurrentglyphspacing: bool = ...,
|
||||
missing: dict[str, Any] | Glyph = ...,
|
||||
) -> Bitmap:
|
||||
"""
|
||||
Draw (render) the glyphs of the specified words / setences / paragraphs (as a `str`), to a `Bitmap` object.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/font#draw
|
||||
"""
|
||||
...
|
||||
def drawall(
|
||||
self,
|
||||
order=...,
|
||||
r=...,
|
||||
linelimit=...,
|
||||
mode=...,
|
||||
direction=...,
|
||||
usecurrentglyphspacing=...,
|
||||
):
|
||||
"""
|
||||
Draw all the glyphs in the font (default) or in the specified codepoint range in the font, sorted by the specified order (or by the ascending codepoint order by default), to a `Bitmap` object.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/font#drawall
|
||||
"""
|
||||
...
|
||||
|
||||
class Glyph:
|
||||
"""
|
||||
`Glyph` object
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/glyph
|
||||
"""
|
||||
|
||||
def __init__(self, meta_dict, font) -> None:
|
||||
"""
|
||||
Initialize a `Glyph` object. Load a `dict` of meta information and the font the glyph belongs.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/glyph#glyph
|
||||
"""
|
||||
...
|
||||
def __str__(self) -> str:
|
||||
"""
|
||||
Gets a human-readable (multi-line) `str` representation of the `Glyph` object.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/glyph#str-and-print
|
||||
"""
|
||||
...
|
||||
def __repr__(self): # -> str:
|
||||
"""
|
||||
Gets a programmer-readable `str` representation of the `Glyph` object.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/glyph#repr
|
||||
"""
|
||||
...
|
||||
def cp(self):
|
||||
"""
|
||||
Get the codepoint of the glyph.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/glyph#cp
|
||||
"""
|
||||
...
|
||||
def chr(self): # -> str:
|
||||
"""
|
||||
Get the character of the glyph.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/glyph#chr
|
||||
"""
|
||||
...
|
||||
def draw(self, mode=..., bb=...): # -> Bitmap:
|
||||
"""
|
||||
Draw the glyph to a `Bitmap` object.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/glyph#draw
|
||||
"""
|
||||
...
|
||||
def origin(
|
||||
self, mode=..., fromorigin=..., xoff=..., yoff=...
|
||||
): # -> tuple[Any, Any]:
|
||||
"""
|
||||
Get the relative position (displacement) of the origin from the left bottom corner of the bitmap drawn by the method `.draw()`, or vice versa.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/glyph#origin
|
||||
"""
|
||||
...
|
||||
|
||||
class Bitmap:
|
||||
"""
|
||||
`Bitmap` object
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/bitmap
|
||||
"""
|
||||
|
||||
def __init__(self, bin_bitmap_list) -> None:
|
||||
"""
|
||||
Initialize a `Bitmap` object. Load binary bitmap data (`list` of `str`s).
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/bitmap#bitmap
|
||||
"""
|
||||
...
|
||||
def __str__(self) -> str:
|
||||
"""
|
||||
Gets a human-readable (multi-line) `str` representation of the `Bitmap` object.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/bitmap#str-and-print
|
||||
"""
|
||||
...
|
||||
def __repr__(self): # -> str:
|
||||
"""
|
||||
Gets a programmer-readable (multi-line) `str` representation of the `Bitmap` object.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/bitmap#repr
|
||||
"""
|
||||
...
|
||||
def width(self): # -> int:
|
||||
"""
|
||||
Get the width of the bitmap.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/bitmap#width
|
||||
"""
|
||||
...
|
||||
def height(self): # -> int:
|
||||
"""
|
||||
Get the height of the bitmap.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/bitmap#height
|
||||
"""
|
||||
...
|
||||
def clone(self): # -> Self:
|
||||
"""
|
||||
Get a deep copy / clone of the `Bitmap` object.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/bitmap#clone
|
||||
"""
|
||||
...
|
||||
def crop(
|
||||
self, w: int, h: int, xoff: int = ..., yoff: int = ...
|
||||
) -> Bitmap: # -> Self:
|
||||
"""
|
||||
Crop and/or extend the bitmap.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/bitmap#crop
|
||||
"""
|
||||
...
|
||||
def overlay(self, bitmap): # -> Self:
|
||||
"""
|
||||
Overlay another bitmap over the current one.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/bitmap#overlay
|
||||
"""
|
||||
...
|
||||
@classmethod
|
||||
def concatall(
|
||||
cls, bitmaplist, direction=..., align=..., offsetlist=...
|
||||
): # -> Self:
|
||||
"""
|
||||
Concatenate all `Bitmap` objects in a `list`.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/bitmap#bitmapconcatall
|
||||
"""
|
||||
...
|
||||
def __add__(self, bitmap): # -> Self:
|
||||
"""
|
||||
`+` is a shortcut of `Bitmap.concatall()`. Use `+` to concatenate two `Bitmap` objects and get a new `Bitmap` objects.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/bitmap#-concat
|
||||
"""
|
||||
...
|
||||
def concat(self, bitmap, direction=..., align=..., offset=...): # -> Self:
|
||||
"""
|
||||
Concatenate another `Bitmap` objects to the current one.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/bitmap#concat
|
||||
"""
|
||||
...
|
||||
def enlarge(self, x=..., y=...): # -> Self:
|
||||
"""
|
||||
Enlarge a `Bitmap` object, by multiplying every pixel in x (right) direction and in y (top) direction.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/bitmap#enlarge
|
||||
"""
|
||||
...
|
||||
def __mul__(self, mul): # -> Self:
|
||||
"""
|
||||
`*` is a shortcut of `.enlarge()`.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/bitmap#-enlarge
|
||||
"""
|
||||
...
|
||||
def replace(self, substr, newsubstr): # -> Self:
|
||||
"""
|
||||
Replace a string by another in the bitmap.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/bitmap#replace
|
||||
"""
|
||||
...
|
||||
def shadow(self, xoff=..., yoff=...): # -> Self:
|
||||
"""
|
||||
Add shadow to the shape in the bitmap.
|
||||
|
||||
The shadow will be filled by `'2'`s.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/bitmap#shadow
|
||||
"""
|
||||
...
|
||||
def glow(self, mode=...): # -> Self:
|
||||
"""
|
||||
Add glow effect to the shape in the bitmap.
|
||||
|
||||
The glowing area is one pixel up, right, bottom and left to the original pixels (corners will not be filled in default mode 0 but will in mode 1), and will be filled by `'2'`s.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/bitmap#glow
|
||||
"""
|
||||
...
|
||||
def bytepad(self, bits=...): # -> Self:
|
||||
"""
|
||||
Pad each line (row) to multiple of 8 (or other numbers) bits/pixels, with `'0'`s.
|
||||
|
||||
Do this before using the bitmap for a glyph in a BDF font.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/bitmap#bytepad
|
||||
"""
|
||||
...
|
||||
def todata(
|
||||
self, datatype: int = ...
|
||||
) -> (
|
||||
Any
|
||||
): # -> LiteralString | Any | list[list[int]] | list[int] | list[str] | None:
|
||||
"""
|
||||
Get the bitmap's data in the specified type and format.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/bitmap#todata
|
||||
"""
|
||||
...
|
||||
def tobytes(self, mode=..., bytesdict=...): # -> bytes:
|
||||
"""
|
||||
Get the bitmap's data as `bytes` to be used with Pillow library's `Image.frombytes(mode, size, data)`.
|
||||
|
||||
https://font.tomchen.org/bdfparser_py/bitmap#tobytes
|
||||
"""
|
||||
...
|
||||
|
|
@ -1,19 +0,0 @@
|
|||
"""
|
||||
This type stub file was generated by pyright.
|
||||
"""
|
||||
|
||||
from .sense_hat import SenseHat
|
||||
from .stick import (
|
||||
ACTION_HELD,
|
||||
ACTION_PRESSED,
|
||||
ACTION_RELEASED,
|
||||
DIRECTION_DOWN,
|
||||
DIRECTION_LEFT,
|
||||
DIRECTION_MIDDLE,
|
||||
DIRECTION_RIGHT,
|
||||
DIRECTION_UP,
|
||||
InputEvent,
|
||||
SenseStick,
|
||||
)
|
||||
|
||||
__version__ = ...
|
||||
|
|
@ -1,222 +0,0 @@
|
|||
"""
|
||||
This type stub file was generated by pyright.
|
||||
"""
|
||||
|
||||
"""
|
||||
Python library for the TCS3472x and TCS340x Color Sensors
|
||||
Documentation (including datasheet): https://ams.com/tcs34725#tab/documents
|
||||
https://ams.com/tcs3400#tab/documents
|
||||
The sense hat for AstroPi on the ISS uses the TCS34725.
|
||||
The sense hat v2 uses the TCS3400 the successor of the TCS34725.
|
||||
The TCS34725 is not available any more. It was discontinued by ams in 2021.
|
||||
"""
|
||||
|
||||
class HardwareInterface:
|
||||
"""
|
||||
`HardwareInterface` is the abstract class that sits between the
|
||||
`ColourSensor` class (providing the TCS34725/TCS3400 sensor API)
|
||||
and the actual hardware. Using this intermediate layer of abstraction,
|
||||
a `ColourSensor` object interacts with the hardware without being
|
||||
aware of how this interaction is implemented.
|
||||
Different subclasses of the `HardwareInterface` class can provide
|
||||
access to the hardware through e.g. I2C, `libiio` and its system
|
||||
files or even a hardware emulator.
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def max_value(integration_cycles): # -> Literal[65535]:
|
||||
"""
|
||||
The maximum raw value for the RBGC channels depends on the number
|
||||
of integration cycles.
|
||||
"""
|
||||
...
|
||||
def get_enabled(self):
|
||||
"""
|
||||
Return True if the sensor is enabled and False otherwise
|
||||
"""
|
||||
...
|
||||
def set_enabled(self, status):
|
||||
"""
|
||||
Enable or disable the sensor, depending on the boolean `status` flag
|
||||
"""
|
||||
...
|
||||
def get_gain(self):
|
||||
"""
|
||||
Return the current value of the sensor gain.
|
||||
See GAIN_VALUES for the set of possible values.
|
||||
"""
|
||||
...
|
||||
def set_gain(self, gain):
|
||||
"""
|
||||
Set the value for the sensor `gain`.
|
||||
See GAIN_VALUES for the set of possible values.
|
||||
"""
|
||||
...
|
||||
def get_integration_cycles(self):
|
||||
"""
|
||||
Return the current number of integration_cycles (1-256).
|
||||
It takes `integration_cycles` * CLOCK_STEP to obtain a new
|
||||
sensor reading.
|
||||
"""
|
||||
...
|
||||
def set_integration_cycles(self, integration_cycles):
|
||||
"""
|
||||
Set the current number of integration_cycles (1-256).
|
||||
It takes `integration_cycles` * CLOCK_STEP to obtain a new
|
||||
sensor reading.
|
||||
"""
|
||||
...
|
||||
def get_raw(self):
|
||||
"""
|
||||
Return a tuple containing the raw values of the RGBC channels.
|
||||
The maximum for these raw values depends on the number of
|
||||
integration cycles and can be computed using `max_value`.
|
||||
"""
|
||||
...
|
||||
def get_red(self):
|
||||
"""
|
||||
Return the raw value of the R (red) channel.
|
||||
The maximum for this raw value depends on the number of
|
||||
integration cycles and can be computed using `max_value`.
|
||||
"""
|
||||
...
|
||||
def get_green(self):
|
||||
"""
|
||||
Return the raw value of the G (green) channel.
|
||||
The maximum for this raw value depends on the number of
|
||||
integration cycles and can be computed using `max_value`.
|
||||
"""
|
||||
...
|
||||
def get_blue(self):
|
||||
"""
|
||||
Return the raw value of the B (blue) channel.
|
||||
The maximum for this raw value depends on the number of
|
||||
integration cycles and can be computed using `max_value`.
|
||||
"""
|
||||
...
|
||||
def get_clear(self):
|
||||
"""
|
||||
Return the raw value of the C (clear light) channel.
|
||||
The maximum for this raw value depends on the number of
|
||||
integration cycles and can be computed using `max_value`.
|
||||
"""
|
||||
...
|
||||
|
||||
class I2C(HardwareInterface):
|
||||
"""
|
||||
An implementation of the `HardwareInterface` for the TCS34725/TCS3400
|
||||
sensor that uses I2C to control the sensor and retrieve measurements.
|
||||
Use the datasheets as a reference.
|
||||
"""
|
||||
|
||||
BUS = ...
|
||||
ENABLE = ...
|
||||
ATIME = ...
|
||||
CONTROL = ...
|
||||
ID = ...
|
||||
STATUS = ...
|
||||
CDATA = ...
|
||||
RDATA = ...
|
||||
GDATA = ...
|
||||
BDATA = ...
|
||||
OFF = ...
|
||||
PON = ...
|
||||
AEN = ...
|
||||
ON = ...
|
||||
AVALID = ...
|
||||
GAIN_REG_VALUES = ...
|
||||
ADDR = ...
|
||||
GAIN_VALUES = ...
|
||||
CLOCK_STEP = ...
|
||||
GAIN_TO_REG = ...
|
||||
REG_TO_GAIN = ...
|
||||
def __init__(self) -> None: ...
|
||||
@staticmethod
|
||||
def i2c_enabled(): # -> bool:
|
||||
"""Returns True if I2C is enabled or False otherwise."""
|
||||
...
|
||||
def get_enabled(self):
|
||||
"""
|
||||
Return True if the sensor is enabled and False otherwise
|
||||
"""
|
||||
...
|
||||
def set_enabled(self, status): # -> None:
|
||||
"""
|
||||
Enable or disable the sensor, depending on the boolean `status` flag
|
||||
"""
|
||||
...
|
||||
def get_gain(self): # -> Literal[1, 4, 16, 60, 64]:
|
||||
"""
|
||||
Return the current value of the sensor gain.
|
||||
See GAIN_VALUES for the set of possible values.
|
||||
"""
|
||||
...
|
||||
def set_gain(self, gain): # -> None:
|
||||
"""
|
||||
Set the value for the sensor `gain`.
|
||||
See GAIN_VALUES for the set of possible values.
|
||||
"""
|
||||
...
|
||||
def get_integration_cycles(self):
|
||||
"""
|
||||
Return the current number of integration_cycles (1-256).
|
||||
It takes `integration_cycles` * CLOCK_STEP to obtain a new
|
||||
sensor reading.
|
||||
"""
|
||||
...
|
||||
def set_integration_cycles(self, integration_cycles): # -> None:
|
||||
"""
|
||||
Set the current number of integration_cycles (1-256).
|
||||
It takes `integration_cycles` * CLOCK_STEP to obtain a new
|
||||
sensor reading.
|
||||
"""
|
||||
...
|
||||
def get_raw(self): # -> tuple[Any, Any, Any, Any]:
|
||||
"""
|
||||
Return a tuple containing the raw values of the RGBC channels.
|
||||
The maximum for these raw values depends on the number of
|
||||
integration cycles and can be computed using `max_value`.
|
||||
"""
|
||||
...
|
||||
get_red = ...
|
||||
get_green = ...
|
||||
get_blue = ...
|
||||
get_clear = ...
|
||||
|
||||
class ColourSensor:
|
||||
def __init__(self, gain=..., integration_cycles=..., interface=...) -> None: ...
|
||||
@property
|
||||
def enabled(self): ...
|
||||
@enabled.setter
|
||||
def enabled(self, status): ...
|
||||
@property
|
||||
def gain(self): ...
|
||||
@gain.setter
|
||||
def gain(self, gain): ...
|
||||
@property
|
||||
def integration_cycles(self): ...
|
||||
@integration_cycles.setter
|
||||
def integration_cycles(self, integration_cycles): ...
|
||||
@property
|
||||
def integration_time(self): ...
|
||||
@property
|
||||
def max_raw(self): ...
|
||||
@property
|
||||
def colour_raw(self): ...
|
||||
|
||||
color_raw = ...
|
||||
red_raw = ...
|
||||
green_raw = ...
|
||||
blue_raw = ...
|
||||
clear_raw = ...
|
||||
brightness = ...
|
||||
@property
|
||||
def colour(self): ...
|
||||
@property
|
||||
def rgb(self): ...
|
||||
|
||||
color = ...
|
||||
red = ...
|
||||
green = ...
|
||||
blue = ...
|
||||
clear = ...
|
||||
|
|
@ -1,20 +0,0 @@
|
|||
"""
|
||||
This type stub file was generated by pyright.
|
||||
"""
|
||||
|
||||
class SenseHatException(Exception):
|
||||
"""
|
||||
The base exception class for all SenseHat exceptions.
|
||||
"""
|
||||
|
||||
fmt = ...
|
||||
def __init__(self, **kwargs) -> None: ...
|
||||
|
||||
class ColourSensorInitialisationError(SenseHatException):
|
||||
fmt = ...
|
||||
|
||||
class InvalidGainError(SenseHatException):
|
||||
fmt = ...
|
||||
|
||||
class InvalidIntegrationCyclesError(SenseHatException):
|
||||
fmt = ...
|
||||
|
|
@ -1,226 +0,0 @@
|
|||
"""
|
||||
This type stub file was generated by pyright.
|
||||
"""
|
||||
from sense_hat.stick import SenseStick
|
||||
|
||||
class SenseHat:
|
||||
SENSE_HAT_FB_NAME = ...
|
||||
SENSE_HAT_FB_FBIOGET_GAMMA = ...
|
||||
SENSE_HAT_FB_FBIOSET_GAMMA = ...
|
||||
SENSE_HAT_FB_FBIORESET_GAMMA = ...
|
||||
SENSE_HAT_FB_GAMMA_DEFAULT = ...
|
||||
SENSE_HAT_FB_GAMMA_LOW = ...
|
||||
SENSE_HAT_FB_GAMMA_USER = ...
|
||||
SETTINGS_HOME_PATH = ...
|
||||
def __init__(self, imu_settings_file=..., text_assets=...) -> None: ...
|
||||
@property
|
||||
def stick(self) -> SenseStick: ...
|
||||
@property
|
||||
def colour(self): ...
|
||||
|
||||
color = ...
|
||||
def has_colour_sensor(self): ...
|
||||
@property
|
||||
def rotation(self): ...
|
||||
@rotation.setter
|
||||
def rotation(self, r): ...
|
||||
def set_rotation(self, r=..., redraw=...): # -> None:
|
||||
"""
|
||||
Sets the LED matrix rotation for viewing, adjust if the Pi is upside
|
||||
down or sideways. 0 is with the Pi HDMI port facing downwards
|
||||
"""
|
||||
...
|
||||
def flip_h(self, redraw=...): # -> list[Any]:
|
||||
"""
|
||||
Flip LED matrix horizontal
|
||||
"""
|
||||
...
|
||||
def flip_v(self, redraw=...): # -> list[Any]:
|
||||
"""
|
||||
Flip LED matrix vertical
|
||||
"""
|
||||
...
|
||||
def set_pixels(self, pixel_list: list[tuple[int, int, int]]) -> None: # -> None:
|
||||
"""
|
||||
Accepts a list containing 64 smaller lists of [R,G,B] pixels and
|
||||
updates the LED matrix. R,G,B elements must integers between 0
|
||||
and 255
|
||||
"""
|
||||
...
|
||||
def get_pixels(self): # -> list[Any]:
|
||||
"""
|
||||
Returns a list containing 64 smaller lists of [R,G,B] pixels
|
||||
representing what is currently displayed on the LED matrix
|
||||
"""
|
||||
...
|
||||
def set_pixel(self, x, y, *args): # -> None:
|
||||
"""
|
||||
Updates the single [R,G,B] pixel specified by x and y on the LED matrix
|
||||
Top left = 0,0 Bottom right = 7,7
|
||||
|
||||
e.g. ap.set_pixel(x, y, r, g, b)
|
||||
or
|
||||
pixel = (r, g, b)
|
||||
ap.set_pixel(x, y, pixel)
|
||||
"""
|
||||
...
|
||||
def get_pixel(self, x, y): # -> list[int]:
|
||||
"""
|
||||
Returns a list of [R,G,B] representing the pixel specified by x and y
|
||||
on the LED matrix. Top left = 0,0 Bottom right = 7,7
|
||||
"""
|
||||
...
|
||||
def load_image(self, file_path, redraw=...): # -> list[list[Any]]:
|
||||
"""
|
||||
Accepts a path to an 8 x 8 image file and updates the LED matrix with
|
||||
the image
|
||||
"""
|
||||
...
|
||||
def clear(self, *args): # -> None:
|
||||
"""
|
||||
Clears the LED matrix with a single colour, default is black / off
|
||||
|
||||
e.g. ap.clear()
|
||||
or
|
||||
ap.clear(r, g, b)
|
||||
or
|
||||
colour = (r, g, b)
|
||||
ap.clear(colour)
|
||||
"""
|
||||
...
|
||||
def show_message(
|
||||
self, text_string, scroll_speed=..., text_colour=..., back_colour=...
|
||||
): # -> None:
|
||||
"""
|
||||
Scrolls a string of text across the LED matrix using the specified
|
||||
speed and colours
|
||||
"""
|
||||
...
|
||||
def show_letter(self, s, text_colour=..., back_colour=...): # -> None:
|
||||
"""
|
||||
Displays a single text character on the LED matrix using the specified
|
||||
colours
|
||||
"""
|
||||
...
|
||||
@property
|
||||
def gamma(self): ...
|
||||
@gamma.setter
|
||||
def gamma(self, buffer): ...
|
||||
def gamma_reset(self) -> None: # -> None:
|
||||
"""
|
||||
Resets the LED matrix gamma correction to default
|
||||
"""
|
||||
...
|
||||
@property
|
||||
def low_light(self): ...
|
||||
@low_light.setter
|
||||
def low_light(self, value): ...
|
||||
def get_humidity(self): # -> Literal[0]:
|
||||
"""
|
||||
Returns the percentage of relative humidity
|
||||
"""
|
||||
...
|
||||
@property
|
||||
def humidity(self): ...
|
||||
def get_temperature_from_humidity(self): # -> Literal[0]:
|
||||
"""
|
||||
Returns the temperature in Celsius from the humidity sensor
|
||||
"""
|
||||
...
|
||||
def get_temperature_from_pressure(self): # -> Literal[0]:
|
||||
"""
|
||||
Returns the temperature in Celsius from the pressure sensor
|
||||
"""
|
||||
...
|
||||
def get_temperature(self): # -> Literal[0]:
|
||||
"""
|
||||
Returns the temperature in Celsius
|
||||
"""
|
||||
...
|
||||
@property
|
||||
def temp(self): ...
|
||||
@property
|
||||
def temperature(self): ...
|
||||
def get_pressure(self): # -> Literal[0]:
|
||||
"""
|
||||
Returns the pressure in Millibars
|
||||
"""
|
||||
...
|
||||
@property
|
||||
def pressure(self): ...
|
||||
def set_imu_config(self, compass_enabled, gyro_enabled, accel_enabled): # -> None:
|
||||
"""
|
||||
Enables and disables the gyroscope, accelerometer and/or magnetometer
|
||||
input to the orientation functions
|
||||
"""
|
||||
...
|
||||
def get_orientation_radians(self): # -> dict[str, Any] | dict[str, int]:
|
||||
"""
|
||||
Returns a dictionary object to represent the current orientation in
|
||||
radians using the aircraft principal axes of pitch, roll and yaw
|
||||
"""
|
||||
...
|
||||
@property
|
||||
def orientation_radians(self): ...
|
||||
def get_orientation_degrees(self): # -> dict[str, Any] | dict[str, int]:
|
||||
"""
|
||||
Returns a dictionary object to represent the current orientation
|
||||
in degrees, 0 to 360, using the aircraft principal axes of
|
||||
pitch, roll and yaw
|
||||
"""
|
||||
...
|
||||
def get_orientation(self): ...
|
||||
@property
|
||||
def orientation(self): ...
|
||||
def get_compass(self): # -> int | None:
|
||||
"""
|
||||
Gets the direction of North from the magnetometer in degrees
|
||||
"""
|
||||
...
|
||||
@property
|
||||
def compass(self): ...
|
||||
def get_compass_raw(self): # -> dict[str, Any] | dict[str, int]:
|
||||
"""
|
||||
Magnetometer x y z raw data in uT (micro teslas)
|
||||
"""
|
||||
...
|
||||
@property
|
||||
def compass_raw(self): ...
|
||||
def get_gyroscope(self): # -> dict[str, Any] | dict[str, int]:
|
||||
"""
|
||||
Gets the orientation in degrees from the gyroscope only
|
||||
"""
|
||||
...
|
||||
@property
|
||||
def gyro(self): ...
|
||||
@property
|
||||
def gyroscope(self): ...
|
||||
def get_gyroscope_raw(self): # -> dict[str, Any] | dict[str, int]:
|
||||
"""
|
||||
Gyroscope x y z raw data in radians per second
|
||||
"""
|
||||
...
|
||||
@property
|
||||
def gyro_raw(self): ...
|
||||
@property
|
||||
def gyroscope_raw(self): ...
|
||||
def get_accelerometer(self): # -> dict[str, Any] | dict[str, int]:
|
||||
"""
|
||||
Gets the orientation in degrees from the accelerometer only
|
||||
"""
|
||||
...
|
||||
@property
|
||||
def accel(self): ...
|
||||
@property
|
||||
def accelerometer(self): ...
|
||||
def get_accelerometer_raw(
|
||||
self,
|
||||
) -> dict[str, int]: # -> dict[str, Any] | dict[str, int]:
|
||||
"""
|
||||
Accelerometer x y z raw data in Gs
|
||||
"""
|
||||
...
|
||||
@property
|
||||
def accel_raw(self): ...
|
||||
@property
|
||||
def accelerometer_raw(self): ...
|
||||
|
|
@ -1,144 +0,0 @@
|
|||
"""
|
||||
This type stub file was generated by pyright.
|
||||
"""
|
||||
from collections.abc import Callable
|
||||
from typing import Any, NamedTuple
|
||||
|
||||
# native_str = str
|
||||
# str = ...
|
||||
DIRECTION_UP = ...
|
||||
DIRECTION_DOWN = ...
|
||||
DIRECTION_LEFT = ...
|
||||
DIRECTION_RIGHT = ...
|
||||
DIRECTION_MIDDLE = ...
|
||||
ACTION_PRESSED = ...
|
||||
ACTION_RELEASED = ...
|
||||
ACTION_HELD = ...
|
||||
|
||||
class InputEvent(NamedTuple):
|
||||
timestamp: int
|
||||
direction: str
|
||||
action: str
|
||||
|
||||
class SenseStick:
|
||||
"""
|
||||
Represents the joystick on the Sense HAT.
|
||||
"""
|
||||
|
||||
SENSE_HAT_EVDEV_NAME = ...
|
||||
EVENT_FORMAT = ...
|
||||
EVENT_SIZE = ...
|
||||
EV_KEY = ...
|
||||
STATE_RELEASE = ...
|
||||
STATE_PRESS = ...
|
||||
STATE_HOLD = ...
|
||||
KEY_UP = ...
|
||||
KEY_LEFT = ...
|
||||
KEY_RIGHT = ...
|
||||
KEY_DOWN = ...
|
||||
KEY_ENTER = ...
|
||||
def __init__(self) -> None: ...
|
||||
def close(self): ...
|
||||
def __enter__(self): ...
|
||||
def __exit__(self, exc_type, exc_value, exc_tb): ...
|
||||
def wait_for_event(self, emptybuffer=...): # -> InputEvent | None:
|
||||
"""
|
||||
Waits until a joystick event becomes available. Returns the event, as
|
||||
an `InputEvent` tuple.
|
||||
|
||||
If *emptybuffer* is `True` (it defaults to `False`), any pending
|
||||
events will be thrown away first. This is most useful if you are only
|
||||
interested in "pressed" events.
|
||||
"""
|
||||
...
|
||||
def get_events(self): # -> list[Any]:
|
||||
"""
|
||||
Returns a list of all joystick events that have occurred since the last
|
||||
call to `get_events`. The list contains events in the order that they
|
||||
occurred. If no events have occurred in the intervening time, the
|
||||
result is an empty list.
|
||||
"""
|
||||
...
|
||||
@property
|
||||
def direction_up(self): # -> None:
|
||||
"""
|
||||
The function to be called when the joystick is pushed up. The function
|
||||
can either take a parameter which will be the `InputEvent` tuple that
|
||||
has occurred, or the function can take no parameters at all.
|
||||
"""
|
||||
...
|
||||
@direction_up.setter
|
||||
def direction_up(
|
||||
self, value: Callable[[InputEvent], Any] | Callable[[], Any]
|
||||
) -> None: ...
|
||||
@property
|
||||
def direction_down(self): # -> None:
|
||||
"""
|
||||
The function to be called when the joystick is pushed down. The
|
||||
function can either take a parameter which will be the `InputEvent`
|
||||
tuple that has occurred, or the function can take no parameters at all.
|
||||
|
||||
Assign `None` to prevent this event from being fired.
|
||||
"""
|
||||
...
|
||||
@direction_down.setter
|
||||
def direction_down(
|
||||
self, value: Callable[[InputEvent], Any] | Callable[[], Any]
|
||||
) -> None: ...
|
||||
@property
|
||||
def direction_left(self): # -> None:
|
||||
"""
|
||||
The function to be called when the joystick is pushed left. The
|
||||
function can either take a parameter which will be the `InputEvent`
|
||||
tuple that has occurred, or the function can take no parameters at all.
|
||||
|
||||
Assign `None` to prevent this event from being fired.
|
||||
"""
|
||||
...
|
||||
@direction_left.setter
|
||||
def direction_left(
|
||||
self, value: Callable[[InputEvent], Any] | Callable[[], Any]
|
||||
) -> None: ...
|
||||
@property
|
||||
def direction_right(self): # -> None:
|
||||
"""
|
||||
The function to be called when the joystick is pushed right. The
|
||||
function can either take a parameter which will be the `InputEvent`
|
||||
tuple that has occurred, or the function can take no parameters at all.
|
||||
|
||||
Assign `None` to prevent this event from being fired.
|
||||
"""
|
||||
...
|
||||
@direction_right.setter
|
||||
def direction_right(
|
||||
self, value: Callable[[InputEvent], Any] | Callable[[], Any]
|
||||
) -> None: ...
|
||||
@property
|
||||
def direction_middle(self): # -> None:
|
||||
"""
|
||||
The function to be called when the joystick middle click is pressed. The
|
||||
function can either take a parameter which will be the `InputEvent` tuple
|
||||
that has occurred, or the function can take no parameters at all.
|
||||
|
||||
Assign `None` to prevent this event from being fired.
|
||||
"""
|
||||
...
|
||||
@direction_middle.setter
|
||||
def direction_middle(
|
||||
self, value: Callable[[InputEvent], Any] | Callable[[], Any]
|
||||
) -> None: ...
|
||||
@property
|
||||
def direction_any(self): # -> None:
|
||||
"""
|
||||
The function to be called when the joystick is used. The function
|
||||
can either take a parameter which will be the `InputEvent` tuple that
|
||||
has occurred, or the function can take no parameters at all.
|
||||
|
||||
This event will always be called *after* events associated with a
|
||||
specific action. Assign `None` to prevent this event from being fired.
|
||||
"""
|
||||
...
|
||||
@direction_any.setter
|
||||
def direction_any(
|
||||
self, value: Callable[[InputEvent], Any] | Callable[[], Any]
|
||||
) -> None: ...
|
||||
Reference in a new issue