Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions functions-python/helpers/locations.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,19 @@
from enum import Enum
from typing import Dict, Optional
from sqlalchemy.orm import Session
import pycountry
from shared.database_gen.sqlacodegen_models import Feed, Location
import logging


class ReverseGeocodingStrategy(str, Enum):
"""
Enum for reverse geocoding strategies.
"""

PER_POINT = "per-point"


def get_country_code(country_name: str) -> Optional[str]:
"""
Get ISO 3166 country code from country name
Expand Down
5 changes: 4 additions & 1 deletion functions-python/helpers/logger.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,9 @@ def init_logger():
Initializes the logger with level INFO if not set in the environment.
On cloud environment it will also initialize the GCP logger.
"""
logging.basicConfig(level=get_env_logging_level())
logging_level = get_env_logging_level()
logging.basicConfig(level=logging_level)
logging.info("Logger initialized with level: %s", logging_level)
global _logging_initialized
if not is_local_env() and not _logging_initialized:
# Avoids initializing the logs multiple times due to performance concerns
Expand Down Expand Up @@ -81,6 +83,7 @@ def get_logger(name: str, stable_id: str = None):
if stable_id
else logging.getLogger(name)
)
logger.setLevel(level=get_env_logging_level())
if stable_id and not any(
isinstance(log_filter, StableIdFilter) for log_filter in logger.filters
):
Expand Down
65 changes: 65 additions & 0 deletions functions-python/helpers/runtime_metrics.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,65 @@
import functools
import time
import tracemalloc
import psutil
import logging


def track_metrics(metrics=("time", "memory", "cpu")):
"""Decorator to track specified metrics (time, memory, cpu) during function execution.
The decorator logs the metrics using the provided logger or a default logger if none is provided.
Args:
metrics (tuple): Metrics to track. Options are "time", "memory", "cpu".
Usage:
@track_metrics(metrics=("time", "memory", "cpu"))
def example_function():
data = [i for i in range(10**6)] # Simulate work
time.sleep(1) # Simulate delay
return sum(data)
"""

def decorator(funct):
@functools.wraps(funct)
def wrapper(*args, **kwargs):
logger = kwargs.get("logger")
if not logger:
# Use a default logger if none is provided
logger = logging.getLogger(funct.__name__)

process = psutil.Process()
tracemalloc.start() if "memory" in metrics else None
start_time = time.time() if "time" in metrics else None
cpu_before = (
process.cpu_percent(interval=None) if "cpu" in metrics else None
)

try:
result = funct(*args, **kwargs)
except Exception as e:
logger.error(f"Function '{funct.__name__}' raised an exception: {e}")
raise
finally:
metrics_message = ""
if "time" in metrics:
duration = time.time() - start_time
metrics_message = f"time: {duration:.2f} seconds"
if "memory" in metrics:
current, peak = tracemalloc.get_traced_memory()
tracemalloc.stop()
if metrics_message:
metrics_message += ", "
metrics_message += f"memory: {current / (1024 ** 2):.2f} MB (peak: {peak / (1024 ** 2):.2f} MB)"
if "cpu" in metrics:
cpu_after = process.cpu_percent(interval=None)
if metrics_message:
metrics_message += ", "
metrics_message += f"cpu: {cpu_after - cpu_before:.2f}%"
if len(metrics_message) > 0:
logger.info(
"Function metrics('%s'): %s", funct.__name__, metrics_message
)
return result

return wrapper

return decorator
2 changes: 1 addition & 1 deletion functions-python/helpers/tests/test_transform.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ def test_to_boolean():
assert to_boolean("0") is False
assert to_boolean("no") is False
assert to_boolean("n") is False
assert to_boolean(1) is False
assert to_boolean(1) is True
assert to_boolean(0) is False
assert to_boolean(None) is False
assert to_boolean([]) is False
Expand Down
28 changes: 25 additions & 3 deletions functions-python/helpers/transform.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,15 +16,17 @@
from typing import List, Optional


def to_boolean(value):
def to_boolean(value, default_value: Optional[bool] = False) -> bool:
"""
Convert a value to a boolean.
"""
if isinstance(value, bool):
return value
if isinstance(value, (int, float)):
return value != 0
if isinstance(value, str):
return value.lower() in ["true", "1", "yes", "y"]
return False
return value.strip().lower() in ["true", "1", "yes", "y"]
return default_value


def get_nested_value(
Expand Down Expand Up @@ -53,3 +55,23 @@ def get_nested_value(
result = current_data.strip()
return result if result else default_value
return current_data


def to_enum(value, enum_class=None, default_value=None):
"""
Convert a value to an enum member of the specified enum class.

Args:
value: The value to convert.
enum_class: The enum class to convert the value to.
default_value: The default value to return if conversion fails.

Returns:
An enum member if conversion is successful, otherwise the default value.
"""
if enum_class and isinstance(value, enum_class):
return value
try:
return enum_class(str(value))
except (ValueError, TypeError):
return default_value
1 change: 1 addition & 0 deletions functions-python/reverse_geolocation/.coveragerc
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ omit =
*/helpers/*
*/database_gen/*
*/shared/*
*/scripts/*

[report]
exclude_lines =
Expand Down
2 changes: 2 additions & 0 deletions functions-python/reverse_geolocation/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,8 @@ This function performs the core reverse geolocation logic. It processes location
- `vehicle_status_url`: Required if `data_type` is `gbfs` and `station_information_url` and `free_bike_status_url` are omitted. URL of the GBFS `vehicle_status.json` file.
- `free_bike_status_url`: Required if `data_type` is `gbfs` and `station_information_url` and `vehicle_status_url` are omitted. URL of the GBFS `free_bike_status.json` file.
- `data_type`: Optional. Specifies the type of data being processed. Can be `gtfs` or `gbfs`. If not provided, the function will attempt to determine the type based on the URLs provided.
- `strategy`: Optional. Specifies the reverse geolocation strategy to use. Defaults to `per-point`.
- `public`: Optional. Indicates whether the resulting geojson files will be public or private. Defaults to `true`.

### Processing Steps:

Expand Down
1 change: 1 addition & 0 deletions functions-python/reverse_geolocation/requirements.txt
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ shapely
gtfs-kit
matplotlib
jsonpath_ng
psutil

# Configuration
python-dotenv==1.0.0
4 changes: 3 additions & 1 deletion functions-python/reverse_geolocation/requirements_dev.txt
Original file line number Diff line number Diff line change
@@ -1,4 +1,6 @@
Faker
pytest~=7.4.3
urllib3-mock
requests-mock
requests-mock
gcp-storage-emulator
geopandas
17 changes: 16 additions & 1 deletion functions-python/reverse_geolocation/src/parse_request.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,9 @@
import requests
from jsonpath_ng import parse

from shared.helpers.locations import ReverseGeocodingStrategy
from shared.helpers.transform import to_boolean, to_enum


def parse_request_parameters(
request: flask.Request,
Expand Down Expand Up @@ -45,7 +48,19 @@ def parse_request_parameters(
raise ValueError(
f"Invalid data_type '{data_type}'. Supported types are 'gtfs' and 'gbfs'."
)
return df, stable_id, dataset_id, data_type, urls
public = True
if "public" in request_json:
public = to_boolean(request_json["public"], default_value=True)
strategy = ReverseGeocodingStrategy.PER_POINT
if "strategy" in request_json:
strategy = to_enum(
value=request_json["strategy"],
default_value=ReverseGeocodingStrategy.PER_POINT,
)
else:
logging.info("No strategy provided, using default")
logging.info("Strategy set to: %s.", strategy)
return df, stable_id, dataset_id, data_type, urls, public, strategy


def parse_request_parameters_gtfs(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,13 @@ def parse_resource_data(data: dict) -> Tuple[str, str, str]:
def reverse_geolocation_pubsub(request: CloudEvent) -> None:
"""
Reverse geolocation function triggered by a Pub/Sub message.
@:request: CloudEvent containing the Pub/Sub message data. Example data:
{
"stable_id": "example_stable_id",
"dataset_id": "example_dataset_id",
"url": "https://example.com/path/to/feed.zip"
}

"""
try:
init(request)
Expand Down
Loading