Added test and small date fix.
This commit is contained in:
+3
-3
@@ -1,9 +1,9 @@
|
||||
# Network & Sensor
|
||||
SENSOR_IP=192.168.0.188
|
||||
SENSOR_IP=10.11.4.78
|
||||
MQTT_USERNAME=weather_station
|
||||
MQTT_PASSWORD=SensorData12345!
|
||||
MQTT_FALLBACK_IP=192.168.0.168
|
||||
MQTT_FALLBACK_IP=10.11.0.33
|
||||
|
||||
# Location (Defaulting to your Skåne coordinates)
|
||||
# Location
|
||||
LATITUDE=56.2006
|
||||
LONGITUDE=12.5553
|
||||
Vendored
+7
@@ -0,0 +1,7 @@
|
||||
{
|
||||
"python.testing.pytestArgs": [
|
||||
"."
|
||||
],
|
||||
"python.testing.unittestEnabled": false,
|
||||
"python.testing.pytestEnabled": true
|
||||
}
|
||||
+4
-3
@@ -7,7 +7,7 @@ from PIL import Image
|
||||
from .epd4in2 import EPD, epdconfig
|
||||
from .image_maker import eInkImage
|
||||
|
||||
logging.getLogger(__name__)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Display:
|
||||
@@ -23,7 +23,8 @@ class Display:
|
||||
epd.Clear()
|
||||
return epd
|
||||
except Exception as e:
|
||||
logging.error(f"eInk Display not found! | {e}")
|
||||
logger.error(f"eInk Display not found! | {e}")
|
||||
raise e
|
||||
|
||||
@staticmethod
|
||||
def _wait_until(timestamp: str, absolute_date: datetime) -> None:
|
||||
@@ -46,4 +47,4 @@ class Display:
|
||||
self.e_Paper.display_4Gray(self.e_Paper.getbuffer_4Gray(image))
|
||||
self.e_Paper.sleep()
|
||||
except Exception as e:
|
||||
logging.error(f"Failed displaying image - ERROR: {e}")
|
||||
logger.error(f"Failed displaying image - ERROR: {e}")
|
||||
|
||||
@@ -27,7 +27,7 @@ class DummyGPIO:
|
||||
|
||||
@staticmethod
|
||||
def input(*args):
|
||||
return 0
|
||||
return 1
|
||||
|
||||
@staticmethod
|
||||
def cleanup(*args):
|
||||
|
||||
@@ -0,0 +1,40 @@
|
||||
# tests/test_geometry.py
|
||||
|
||||
import pytest
|
||||
from tools.geometry import Point, Line, Polygon, Coordinate
|
||||
|
||||
|
||||
def test_coordinate_to_point():
|
||||
coord = Coordinate(lat=56.2006, long=12.5553)
|
||||
p1 = coord.to_Point()
|
||||
assert p1.x == 56.2006
|
||||
assert p1.y == 12.5553
|
||||
|
||||
p2 = coord.to_Point(reversed=True)
|
||||
assert p2.x == 12.5553
|
||||
assert p2.y == 56.2006
|
||||
|
||||
|
||||
def test_line_distance_to_point():
|
||||
# A straight line from (0,0) to (10,0)
|
||||
line = Line(Point(0, 0), Point(10, 0))
|
||||
|
||||
# Point exactly in the middle, 5 units above
|
||||
dist1 = line.distance_to_point(Point(5, 5))
|
||||
assert dist1 == 5.0
|
||||
|
||||
# Point way past the end of the line (closest point is the end (10,0))
|
||||
dist2 = line.distance_to_point(Point(13, 4))
|
||||
assert dist2 == 5.0 # 3-4-5 right triangle from (10,0) to (13,4)
|
||||
|
||||
|
||||
def test_polygon_contains():
|
||||
# A simple 10x10 square
|
||||
square = Polygon([[0, 0], [10, 0], [10, 10], [0, 10]])
|
||||
|
||||
# Inside
|
||||
assert square.contains(Point(5, 5)) is True
|
||||
# Outside
|
||||
assert square.contains(Point(15, 5)) is False
|
||||
# On the line (Edge case)
|
||||
assert square.contains(Point(10, 5)) is True
|
||||
@@ -0,0 +1,83 @@
|
||||
# tests/test_image_maker.py
|
||||
|
||||
import pytest
|
||||
from PIL import Image
|
||||
from display.image_maker import closest, eInkImage
|
||||
|
||||
|
||||
def test_closest_math_logic():
|
||||
# Test that your wind direction logic snaps to the correct cardinal degree
|
||||
angles = [0, 45, 90, 135, 180, 225, 270, 315]
|
||||
|
||||
assert closest(angles, 10) == 0 # 10 is closest to 0
|
||||
assert closest(angles, 85) == 90 # 85 is closest to 90
|
||||
assert closest(angles, 190) == 180 # 190 is closest to 180
|
||||
assert closest(angles, 100) == 90
|
||||
|
||||
|
||||
def test_make_image_compilation():
|
||||
"""
|
||||
Tests the entire Pillow drawing loop.
|
||||
If you miss a .get() check or use an invalid font size, this test will catch it.
|
||||
"""
|
||||
image_maker = eInkImage()
|
||||
|
||||
# A perfectly standard mock payload simulating SMHI and sensors
|
||||
dummy_weather_data = {
|
||||
"indoor_temp": 22.5,
|
||||
"indoor_humidity": 45,
|
||||
"pressure": 1013,
|
||||
"sunrise": "06:30",
|
||||
"sunset": "18:45",
|
||||
"moon_illumination": 0.8,
|
||||
"moon_phase": "Fullmåne",
|
||||
"moon_to_full": False,
|
||||
"temperature": 15.2,
|
||||
"humidity": 60,
|
||||
"dew": 5.0,
|
||||
"uv_index": 3,
|
||||
"fire_index": 1,
|
||||
"weather_alerts": [],
|
||||
"forecast": {
|
||||
"wind": {"deg": 180, "speed": 5.0, "gust": 7.5},
|
||||
"weather_symbol": 1,
|
||||
},
|
||||
}
|
||||
|
||||
try:
|
||||
# Generate the image
|
||||
img = image_maker.make_image(
|
||||
date=("Torsdag, 19 mars 2026", "22:10"),
|
||||
weather_data=dummy_weather_data,
|
||||
network_status=True,
|
||||
)
|
||||
|
||||
# Verify it successfully created an image object
|
||||
assert isinstance(img, Image.Image)
|
||||
|
||||
# Verify the dimensions match your Waveshare 4.2inch E-ink display
|
||||
assert img.size == (400, 300)
|
||||
|
||||
# Verify it converted it to a 1-bit color palette (pure black and white) for the e-ink screen
|
||||
assert img.mode == "L"
|
||||
|
||||
except Exception as e:
|
||||
pytest.fail(f"Image generation crashed with valid data: {e}")
|
||||
|
||||
|
||||
def test_make_image_with_missing_data():
|
||||
"""
|
||||
Tests that the image maker doesn't crash if the internet is down
|
||||
and the weather data is mostly empty.
|
||||
"""
|
||||
image_maker = eInkImage()
|
||||
|
||||
try:
|
||||
img = image_maker.make_image(
|
||||
date=("Torsdag, 19 mars 2026", "22:10"),
|
||||
weather_data={},
|
||||
network_status=False,
|
||||
)
|
||||
assert isinstance(img, Image.Image)
|
||||
except Exception as e:
|
||||
pytest.fail(f"Image generation crashed when weather data was missing: {e}")
|
||||
@@ -0,0 +1,30 @@
|
||||
# tests/test_julian.py
|
||||
|
||||
from datetime import datetime, timezone, timedelta
|
||||
from freezegun import freeze_time
|
||||
from tools.julian import julian_date, epoch_days
|
||||
|
||||
|
||||
# Freeze time to exactly Jan 1, 2000, 12:00:00 UTC (The J2000 epoch)
|
||||
@freeze_time("2000-01-01 12:00:00", tz_offset=0)
|
||||
def test_julian_date_j2000():
|
||||
# J2000 should be exactly Julian Date 2451545.0
|
||||
jd = julian_date()
|
||||
assert jd == 2451545.0
|
||||
|
||||
|
||||
@freeze_time("2000-01-01 12:00:00", tz_offset=0)
|
||||
def test_epoch_days():
|
||||
# Since we are exactly on the epoch, days should be 0
|
||||
days = epoch_days()
|
||||
assert days == 0.0
|
||||
|
||||
|
||||
def test_julian_date_with_timezone():
|
||||
# Create a timezone-aware datetime
|
||||
tz = timezone(timedelta(hours=2))
|
||||
dt = datetime(2023, 10, 15, 12, 0, 0, tzinfo=tz)
|
||||
|
||||
jd = julian_date(dt)
|
||||
# 2023-10-15 12:00 GMT+2 is exactly 2460232.91666...
|
||||
assert round(jd, 4) == 2460232.9167
|
||||
@@ -0,0 +1,53 @@
|
||||
# tests/test_main.py
|
||||
|
||||
import pytest
|
||||
from datetime import datetime
|
||||
import main
|
||||
|
||||
|
||||
def test_date_string_formatting():
|
||||
# March 19, 2026 is a Thursday (weekday index 3)
|
||||
dt = datetime(2026, 3, 19, 14, 5)
|
||||
date_line, time_line = main.date_string(dt)
|
||||
|
||||
# Verify the translation dictionaries work properly
|
||||
assert date_line == "Torsdag, 19 mars 2026"
|
||||
assert time_line == "14:05" # Ensure the leading zero logic works
|
||||
|
||||
|
||||
def test_is_network_up_success(mocker):
|
||||
# Create a fake ping object that says the internet is alive
|
||||
mock_host = mocker.MagicMock()
|
||||
mock_host.is_alive = True
|
||||
|
||||
# Intercept the `_ping` function in main.py
|
||||
mocker.patch("main._ping", return_value=mock_host)
|
||||
|
||||
assert main.is_network_up() is True
|
||||
|
||||
|
||||
def test_is_network_up_failure(mocker):
|
||||
# Simulate the ping throwing an exception (e.g., DNS failure or cable unplugged)
|
||||
mocker.patch("main._ping", side_effect=Exception("Network down"))
|
||||
|
||||
# Assert our function catches it and safely returns False
|
||||
assert main.is_network_up() is False
|
||||
|
||||
|
||||
def test_update_orchestration(mocker):
|
||||
# We want to test the update() function without making real web requests or drawing to the screen.
|
||||
# So, we "mock" (fake) the inner components.
|
||||
mocker.patch("main.weather.get", return_value={"temperature": 20.0})
|
||||
mock_mqtt = mocker.patch("main.mqtt.send")
|
||||
mock_make_image = mocker.patch(
|
||||
"main.display.image.make_image", return_value="dummy_image_object"
|
||||
)
|
||||
mock_refresh = mocker.patch("main.display.refresh")
|
||||
|
||||
# Run the function
|
||||
main.update(first=True)
|
||||
|
||||
# Verify that the orchestrator passed the data perfectly down the chain!
|
||||
mock_mqtt.assert_called_once_with({"temperature": 20.0})
|
||||
mock_make_image.assert_called_once()
|
||||
mock_refresh.assert_called_once()
|
||||
@@ -0,0 +1,37 @@
|
||||
# tests/test_mqtt.py
|
||||
|
||||
import pytest
|
||||
import socket
|
||||
from tools.mqtt import MQTT
|
||||
|
||||
|
||||
def test_mqtt_ip_resolution_success(mocker):
|
||||
# Force socket.gethostbyname to simulate a working DNS
|
||||
mocker.patch("socket.gethostbyname", return_value="10.11.0.33")
|
||||
|
||||
mqtt_client = MQTT()
|
||||
assert mqtt_client.IP == "10.11.0.33"
|
||||
|
||||
|
||||
def test_mqtt_ip_resolution_fallback(mocker, caplog):
|
||||
# Simulate a DNS timeout (like what happened to you earlier!)
|
||||
mocker.patch("socket.gethostbyname", side_effect=socket.timeout("DNS down"))
|
||||
|
||||
mqtt_client = MQTT()
|
||||
|
||||
# Assert it fell back to the static IP from your .env / constants
|
||||
assert mqtt_client.IP == "10.11.0.33"
|
||||
assert "Falling back to static IP" in caplog.text
|
||||
|
||||
|
||||
def test_mqtt_send_error_handling(mocker, caplog):
|
||||
# Mock the actual publish function to throw a connection error
|
||||
mocker.patch(
|
||||
"tools.mqtt.mqtt_publish", side_effect=ConnectionRefusedError("Broker offline")
|
||||
)
|
||||
|
||||
mqtt_client = MQTT()
|
||||
mqtt_client.send({"temp": 20})
|
||||
|
||||
# Assert it didn't crash, and logged the error
|
||||
assert "MQTT data couldn't be sent" in caplog.text
|
||||
@@ -0,0 +1,68 @@
|
||||
# tests/test_smhi.py
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
from datetime import datetime
|
||||
from environment.smhi import SMHI
|
||||
from tools.geometry import Coordinate
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def smhi_client():
|
||||
session = requests.Session()
|
||||
location = Coordinate(56.2006, 12.5553)
|
||||
return SMHI(location, session)
|
||||
|
||||
|
||||
def test_smhi_fire_warning_success(requests_mock, smhi_client):
|
||||
# Mock a successful API response
|
||||
mock_response = {
|
||||
"timeSeries": [{"parameters": [{"name": "fwiindex", "values": [4]}]}]
|
||||
}
|
||||
# Intercept the exact URL your code generates
|
||||
url = smhi_client.fire_warning_api.format(lon=12.5553, lat=56.2006)
|
||||
requests_mock.get(url, json=mock_response, status_code=200)
|
||||
|
||||
# Run the function
|
||||
result = smhi_client._get_fire_warning()
|
||||
|
||||
# Assert it succeeded and parsed the data correctly
|
||||
assert result is True
|
||||
assert smhi_client._data["fire_index"]["data"] == 4
|
||||
assert isinstance(smhi_client._data["fire_index"]["valid_time"], datetime)
|
||||
|
||||
|
||||
def test_smhi_fire_warning_http_error(requests_mock, smhi_client, caplog):
|
||||
# Mock SMHI's servers crashing (500 Internal Server Error)
|
||||
url = smhi_client.fire_warning_api.format(lon=12.5553, lat=56.2006)
|
||||
requests_mock.get(url, status_code=500)
|
||||
|
||||
result = smhi_client._get_fire_warning()
|
||||
|
||||
# Assert the function returned False instead of crashing the app
|
||||
assert result is False
|
||||
# Assert that our logger caught the HTTP error
|
||||
assert "Network error getting SMHI Fire index" in caplog.text
|
||||
|
||||
|
||||
def test_smhi_fire_warning_timeout(requests_mock, smhi_client, caplog):
|
||||
# Mock a network timeout
|
||||
url = smhi_client.fire_warning_api.format(lon=12.5553, lat=56.2006)
|
||||
requests_mock.get(url, exc=requests.exceptions.ConnectTimeout)
|
||||
|
||||
result = smhi_client._get_fire_warning()
|
||||
|
||||
assert result is False
|
||||
assert "Network error getting SMHI Fire index" in caplog.text
|
||||
|
||||
|
||||
def test_smhi_bad_json_parsing(requests_mock, smhi_client, caplog):
|
||||
# Mock SMHI returning incomplete/bad JSON that is missing the "parameters" key
|
||||
mock_response = {"timeSeries": [{"wrong_key": "data"}]}
|
||||
url = smhi_client.fire_warning_api.format(lon=12.5553, lat=56.2006)
|
||||
requests_mock.get(url, json=mock_response, status_code=200)
|
||||
|
||||
result = smhi_client._get_fire_warning()
|
||||
|
||||
assert result is False
|
||||
assert "Parsing error for SMHI Fire index" in caplog.text
|
||||
+14
-16
@@ -4,35 +4,33 @@ from datetime import datetime, timedelta
|
||||
import math
|
||||
|
||||
|
||||
def julian_date(date: datetime | None = None):
|
||||
def julian_date(date: datetime | None = None) -> float:
|
||||
"""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
|
||||
time_ms = date.timestamp() * 1000
|
||||
return (time_ms / 86400000) + 2440587.5
|
||||
|
||||
|
||||
def CJDN(date: datetime | None = None):
|
||||
def CJDN(date: datetime | None = None) -> int:
|
||||
return round(julian_date(date))
|
||||
|
||||
|
||||
def current_julian_date():
|
||||
def current_julian_date() -> float:
|
||||
"""Returns current julian date"""
|
||||
return julian_date()
|
||||
|
||||
|
||||
def tomorrow_julian_date():
|
||||
def tomorrow_julian_date() -> float:
|
||||
"""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):
|
||||
def future_julian_date(
|
||||
hoursadd: int = 0, minutesadd: int = 0, secondsadd: int = 0
|
||||
) -> float:
|
||||
"""defaults to current if no specifications made. HOURS, MINUTES, AND SECONDS CANNOT BE NEGATIVE"""
|
||||
date = datetime.now() + timedelta(
|
||||
hours=hoursadd, minutes=minutesadd, seconds=secondsadd
|
||||
@@ -40,19 +38,19 @@ def future_julian_date(hoursadd=0, minutesadd=0, secondsadd=0):
|
||||
return julian_date(date)
|
||||
|
||||
|
||||
def epoch_days(date: datetime | None = None):
|
||||
def epoch_days(date: datetime | None = None) -> float:
|
||||
"""returns days since Jan 1st, 2000. Negative if before this date"""
|
||||
return julian_date(date) - 2451545
|
||||
return julian_date(date) - 2451545.0
|
||||
|
||||
|
||||
def day_percent(date: datetime | None = None):
|
||||
def day_percent(date: datetime | None = None) -> float:
|
||||
"""Returns decimal portion of Julian Date"""
|
||||
whole = julian_date(date)
|
||||
return whole - math.floor(whole)
|
||||
|
||||
|
||||
def from_julian(j):
|
||||
def from_julian(j: float) -> datetime:
|
||||
"""Returns datetime.datetime object given julian date J"""
|
||||
J1970 = 2440588
|
||||
J1970 = 2440588.0
|
||||
dayMs = 24 * 60 * 60 * 1000
|
||||
return datetime.fromtimestamp((j + 0.5 - J1970) * dayMs / 1000.0)
|
||||
|
||||
Reference in New Issue
Block a user