Initial commit
This commit is contained in:
@@ -0,0 +1,186 @@
|
||||
# tools/geometry.py
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
@dataclass
|
||||
class Point:
|
||||
"""Represents a 2D point in a Cartesian coordinate system."""
|
||||
|
||||
x: float
|
||||
y: float
|
||||
|
||||
|
||||
class Line:
|
||||
"""Represents a finite line segment between two Points."""
|
||||
|
||||
def __init__(self, p1: Point, p2: Point) -> None:
|
||||
self.p1 = p1
|
||||
self.p2 = p2
|
||||
|
||||
def distance_to_point(self, point: Point) -> float:
|
||||
"""
|
||||
Calculates the shortest distance from this line segment to a given point.
|
||||
Uses vector projection to find the closest point on the segment.
|
||||
"""
|
||||
x1, y1 = self.p1.x, self.p1.y
|
||||
x2, y2 = self.p2.x, self.p2.y
|
||||
x3, y3 = point.x, point.y
|
||||
|
||||
# If the line is actually just a single point
|
||||
if x1 == x2 and y1 == y2:
|
||||
return ((x1 - x3) ** 2 + (y1 - y3) ** 2) ** 0.5
|
||||
|
||||
px, py = x2 - x1, y2 - y1
|
||||
norm = px * px + py * py
|
||||
|
||||
# Calculate the projection scalar (u) of the point onto the line
|
||||
u = ((x3 - x1) * px + (y3 - y1) * py) / float(norm)
|
||||
|
||||
# Clamp u to the [0, 1] range to ensure we stay on the line segment
|
||||
u = max(0.0, min(1.0, u))
|
||||
|
||||
# Find the exact coordinates of the closest point on the segment
|
||||
closest_x = x1 + u * px
|
||||
closest_y = y1 + u * py
|
||||
|
||||
# Return distance from the target point to the closest point
|
||||
dx, dy = closest_x - x3, closest_y - y3
|
||||
return (dx * dx + dy * dy) ** 0.5
|
||||
|
||||
|
||||
class LineString:
|
||||
"""Represents a path formed by a sequence of connected line segments."""
|
||||
|
||||
def __init__(self, lines: list[list[float]]) -> None:
|
||||
if len(lines) < 2:
|
||||
raise ValueError("A LineString requires at least 2 coordinate pairs.")
|
||||
|
||||
self.lineList: list[Line] = []
|
||||
for i in range(len(lines) - 1):
|
||||
p1 = Point(lines[i][0], lines[i][1])
|
||||
p2 = Point(lines[i + 1][0], lines[i + 1][1])
|
||||
self.lineList.append(Line(p1, p2))
|
||||
|
||||
def distance_to(self, point: Point) -> float:
|
||||
"""Calculates the minimum distance from the given point to the LineString."""
|
||||
# Cleanly check the distance to all segments and return the smallest one
|
||||
return min(line.distance_to_point(point) for line in self.lineList)
|
||||
|
||||
def is_intersecting(self, point: Point) -> bool:
|
||||
"""Checks if a given point lies exactly on the LineString."""
|
||||
return self.distance_to(point) == 0.0
|
||||
|
||||
|
||||
class Polygon:
|
||||
"""Represents a 2D shape enclosed by a series of connected points."""
|
||||
|
||||
def __init__(self, points: list[list[float]]) -> None:
|
||||
self.listPoints: list[Point] = [Point(p[0], p[1]) for p in points]
|
||||
|
||||
def on_line(self, l1: Line, p: Point) -> bool:
|
||||
"""Checks if collinear point 'p' lies strictly on the line segment 'l1'."""
|
||||
return min(l1.p1.x, l1.p2.x) <= p.x <= max(l1.p1.x, l1.p2.x) and min(
|
||||
l1.p1.y, l1.p2.y
|
||||
) <= p.y <= max(l1.p1.y, l1.p2.y)
|
||||
|
||||
def direction(self, a: Point, b: Point, c: Point) -> int:
|
||||
"""
|
||||
Finds the orientation of an ordered triplet (a, b, c).
|
||||
Returns:
|
||||
0 : Collinear
|
||||
1 : Clockwise
|
||||
2 : Counterclockwise
|
||||
"""
|
||||
val = (b.y - a.y) * (c.x - b.x) - (b.x - a.x) * (c.y - b.y)
|
||||
if val == 0:
|
||||
return 0
|
||||
return 2 if val < 0 else 1
|
||||
|
||||
def is_intersect(self, l1: Line, l2: Line) -> bool:
|
||||
"""Checks if line segment l1 intersects with line segment l2."""
|
||||
dir1 = self.direction(l1.p1, l1.p2, l2.p1)
|
||||
dir2 = self.direction(l1.p1, l1.p2, l2.p2)
|
||||
dir3 = self.direction(l2.p1, l2.p2, l1.p1)
|
||||
dir4 = self.direction(l2.p1, l2.p2, l1.p2)
|
||||
|
||||
# General case intersection
|
||||
if dir1 != dir2 and dir3 != dir4:
|
||||
return True
|
||||
|
||||
# Special collinear cases
|
||||
if dir1 == 0 and self.on_line(l1, l2.p1):
|
||||
return True
|
||||
if dir2 == 0 and self.on_line(l1, l2.p2):
|
||||
return True
|
||||
if dir3 == 0 and self.on_line(l2, l1.p1):
|
||||
return True
|
||||
if dir4 == 0 and self.on_line(l2, l1.p2):
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def contains(self, p: Point) -> bool:
|
||||
"""
|
||||
Determines if a point is strictly inside the Polygon using the Ray-Casting algorithm.
|
||||
Draws a horizontal line to the right of the point and counts edge intersections.
|
||||
"""
|
||||
n = len(self.listPoints)
|
||||
if n < 3:
|
||||
return False
|
||||
|
||||
# Create a horizontal ray starting from the point and going infinitely right
|
||||
exline = Line(p, Point(99999.0, p.y))
|
||||
count = 0
|
||||
|
||||
for i in range(n):
|
||||
side = Line(self.listPoints[i], self.listPoints[(i + 1) % n])
|
||||
|
||||
if self.is_intersect(side, exline):
|
||||
# If the point is collinear with the side, check if it's strictly on the side
|
||||
if self.direction(side.p1, p, side.p2) == 0:
|
||||
return self.on_line(side, p)
|
||||
count += 1
|
||||
|
||||
# If the number of intersections is odd, the point is inside the polygon
|
||||
return bool(count & 1)
|
||||
|
||||
|
||||
class MultiPolygon:
|
||||
"""Represents a collection of multiple separate Polygons."""
|
||||
|
||||
def __init__(self, polygons: list) -> None:
|
||||
self.listPolygons: list[Polygon] = [Polygon(polygon[0]) for polygon in polygons]
|
||||
|
||||
def contains(self, p: Point) -> bool:
|
||||
"""Returns True if the point is inside ANY of the contained Polygons."""
|
||||
for polygon in self.listPolygons:
|
||||
if polygon.contains(p):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
@dataclass
|
||||
class Coordinate:
|
||||
"""
|
||||
Custom type for the management of geographical coordinates.
|
||||
Provides utility to switch between lat/long and x/y point spaces.
|
||||
"""
|
||||
|
||||
lat: float
|
||||
long: float
|
||||
|
||||
@property
|
||||
def tuple(self) -> tuple[float, float]:
|
||||
"""Returns the coordinates as a (Latitude, Longitude) tuple."""
|
||||
return (self.lat, self.long)
|
||||
|
||||
def to_Point(self, reversed: bool = False) -> Point:
|
||||
"""
|
||||
Returns Coordinates mapped to a Cartesian Point.
|
||||
|
||||
Args:
|
||||
reversed (bool): If True, switches orientation so x=long, y=lat.
|
||||
Defaults to False (x=lat, y=long).
|
||||
"""
|
||||
return Point(self.long, self.lat) if reversed else Point(self.lat, self.long)
|
||||
@@ -0,0 +1,58 @@
|
||||
# tools/julian.py
|
||||
|
||||
from datetime import datetime, timedelta
|
||||
import math
|
||||
|
||||
|
||||
def julian_date(date: datetime | None = None):
|
||||
"""Given any date in the future or past, return julian date"""
|
||||
if date is None:
|
||||
date = datetime.now()
|
||||
|
||||
time = date.timestamp() * 1000
|
||||
|
||||
offset = date.utcoffset()
|
||||
tzoffset = offset.total_seconds() // 60 if offset is not None else 0
|
||||
|
||||
return (time / 86400000) - (tzoffset / 1440) + 2440587.5
|
||||
|
||||
|
||||
def CJDN(date: datetime | None = None):
|
||||
return round(julian_date(date))
|
||||
|
||||
|
||||
def current_julian_date():
|
||||
"""Returns current julian date"""
|
||||
return julian_date()
|
||||
|
||||
|
||||
def tomorrow_julian_date():
|
||||
"""returns tomorrow's julian date"""
|
||||
date = datetime.now() + timedelta(days=1)
|
||||
return julian_date(date)
|
||||
|
||||
|
||||
def future_julian_date(hoursadd=0, minutesadd=0, secondsadd=0):
|
||||
"""defaults to current if no specifications made. HOURS, MINUTES, AND SECONDS CANNOT BE NEGATIVE"""
|
||||
date = datetime.now() + timedelta(
|
||||
hours=hoursadd, minutes=minutesadd, seconds=secondsadd
|
||||
)
|
||||
return julian_date(date)
|
||||
|
||||
|
||||
def epoch_days(date: datetime | None = None):
|
||||
"""returns days since Jan 1st, 2000. Negative if before this date"""
|
||||
return julian_date(date) - 2451545
|
||||
|
||||
|
||||
def day_percent(date: datetime | None = None):
|
||||
"""Returns decimal portion of Julian Date"""
|
||||
whole = julian_date(date)
|
||||
return whole - math.floor(whole)
|
||||
|
||||
|
||||
def from_julian(j):
|
||||
"""Returns datetime.datetime object given julian date J"""
|
||||
J1970 = 2440588
|
||||
dayMs = 24 * 60 * 60 * 1000
|
||||
return datetime.fromtimestamp((j + 0.5 - J1970) * dayMs / 1000.0)
|
||||
@@ -0,0 +1,47 @@
|
||||
# tools/mqtt.py
|
||||
|
||||
import json
|
||||
import socket
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from paho.mqtt.publish import single as mqtt_publish
|
||||
|
||||
from constants import MQTT_USERNAME, MQTT_PASSWORD, MQTT_FALLBACK_IP
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class MQTT:
|
||||
def __init__(self):
|
||||
self._username = MQTT_USERNAME
|
||||
self._password = MQTT_PASSWORD
|
||||
self.IP = self._resolve_ip()
|
||||
|
||||
def _resolve_ip(self) -> str:
|
||||
"""Attempts to resolve the Home Assistant IP, falls back to static on failure/timeout."""
|
||||
socket.setdefaulttimeout(2.0)
|
||||
try:
|
||||
return socket.gethostbyname("homeassistant.local")
|
||||
except (socket.gaierror, socket.timeout, OSError) as e:
|
||||
logger.warning(
|
||||
f"Could not resolve homeassistant.local ({e}). Falling back to static IP."
|
||||
)
|
||||
return MQTT_FALLBACK_IP
|
||||
finally:
|
||||
socket.setdefaulttimeout(None)
|
||||
|
||||
def send(self, data: dict[str, Any], topic: str = "weather_station") -> None:
|
||||
logger.debug("Sending MQTT payload to server...")
|
||||
try:
|
||||
mqtt_publish(
|
||||
topic,
|
||||
json.dumps(data),
|
||||
retain=True,
|
||||
hostname=self.IP,
|
||||
keepalive=120,
|
||||
client_id="weather-station",
|
||||
auth={"password": self._password, "username": self._username},
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"MQTT data couldn't be sent: {e}")
|
||||
@@ -0,0 +1,44 @@
|
||||
# tools/weather_logger.py
|
||||
|
||||
import logging
|
||||
import logging.handlers
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def set_up_logger():
|
||||
logger = logging.getLogger()
|
||||
|
||||
if logger.hasHandlers():
|
||||
return
|
||||
|
||||
logger.setLevel(logging.DEBUG)
|
||||
logging.getLogger("PIL").setLevel(logging.WARNING)
|
||||
formatter = logging.Formatter(
|
||||
"{asctime} - {levelname:>7}: {message}", "%y-%m-%d %H:%M:%S", style="{"
|
||||
)
|
||||
log_dir = Path(__file__).parent.parent.joinpath("logs")
|
||||
log_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
debugFileHandler = logging.handlers.RotatingFileHandler(
|
||||
filename=log_dir.joinpath("debug.log"),
|
||||
maxBytes=1_048_576,
|
||||
backupCount=5,
|
||||
)
|
||||
debugFileHandler.setFormatter(formatter)
|
||||
debugFileHandler.setLevel(logging.DEBUG)
|
||||
|
||||
infoFileHandler = logging.handlers.RotatingFileHandler(
|
||||
filename=log_dir.joinpath("info.log"),
|
||||
maxBytes=1_048_576,
|
||||
backupCount=3,
|
||||
)
|
||||
infoFileHandler.setFormatter(formatter)
|
||||
infoFileHandler.setLevel(logging.INFO)
|
||||
|
||||
console = logging.StreamHandler()
|
||||
console.setFormatter(formatter)
|
||||
console.setLevel(logging.INFO)
|
||||
|
||||
logger.addHandler(debugFileHandler)
|
||||
logger.addHandler(infoFileHandler)
|
||||
logger.addHandler(console)
|
||||
Reference in New Issue
Block a user