Detailed documentation of all internal modules

Todo

This is not yet complete, see issue #52

app

flexmeasures.app.create(env: str | None = None, path_to_config: str | None = None, plugins: List[str] | None = None) → Flask

Create a Flask app and configure it.

Set the environment by setting FLASK_ENV as environment variable (also possible in .env). Or, overwrite any FLASK_ENV setting by passing an env in directly (useful for testing for instance).

A path to a config file can be passed in (otherwise a config file will be searched in the home or instance directories).

Also, a list of plugins can be set. Usually this works as a config setting, but this is useful for automated testing.

data.models

exception flexmeasures.data.models.ModelException

utils

Utilities for the FlexMeasures project.

utils.calculations

Calculations

flexmeasures.utils.calculations.integrate_time_series(series: pd.Series, initial_stock: float, up_efficiency: float | pd.Series = 1, down_efficiency: float | pd.Series = 1, decimal_precision: int | None = None) → pd.Series

Integrate time series of length n and inclusive=”left” (representing a flow) to a time series of length n+1 and inclusive=”both” (representing a stock), given an initial stock (i.e. the constant of integration). The unit of time is hours: i.e. the stock unit is flow unit times hours (e.g. a flow in kW becomes a stock in kWh). Optionally, set a decimal precision to round off the results (useful for tests failing over machine precision).

>>> s = pd.Series([1, 2, 3, 4], index=pd.date_range(datetime(2001, 1, 1, 5), datetime(2001, 1, 1, 6), freq=timedelta(minutes=15), inclusive="left"))
>>> integrate_time_series(s, 10)
    2001-01-01 05:00:00    10.00
    2001-01-01 05:15:00    10.25
    2001-01-01 05:30:00    10.75
    2001-01-01 05:45:00    11.50
    2001-01-01 06:00:00    12.50
    Freq: D, dtype: float64

>>> s = pd.Series([1, 2, 3, 4], index=pd.date_range(datetime(2001, 1, 1, 5), datetime(2001, 1, 1, 7), freq=timedelta(minutes=30), inclusive="left"))
>>> integrate_time_series(s, 10)
    2001-01-01 05:00:00    10.0
    2001-01-01 05:30:00    10.5
    2001-01-01 06:00:00    11.5
    2001-01-01 06:30:00    13.0
    2001-01-01 07:00:00    15.0
    dtype: float64

utils.time_utils

flexmeasures.utils.time_utils.as_server_time(dt: datetime) → datetime: The datetime represented in the timezone of the FlexMeasures platform. If dt is naive, we assume it is UTC time.

flexmeasures.utils.time_utils.decide_resolution(start: datetime | None, end: datetime | None) → str: Decide on a practical resolution given the length of the selected time period. Useful for querying or plotting.

flexmeasures.utils.time_utils.determine_minimum_resampling_resolution(event_resolutions: list[datetime.timedelta]) → timedelta: Return minimum non-zero event resolution, or zero resolution if none of the event resolutions is non-zero.

flexmeasures.utils.time_utils.duration_isoformat(duration: timedelta)

Adapted version of isodate.duration_isoformat for formatting a datetime.timedelta.

The difference is that absolute days are not formatted as nominal days. Workaround for https://github.com/gweis/isodate/issues/74.

flexmeasures.utils.time_utils.ensure_local_timezone(dt: Timestamp | datetime, tz_name: str = 'Europe/Amsterdam') → Timestamp | datetime: If no timezone is given, assume the datetime is in the given timezone and make it explicit. Otherwise, if a timezone is given, convert to that timezone.

flexmeasures.utils.time_utils.forecast_horizons_for(resolution: str | timedelta) → List[str] | List[timedelta]: Return a list of horizons that are supported per resolution. Return values or of the same type as the input.

flexmeasures.utils.time_utils.freq_label_to_human_readable_label(freq_label: str) → str: Translate pandas frequency labels to human-readable labels.

flexmeasures.utils.time_utils.get_max_planning_horizon(resolution: timedelta) → timedelta | None: Determine the maximum planning horizon for the given sensor resolution.

flexmeasures.utils.time_utils.get_most_recent_clocktime_window(window_size_in_minutes: int, now: datetime | None = None, grace_period_in_seconds: int | None = 0) → Tuple[datetime, datetime]

Calculate a recent time window, returning a start and end minute so that a full hour can be filled with such windows, e.g.:

Calling this function at 15:01:xx with window size 5 -> (14:55:00, 15:00:00) Calling this function at 03:36:xx with window size 15 -> (03:15:00, 03:30:00)

We can demand a grace period (of x seconds) to have passed before we are ready to accept that we’re in a new window: Calling this function at 15:00:16 with window size 5 and grace period of 30 seconds -> (14:50:00, 14:55:00)

window_size_in_minutes is assumed to > 0 and < = 60, and a divisor of 60 (1, 2, …, 30, 60).

If now is not given, the current server time is used. if now / the current time lies within a boundary minute (e.g. 15 when window_size_in_minutes=5), then the window is not deemed over and the previous one is returned (in this case, [5, 10])

Returns two datetime objects. They’ll be in the timezone (if given) of the now parameter, or in the server timezone (see FLEXMEASURES_TIMEZONE setting).

flexmeasures.utils.time_utils.get_timezone(of_user=False) → BaseTzInfo: Return the FlexMeasures timezone, or if desired try to return the timezone of the current user.

flexmeasures.utils.time_utils.localized_datetime(dt: datetime) → datetime: Localise a datetime to the timezone of the FlexMeasures platform. Note: this will change nothing but the tzinfo field.

flexmeasures.utils.time_utils.localized_datetime_str(dt: datetime, dt_format: str = '%Y-%m-%d %I:%M %p') → str

Localise a datetime to the timezone of the FlexMeasures platform. If no datetime is passed in, use server_now() as basis.

Hint: This can be set as a jinja filter, so we can display local time in the app, e.g.: app.jinja_env.filters[‘localized_datetime’] = localized_datetime_str

flexmeasures.utils.time_utils.naive_utc_from(dt: datetime) → datetime: Return a naive datetime, that is localised to UTC if it has a timezone. If dt is naive, we assume it is already in UTC time.

flexmeasures.utils.time_utils.naturalized_datetime_str(dt: datetime | None, now: datetime | None = None) → str

Naturalise a datetime object (into a human-friendly string). The dt parameter (as well as the now parameter if you use it) can be either naive or tz-aware. We assume UTC in the naive case.

We use the the humanize library to generate a human-friendly string. If dt is not longer ago than 24 hours, we use humanize.naturaltime (e.g. “3 hours ago”), otherwise humanize.naturaldate (e.g. “one week ago”)

Hint: This can be set as a jinja filter, so we can display local time in the app, e.g.: app.jinja_env.filters[‘naturalized_datetime’] = naturalized_datetime_str

flexmeasures.utils.time_utils.resolution_to_hour_factor(resolution: str | timedelta) → float

Return the factor with which a value needs to be multiplied in order to get the value per hour, e.g. 10 MW at a resolution of 15min are 2.5 MWh per time step.

Parameters:: resolution – timedelta or pandas offset such as “15T” or “1H”

flexmeasures.utils.time_utils.server_now() → datetime: The current time (timezone aware), converted to the timezone of the FlexMeasures platform.

flexmeasures.utils.time_utils.to_http_time(dt: pd.Timestamp | datetime) → str

Formats datetime using the Internet Message Format fixdate.

>>> to_http_time(pd.Timestamp("2022-12-13 14:06:23Z"))
Tue, 13 Dec 2022 14:06:23 GMT

References

IMF-fixdate: https://www.rfc-editor.org/rfc/rfc7231#section-7.1.1.1

flexmeasures.utils.time_utils.tz_index_naively(data: DataFrame | Series | DatetimeIndex) → DataFrame | Series | DatetimeIndex: Turn any DatetimeIndex into a tz-naive one, then return. Useful for bokeh, for instance.

ui.utils.view_utils

ui.utils.plotting_utils

ui.views

flexmeasures.ui.views.control.control_view(): Control view. This page lists balancing opportunities for a selected time window. The user can place manual orders or choose to automate the ordering process.