sensors extended, calculation if restrictions are currently active or not, v0.5.0

Author: myTselection

README.md

@@ -41,9 +41,17 @@ To detect exiting a car, an automation can be defined using sensor.smartphone_ha
 ## Integration
 
 ### Sensors
-- <code>sensor.parking_[origin]</code>: sensor with public parking info at location of `origin`
+- <code>sensor.parking_[origin]_address</code>: sensor with adres linked to location of `origin`
+- <code>sensor.parking_[origin]_days_restrictions</code>: sensor with indication of days on which the city parking is restricted
+- <code>sensor.parking_[origin]_max_stay</code>: sensor with max amount of minutes is allowed to park during the days and time restricted schedule
+- <code>sensor.parking_[origin]_price</code>: sensor with price info for the city parking
+- <code>sensor.parking_[origin]_remarks</code>: sensor with remarks related to the the city parking
+- <code>sensor.parking_[origin]_restriction_active</code>: sensor with indication if the time and day restrictions are currenlty active. Eg if Sunday the city parking is not restricted, the restricted_active sensor will be False
+- <code>sensor.parking_[origin]_time_restrictions</code>: sensor with indication of time schedule duding which the city parking is restricted
+- <code>sensor.parking_[origin]_type</code>: sensor with type of city parking zone (eg disk, paid, free, etc)
+- <code>sensor.parking_[origin]_zone_</code>: sensor with zone name of the city parking (eg green, yellow, orange, red, blue, etc)
 - sensor data will be updated every 5min, unless the coordinates of the origin didn't change
-- <details><summary>Sensor attributes</summary>
+- <details><summary>For now all sensors have the same set of attributes</summary>
 
     | Attribute | Description |
     | --------- | ----------- |
@@ -52,13 +60,26 @@ To detect exiting a car, an automation can be defined using sensor.smartphone_ha
     | `latitude`  | Latitude of the origin |
     | `longitude` | Longitude of the origin |
     | `type`      | Type of the public parking, eg paid |
-    | `time restrictions`    | Time restrictions for the public parking |
-    | `days restrictions`  | Days at which the public parking is limited |
-    | `prices`    | Price for 1 or 2 hours parking |
+    | `time_restrictions`    | Time restrictions for the public parking |
+    | `days_restrictions`  | Days at which the public parking is limited |
+    | `prices`    | Price indication for 1 or 2 hours parking and amount of free parking allowed |
     | `remarkds`  | Extra info related to the public parking |
-    | `maxStay`   | Max time you can stay at the public parking |
+    | `max_stay`   | Max time you can stay at the public parking |
     | `zone`      | Name of the zone, blue/orange/red zone indication of public parking |
     | `address`   | Address for which the parking information is shown |
+    | `time_restriction_active_now` | Time restrictions are currently active |
+    | `day_restriction_active_now` | Day restrictions are currently active |
+    | `maxstay_passed_now` | Max stay has passed |
+    | `maxstay_elapsed` | Time (min) elapsed since max stay started |
+    | `maxstay_remaining` | Time (min) remaining until max stay is reached |
+    | `maxstay_start_time` | Time at which max stay started |
+    | `restriction_active` | Time and day restrictions are currently active |
+    | `last_update` | Last time data was updated |
+    | `last_restriction_check` | Last time restrictions were checked |
+    | `attribution` | Attribution for the data |
+    | `device_class` | Device class |
+    | `icon` | Icon to display |
+    | `friendly_name` | Friendly name for the sensor |
 
     </details>
 

custom_components/cityparking/__init__.py

@@ -73,8 +73,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     coordinator: CityParkingUserDataUpdateCoordinator
     httpx_client = get_async_client(hass)
     routeCalculatorClient = WazeRouteCalculator(region="EU", client=httpx_client)
-    coordinator = CityParkingUserDataUpdateCoordinator(
-        hass, seetyApi, entry, routeCalculatorClient)
+    coordinator = CityParkingUserDataUpdateCoordinator(hass, seetyApi, entry, routeCalculatorClient)
 
     hass.data[DOMAIN][entry.entry_id] = coordinator
     await coordinator.async_config_entry_first_refresh()

custom_components/cityparking/coordinator.py

@@ -1,21 +1,21 @@
 """Shell Recharge data update coordinators."""
 
+from copy import deepcopy
 import logging
 import asyncio
 from asyncio.exceptions import CancelledError
-import re
 
 from aiohttp.client_exceptions import ClientError
 from homeassistant.core import HomeAssistant
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 from homeassistant.helpers.location import find_coordinates
+from homeassistant.util import dt as dt_util
 from .seetyApi import SeetyApi, EmptyResponseError
-from .seetyApi.models import Coords, CityParkingModel, ParkingSensorType, SeetyLocationResponse, SeetyUser
+from .seetyApi.models import *
+from .seetyApi.extract_info import *
 # from .location import LocationSession
 from pywaze.route_calculator import CalcRoutesResponse, WazeRouteCalculator
-from typing import List, Tuple
-
 
 from .const import DOMAIN, UPDATE_INTERVAL,CONF_ORIGIN
 _LOGGER = logging.getLogger(__name__)
@@ -45,210 +45,31 @@ async def async_find_city_parking_info(
 
     return cityParkingInfo.model_dump()
 
-def extract_readable_info(cityParkingInfo: CityParkingModel):
-    rules = cityParkingInfo.rules.model_dump() if cityParkingInfo.rules else {}
-    streetComplete = cityParkingInfo.streetComplete.model_dump() if cityParkingInfo.streetComplete else {}
-    locationResults = cityParkingInfo.location.model_dump().get('results', [{}])[0] if cityParkingInfo.location else {}
-    _LOGGER.debug(f"Sensor _read_coordinator_data rules: {rules}")
-    type = rules.get('rules', {}).get('type', 'unknown')
-    zone_type = rules.get('properties', {}).get('type', 'unknown')
-    display, emoji = name_and_emoji(zone_type)
-    rules_complete_zone = streetComplete.get('rules', {}).get(zone_type, {})
-    address = f"{locationResults.get('formatted_address', '')}, {locationResults.get('countryCode', '')}" if locationResults else ''
-    origin_coordinates = cityParkingInfo.origin_coordinates.model_dump() if cityParkingInfo.origin_coordinates else {}
-    extra_data = {
-        "origin": cityParkingInfo.origin,
-        "latitude": origin_coordinates.get('lat', ''),
-        "longitude": origin_coordinates.get('lon', ''),
-        ParkingSensorType.TYPE.value: type,
-        ParkingSensorType.TIME.value: hours_array_to_string(rules.get('rules', {}).get('hours', [])),
-        ParkingSensorType.DAYS.value: days_to_string(rules.get('rules', {}).get('days', [])),
-        ParkingSensorType.PRICE.value: prices_to_string(rules.get('rules', {}).get('prices', {})),
-        ParkingSensorType.REMARKS.value: " - ".join(rules_complete_zone.get('remarks', "")),
-        ParkingSensorType.MAXSTAY.value: minutes_to_string(rules_complete_zone.get('maxStay', "")),
-        ParkingSensorType.ZONE.value: f"{display} {emoji}",
-        ParkingSensorType.ADDRESS.value: address,
-    }
-    cityParkingInfo.extra_data = extra_data
-
-
-def days_to_string(days):
-    names = ['Sun', 'Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat']
-
-    # normalize & sort
-    sorted_days = sorted(set(days))
-
-    # 7d/7
-    if len(sorted_days) == 7:
-        return '7d/7'
-
-    # weekend (sat + sun)
-    if len(sorted_days) == 2 and 0 in sorted_days and 6 in sorted_days:
-        return 'Sat-Sun'
-
-    # check consecutive
-    if len(sorted_days) < 2:
-        return ",".join(names[d] for d in sorted_days)
-
-    consecutive = all(
-        sorted_days[i] == sorted_days[i - 1] + 1
-        for i in range(1, len(sorted_days))
-    )
-
-    if consecutive:
-        return f"{names[sorted_days[0]]}-{names[sorted_days[-1]]}"
-
-    # fallback: comma-separated list
-    return ",".join(names[d] for d in sorted_days)
-
-
-def hours_array_to_string(hours: List[str]) -> str:
-    if not hours or len(hours) != 2:
-        return ""
-
-    start, end = hours
-
-    # Full-day special case
-    if start == "00:00" and end == "24:00":
-        return "24h/24"
-
-    return f"{start} - {end}"
-
-
-def prices_to_string(prices: dict) -> str:
-    if not prices:
-        return ""
-
-    parts = []
-
-    for hours, price in sorted(prices.items(), key=lambda x: int(x[0])):
-        h = int(hours)
-        if h == 0 and price != 0:
-            parts.append(f"Free: {int(price)}min")
-        if h > 0:
-            parts.append(f"{price}€ ({h}h)")
 
-    return " - ".join(parts)
 
-def minutes_to_string(minutes_str: str) -> str:
-    try:
-        minutes = int(minutes_str)
-    except (ValueError, TypeError):
-        return "0m"  # fallback for invalid input
-
-    if minutes <= 0:
-        return "0m"
-
-    hours = minutes // 60
-    mins = minutes % 60
-
-    if hours == 0:
-        return f"{mins}m"
-    if mins == 0:
-        return f"{hours}h"
-
-    return f"{hours}h {mins}m"
-
-
-# Canonical display name + emoji
-_CANONICAL = {
-    "blue": ("Blue", "🔵"),
-    "orange": ("Orange", "🟠"),
-    "orange-dark": ("Orange (dark)", "🟠"),
-    "orange-2": ("Orange (variant)", "🟠"),
-    "pedestrian": ("Pedestrian", "🚶"),
-    "pink": ("Pink", "🩷"),
-    "red": ("Red", "🔴"),
-    "resident": ("Resident", "🏠"),
-    "yellow": ("Yellow", "🟡"),
-    "yellow-dark": ("Yellow (dark)", "🟡"),
-    "yellow-dotted": ("Yellow (dotted)", "🟡"),
-    "yellow-dark-dotted": ("Yellow (dark, dotted)", "🟡"),
-    "no-parking": ("No parking", "🚫"),
-    "freeinv": ("Free", "🆓"),       # best-effort interpretation
-    "disabled": ("Disabled", "♿"),
-}
-
-# Aliases mapped to canonical keys (add more aliases here as needed)
-_ALIASES = {
-    "blue": ["blue"],
-    "freeinv": ["freeinv", "free-inv", "free_inv", "free", "inv"],
-    "no-parking": ["noparking", "no-parking", "no_parking", "no parking"],
-    "orange": ["orange", "oranged", "orange1"],
-    "orange-dark": ["orangedark", "orange-dark", "orange_dark"],
-    "orange-2": ["orange-2", "orange2", "orange variant"],
-    "pedestrian": ["pedestrian", "pedestrain"],  # common misspelling included
-    "pink": ["pink"],
-    "red": ["red"],
-    "resident": ["resident", "residents", "residentship"],
-    "yellow": ["yellow"],
-    "yellow-dark": ["yellowdark", "yellow-dark", "yellow_dark"],
-    "yellow-dotted": ["yellowdotted", "yellow-dotted", "yellow_dotted"],
-    "yellow-dark-dotted": [
-        "yellowdarkdotted",
-        "yellow-dark-dotted",
-        "yellow_dark_dotted",
-    ],
-    "disabled": ["disabled", "disability", "wheelchair", "wheel-chair"],
-}
-
-# Build quick lookup dict from alias -> canonical
-_ALIAS_LOOKUP = {}
-for canonical_key, aliases in _ALIASES.items():
-    for a in aliases:
-        # store several normalized variants for each alias
-        norm = a.lower()
-        _ALIAS_LOOKUP[norm] = canonical_key
-        _ALIAS_LOOKUP[re.sub(r"[^a-z0-9]", "", norm)] = canonical_key  # compact form
-        _ALIAS_LOOKUP[norm.replace("-", " ")] = canonical_key
-
-
-def _normalize_key(raw: str) -> str:
-    """Normalize a raw input key into a canonical lookup form."""
-    if raw is None:
-        return ""
-    s = str(raw).strip().lower()
-    # remove surrounding quotes if any
-    s = s.strip("\"'` ")
-    # collapse whitespace and common separators to single hyphen
-    s = re.sub(r"[ _]+", "-", s)
-    s = re.sub(r"[^a-z0-9\-]", "", s)
-    # special-case trailing 's' (plural) -> try singular
-    if s.endswith("s") and (s[:-1] in _ALIAS_LOOKUP or s[:-1] in _CANONICAL):
-        s = s[:-1]
-    return s
-
-
-def name_and_emoji(raw_name: str) -> Tuple[str, str]:
-    """
-    Return a tuple (clean_display_name, emoji) for a raw key like "blue" or "orange-2".
-    Falls back to capitalized raw_name and a default emoji if unknown.
-    """
-    norm = _normalize_key(raw_name)
-    # direct canonical match
-    if norm in _CANONICAL:
-        return _CANONICAL[norm]
-
-    # alias lookup
-    if norm in _ALIAS_LOOKUP:
-        canon = _ALIAS_LOOKUP[norm]
-        return _CANONICAL.get(canon, (canon.capitalize(), "🔖"))
-
-    # try compact form (remove hyphens)
-    compact = re.sub(r"[^a-z0-9]", "", norm)
-    if compact in _ALIAS_LOOKUP:
-        canon = _ALIAS_LOOKUP[compact]
-        return _CANONICAL.get(canon, (canon.capitalize(), "🔖"))
-
-    # try removing trailing digits (e.g., "orange2" -> "orange")
-    no_digits = re.sub(r"\d+$", "", compact)
-    if no_digits in _ALIAS_LOOKUP:
-        canon = _ALIAS_LOOKUP[no_digits]
-        return _CANONICAL.get(canon, (canon.capitalize(), "🔖"))
+def update_restriction_status(cityParkingInfo: CityParkingModel, update_time: dt_util.dt.datetime = None):
+    """Update restriction status based on current time."""
+    time_restriction = cityParkingInfo.extra_data.get(ParkingSensorType.TIME.value + "_src", None)
+    days_restriction = cityParkingInfo.extra_data.get(ParkingSensorType.DAYS.value + "_src", None)
+    max_stay = cityParkingInfo.extra_data.get(ParkingSensorType.MAXSTAY.value + "_src", None)
+    is_active_now = is_hours_active_now(time_restriction)
+    is_active_today = is_days_active_today(days_restriction)
+    is_restriction_active = is_active_now and is_active_today
+    is_max_stay_passed_value, max_stay_elapsed, max_stay_remaining = is_max_stay_passed(start_dt=update_time, max_minutes=max_stay)
+    extra_data = cityParkingInfo.extra_data if cityParkingInfo.extra_data else {}
+    extra_data.update({
+        TIME_RESTRICTION_ACTIVE_NOW: is_active_now,
+        DAY_RESTRICTION_ACTIVE_NOW: is_active_today,
+        MAXSTAY_PASSED_NOW: is_max_stay_passed_value,
+        MAXSTAY_ELAPSED: max_stay_elapsed,
+        MAXSTAY_REMAINING: max_stay_remaining,
+        MAXSTAY_START_TIME: update_time,
+        RESTRICTION_ACTIVE: is_restriction_active,
+        LAST_UPDATE: update_time,
+        LAST_RESTRICTION_CHECK: dt_util.now(),
+    })
+    cityParkingInfo.extra_data = extra_data
 
-    # final fallback: pretty-print the raw string and use a neutral emoji
-    pretty = raw_name.strip().replace("_", " ").replace("-", " ").title()
-    return pretty, "🔖"
 
 class CityParkingUserDataUpdateCoordinator(DataUpdateCoordinator):
     """Handles data updates for public chargers."""
@@ -272,8 +93,26 @@ def __init__(
         self._previousResults : CityParkingModel = None
         self._previousCoordinates : Coords = None
         self._previousResultAge = 0
-
-
+        self._previousUpdate = None
+
+    # async def async_set_stay_start(self, unique_id: str, start_iso: str) -> None:
+    #     """Set stay_start for a specific parking entry and notify listeners."""
+    #     new_data = deepcopy(self.data)
+
+    #     self._previousUpdate = start_iso
+    #     self._previousResultAge = 0
+    #     self._previousResults = new_data.get(unique_id, {})
+
+    #     # Example for dict-based data:
+    #     item = dict(new_data.get(unique_id, {}))
+    #     extra = dict(item.get("extra_data", {}))
+    #     extra["stay_start"] = start_iso
+    #     item["extra_data"] = extra
+    #     new_data[unique_id] = item
+
+    #     # Atomically set new coordinator data which triggers updates
+    #     await self.async_set_updated_data(new_data)
+        
     async def _async_update_data(self):
         """Fetch data from API endpoint.
 
@@ -286,9 +125,13 @@ async def _async_update_data(self):
         origin_coordinates_json = await self._routeCalculatorClient._ensure_coords(resolved_origin)
         origin_coordinates = Coords.model_validate(origin_coordinates_json)
         _LOGGER.info(f"coordinator origin_coordinates: {origin_coordinates}, resolved_origin: {resolved_origin}, origin: {self._origin}, previousCoordinates: {self._previousCoordinates}")
+
+
         self._previousResultAge += 1
         if self._previousResults is not None and self._previousCoordinates == origin_coordinates and (self._previousResultAge < MAX_RESULT_AGE):
             _LOGGER.debug("Coordinator _async_update_data using cached previousResults, no coordinate change detected.")
+            # extend / update seety info with actual restriction status
+            update_restriction_status(self._previousResults, self._previousUpdate)
             return self._previousResults
         try:
             data = await self._seetyApi.getAddressSeetyInfo(origin_coordinates)
@@ -299,6 +142,9 @@ async def _async_update_data(self):
             self._previousResults = data
             self._previousCoordinates = origin_coordinates
             self._previousResultAge = 0
+            self._previousUpdate = dt_util.now()
+            # extend / update seety info with actual restriction status
+            update_restriction_status(data, self._previousUpdate)
         except EmptyResponseError as exc:
             _LOGGER.error(
                 "EmptyResponseError occurred while fetching data for %s (%s): %s",

custom_components/cityparking/manifest.json

@@ -12,5 +12,5 @@
   "issue_tracker": "https://github.com/myTselection/CityParking/issues",
   "loggers": ["pywaze", "homeassistant.helpers.location"],
   "requirements": ["pywaze==1.1.1"],
-  "version": "0.4.0"
+  "version": "0.5.0"
 }