11741252103

Committed 08 Nov 2024 11:25AM UTC coverage: 67.101% (-0.2%) from 67.328%

Build # 11741252103

Build Type

push

github

Committed by

web-flow

Commit Message

Chore/python 3.12 (#1233)

* feat: test for Python 3.12

* feat: packages for Python 3.12

* fix: called_once_with should be assert_called_once_with

Signed-off-by: F.N. Claessen <felix@seita.nl>

* fix: don't assert assert_method

Signed-off-by: F.N. Claessen <felix@seita.nl>

* docs: changelog entry

Signed-off-by: F.N. Claessen <felix@seita.nl>

* docs: add Python 3.12 classifier in setup.py

Signed-off-by: F.N. Claessen <felix@seita.nl>

* fix: getting the status of a not yet queued job now results in an InvalidJobOperation; see https://github.com/rq/rq/pull/2039

Signed-off-by: F.N. Claessen <felix@seita.nl>

---------

Signed-off-by: F.N. Claessen <felix@seita.nl>

Run Details

6758 of 10948 branches covered (61.73%)

Branch coverage included in aggregate %.

3 of 5 new or added lines in 1 file covered. (60.0%)

2 existing lines in 1 file now uncovered.

7982 of 11019 relevant lines covered (72.44%)

7.16 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

83.29

/flexmeasures/data/services/scheduling.py

"""
Logic around scheduling (jobs)
"""

from __future__ import annotations

from datetime import datetime, timedelta
import os
import sys
import importlib.util
from importlib.abc import Loader
from typing import Type
import inspect


from flask import current_app
import click
from rq import get_current_job, Callback
from rq.exceptions import InvalidJobOperation
from rq.job import Job
import timely_beliefs as tb
import pandas as pd
from sqlalchemy import select

from flexmeasures.data import db
from flexmeasures.data.models.planning import Scheduler, SchedulerOutputType
from flexmeasures.data.models.planning.storage import StorageScheduler
from flexmeasures.data.models.planning.exceptions import InfeasibleProblemException
from flexmeasures.data.models.planning.process import ProcessScheduler
from flexmeasures.data.models.time_series import Sensor, TimedBelief
from flexmeasures.data.models.generic_assets import GenericAsset as Asset
from flexmeasures.data.models.data_sources import DataSource
from flexmeasures.data.utils import get_data_source, save_to_db
from flexmeasures.utils.time_utils import server_now
from flexmeasures.data.services.utils import (
    job_cache,
    get_asset_or_sensor_ref,
    get_asset_or_sensor_from_ref,
    get_scheduler_instance,
)


def load_custom_scheduler(scheduler_specs: dict) -> type:
    """
    Read in custom scheduling spec.
    Attempt to load the Scheduler class to use.

    The scheduler class should be derived from flexmeasures.data.models.planning.Scheduler.
    The scheduler class should have a class method named "compute".

    Example specs:

    {
        "module": "/path/to/module.py",  # or sthg importable, e.g. "package.module"
        "class": "NameOfSchedulerClass",
    }

    """
    assert isinstance(
        scheduler_specs, dict
    ), f"Scheduler specs is {type(scheduler_specs)}, should be a dict"
    assert "module" in scheduler_specs, "scheduler specs have no 'module'."
    assert "class" in scheduler_specs, "scheduler specs have no 'class'"

    scheduler_name = scheduler_specs["class"]

    # import module
    module_descr = scheduler_specs["module"]
    if os.path.exists(module_descr):
        spec = importlib.util.spec_from_file_location(scheduler_name, module_descr)
        assert spec, f"Could not load specs for scheduling module at {module_descr}."
        module = importlib.util.module_from_spec(spec)
        sys.modules[scheduler_name] = module
        assert isinstance(spec.loader, Loader)
        spec.loader.exec_module(module)
    else:  # assume importable module
        try:
            module = importlib.import_module(module_descr)
        except TypeError as te:
            current_app.logger.error(f"Cannot load {module_descr}: {te}.")
            raise
        except ModuleNotFoundError:
            current_app.logger.error(
                f"Attempted to import module {module_descr} (as it is not a valid file path), but it is not installed."
            )
            raise
        assert module, f"Module {module_descr} could not be loaded."

    # get scheduling function
    assert hasattr(
        module, scheduler_specs["class"]
    ), f"Module at {module_descr} has no class {scheduler_specs['class']}"

    scheduler_class = getattr(module, scheduler_specs["class"])
    schedule_function_name = "compute"
    if not hasattr(scheduler_class, schedule_function_name):
        raise NotImplementedError(
            f"No function {schedule_function_name} in {scheduler_class}. Cannot load custom scheduler."
        )
    return scheduler_class


def trigger_optional_fallback(job, connection, type, value, traceback):
    """Create a fallback schedule job when the error is of type InfeasibleProblemException"""

    job.meta["exception"] = value
    job.save_meta()

    if type is InfeasibleProblemException:
        asset_or_sensor = get_asset_or_sensor_from_ref(job.meta.get("asset_or_sensor"))

        scheduler_kwargs = job.meta["scheduler_kwargs"]

        if ("scheduler_specs" in job.kwargs) and (
            job.kwargs["scheduler_specs"] is not None
        ):
            scheduler_class: Type[Scheduler] = load_custom_scheduler(
                job.kwargs["scheduler_specs"]
            )
        else:
            scheduler_class: Type[Scheduler] = find_scheduler_class(asset_or_sensor)

        # only schedule a fallback schedule job if the original job has a fallback
        # mechanism
        if scheduler_class.fallback_scheduler_class is not None:
            scheduler_class = scheduler_class.fallback_scheduler_class
            scheduler_specs = {
                "class": scheduler_class.__name__,
                "module": inspect.getmodule(scheduler_class).__name__,
            }

            fallback_job = create_scheduling_job(
                asset_or_sensor,
                force_new_job_creation=True,
                enqueue=False,
                scheduler_specs=scheduler_specs,
                **scheduler_kwargs,
            )

            job.meta["fallback_job_id"] = fallback_job.id
            job.save_meta()
            current_app.queues["scheduling"].enqueue_job(fallback_job)


@job_cache("scheduling")
def create_scheduling_job(
    asset_or_sensor: Asset | Sensor | None = None,
    sensor: Sensor | None = None,
    job_id: str | None = None,
    enqueue: bool = True,
    requeue: bool = False,
    force_new_job_creation: bool = False,
    scheduler_specs: dict | None = None,
    **scheduler_kwargs,
) -> Job:
    """
    Create a new Job, which is queued for later execution.

    To support quick retrieval of the scheduling job, the job id is the unique entity address of the UDI event.
    That means one event leads to one job (i.e. actions are event driven).

    As a rule of thumb, keep arguments to the job simple, and deserializable.

    The life cycle of a scheduling job:
    1. A scheduling job is born here (in create_scheduling_job).
    2. It is run in make_schedule which writes results to the db.
    3. If an error occurs (and the worker is configured accordingly), handle_scheduling_exception comes in.

    Arguments:
    :param asset_or_sensor:         asset or sensor for which the schedule is computed
    :param job_id:                  optionally, set a job id explicitly
    :param enqueue:                 if True, enqueues the job in case it is new
    :param requeue:                 if True, requeues the job in case it is not new and had previously failed
                                    (this argument is used by the @job_cache decorator)
    :param force_new_job_creation:  if True, this attribute forces a new job to be created (skipping cache)
                                    (this argument is used by the @job_cache decorator)
    :returns: the job

    """
    # We first create a scheduler and check if deserializing works, so the flex config is checked
    # and errors are raised before the job is enqueued (so users get a meaningful response right away).
    # Note: We are putting still serialized scheduler_kwargs into the job!

    if sensor is not None:
        current_app.logger.warning(
            "The `sensor` keyword argument is deprecated. Please, consider using the argument `asset_or_sensor`."
        )
        asset_or_sensor = sensor

    if scheduler_specs:
        scheduler_class: Type[Scheduler] = load_custom_scheduler(scheduler_specs)
    else:
        scheduler_class: Type[Scheduler] = find_scheduler_class(asset_or_sensor)

    scheduler = get_scheduler_instance(
        scheduler_class=scheduler_class,
        asset_or_sensor=asset_or_sensor,
        scheduler_params=scheduler_kwargs,
    )
    scheduler.deserialize_config()

    asset_or_sensor = get_asset_or_sensor_ref(asset_or_sensor)
    job = Job.create(
        make_schedule,
        kwargs=dict(
            asset_or_sensor=asset_or_sensor,
            scheduler_specs=scheduler_specs,
            **scheduler_kwargs,
        ),
        id=job_id,
        connection=current_app.queues["scheduling"].connection,
        ttl=int(
            current_app.config.get(
                "FLEXMEASURES_JOB_TTL", timedelta(-1)
            ).total_seconds()
        ),
        result_ttl=int(
            current_app.config.get(
                "FLEXMEASURES_PLANNING_TTL", timedelta(-1)
            ).total_seconds()
        ),  # NB job.cleanup docs says a negative number of seconds means persisting forever
        on_failure=Callback(trigger_optional_fallback),
    )

    job.meta["asset_or_sensor"] = asset_or_sensor
    job.meta["scheduler_kwargs"] = scheduler_kwargs
    job.save_meta()

    # in case the function enqueues it
    try:
        job_status = job.get_status(refresh=True)
    except InvalidJobOperation:
        job_status = None

    # with job_status=None, we ensure that only fresh new jobs are enqueued (in the contrary they should be requeued)
    if enqueue and not job_status:
        current_app.queues["scheduling"].enqueue_job(job)
        current_app.job_cache.add(
            asset_or_sensor["id"],
            job.id,
            queue="scheduling",
            asset_or_sensor_type=asset_or_sensor["class"].lower(),
        )

    return job


def make_schedule(
    sensor_id: int | None = None,
    start: datetime | None = None,
    end: datetime | None = None,
    resolution: timedelta | None = None,
    asset_or_sensor: dict | None = None,
    belief_time: datetime | None = None,
    flex_model: dict | None = None,
    flex_context: dict | None = None,
    flex_config_has_been_deserialized: bool = False,
    scheduler_specs: dict | None = None,
) -> bool:
    """
    This function computes a schedule. It returns True if it ran successfully.

    It can be queued as a job (see create_scheduling_job).
    In that case, it will probably run on a different FlexMeasures node than where the job is created.
    In any case, this function expects flex_model and flex_context to not have been deserialized yet.

    This is what this function does:
    - Find out which scheduler should be used & compute the schedule
    - Turn scheduled values into beliefs and save them to db
    """
    # https://docs.sqlalchemy.org/en/13/faq/connections.html#how-do-i-use-engines-connections-sessions-with-python-multiprocessing-or-os-fork
    db.engine.dispose()

    if sensor_id is not None:
        current_app.logger.warning(
            "The `sensor_id` keyword argument is deprecated. Please, consider using the argument `asset_or_sensor`."
        )
        asset_or_sensor = {"class": "Sensor", "id": sensor_id}

    asset_or_sensor: Asset | Sensor = get_asset_or_sensor_from_ref(asset_or_sensor)

    rq_job = get_current_job()
    if rq_job:
        click.echo(
            "Running Scheduling Job %s: %s, from %s to %s"
            % (rq_job.id, asset_or_sensor, start, end)
        )

    if scheduler_specs:
        scheduler_class: Type[Scheduler] = load_custom_scheduler(scheduler_specs)
    else:
        scheduler_class: Type[Scheduler] = find_scheduler_class(asset_or_sensor)

    data_source_info = scheduler_class.get_data_source_info()

    if belief_time is None:
        belief_time = server_now()

    scheduler_params = dict(
        start=start,
        end=end,
        resolution=resolution,
        belief_time=belief_time,
        flex_model=flex_model,
        flex_context=flex_context,
        return_multiple=True,
    )

    scheduler: Scheduler = get_scheduler_instance(
        scheduler_class=scheduler_class,
        asset_or_sensor=asset_or_sensor,
        scheduler_params=scheduler_params,
    )

    if flex_config_has_been_deserialized:
        scheduler.config_deserialized = True

    # we get the default scheduler info in case it fails in the compute step
    if rq_job:
        click.echo("Job %s made schedule." % rq_job.id)
        rq_job.meta["scheduler_info"] = scheduler.info

    consumption_schedule: SchedulerOutputType = scheduler.compute()

    # in case we are getting a custom Scheduler that hasn't implemented the multiple output return
    # this should only be called whenever the Scheduler applies to the Sensor.
    if isinstance(consumption_schedule, pd.Series):
        assert isinstance(asset_or_sensor, Sensor), ""
        consumption_schedule = [
            {
                "name": "consumption_schedule",
                "data": consumption_schedule,
                "sensor": asset_or_sensor,
            }
        ]

    if rq_job:
        click.echo("Job %s made schedule." % rq_job.id)
        rq_job.meta["scheduler_info"] = scheduler.info

    data_source = get_data_source(
        data_source_name=data_source_info["name"],
        data_source_model=data_source_info["model"],
        data_source_version=data_source_info["version"],
        data_source_type="scheduler",
    )

    # saving info on the job, so the API for a job can look the data up
    if rq_job:
        data_source_info["id"] = data_source.id
        rq_job.meta["data_source_info"] = data_source_info
        rq_job.save_meta()

    for result in consumption_schedule:
        sign = 1

        if result["sensor"].measures_power and result["sensor"].get_attribute(
            "consumption_is_positive", True
        ):
            sign = -1

        ts_value_schedule = [
            TimedBelief(
                event_start=dt,
                belief_time=belief_time,
                event_value=sign * value,
                sensor=result["sensor"],
                source=data_source,
            )
            for dt, value in result["data"].items()
        ]  # For consumption schedules, positive values denote consumption. For the db, consumption is negative
        bdf = tb.BeliefsDataFrame(ts_value_schedule)
        save_to_db(bdf)

    scheduler.persist_flex_model()
    db.session.commit()

    return True


def find_scheduler_class(asset_or_sensor: Asset | Sensor) -> type:
    """
    Find out which scheduler to use, given an asset or sensor.
    This will morph into a logic store utility, and schedulers should be registered for asset types there,
    instead of this fixed lookup logic.
    """

    # Choose which algorithm to use  TODO: unify loading this into a func store concept
    # first try to look if there's a "custom-scheduler" defined
    if "custom-scheduler" in asset_or_sensor.attributes:
        scheduler_specs = asset_or_sensor.attributes.get("custom-scheduler")
        scheduler_class = load_custom_scheduler(scheduler_specs)
        return scheduler_class

    if isinstance(asset_or_sensor, Sensor):
        asset = asset_or_sensor.generic_asset
    else:
        asset = asset_or_sensor

    if asset.generic_asset_type.name in (
        "battery",
        "one-way_evse",
        "two-way_evse",
    ):
        scheduler_class = StorageScheduler
    elif asset.generic_asset_type.name in ("process", "load"):
        scheduler_class = ProcessScheduler
    else:
        raise ValueError(
            "Scheduling is not (yet) supported for asset type %s."
            % asset.generic_asset_type
        )

    return scheduler_class


def handle_scheduling_exception(job, exc_type, exc_value, traceback):
    """
    Store exception as job meta data.
    """
    click.echo(
        "HANDLING RQ SCHEDULING WORKER EXCEPTION: %s:%s\n" % (exc_type, exc_value)
    )
    # from traceback import print_tb
    # print_tb(traceback)
    job.meta["exception"] = exc_value
    job.save_meta()


def get_data_source_for_job(job: Job) -> DataSource | None:
    """
    Try to find the data source linked by this scheduling job.

    We expect that enough info on the source was placed in the meta dict, either:
    - the DataSource ID itself (i.e. the normal situation), or
    - enough info to facilitate a DataSource query (as a fallback).
    """
    data_source_info = job.meta.get("data_source_info")
    if data_source_info and "id" in data_source_info:
        # this is the expected outcome
        return db.session.get(DataSource, data_source_info["id"])
    if data_source_info is None:
        raise ValueError(
            "Cannot look up scheduling data without knowing the full data_source_info (version)."
        )
    scheduler_sources = db.session.scalars(
        select(DataSource)
        .filter_by(
            type="scheduler",
            **data_source_info,
        )
        .order_by(DataSource.version.desc())
    ).all()  # Might still be more than one, e.g. per user
    if len(scheduler_sources) == 0:
        return None
    return scheduler_sources[0]

1	"""	4✔
2	Logic around scheduling (jobs)
3	"""
4
5	from __future__ import annotations	10✔
6
7	from datetime import datetime, timedelta	10✔
8	import os	10✔
9	import sys	10✔
10	import importlib.util	10✔
11	from importlib.abc import Loader	10✔
12	from typing import Type	10✔
13	import inspect	10✔
14
15
16	from flask import current_app	10✔
17	import click	10✔
18	from rq import get_current_job, Callback	10✔
19	from rq.exceptions import InvalidJobOperation	10✔
20	from rq.job import Job	10✔
21	import timely_beliefs as tb	10✔
22	import pandas as pd	10✔
23	from sqlalchemy import select	10✔
24
25	from flexmeasures.data import db	10✔
26	from flexmeasures.data.models.planning import Scheduler, SchedulerOutputType	10✔
27	from flexmeasures.data.models.planning.storage import StorageScheduler	10✔
28	from flexmeasures.data.models.planning.exceptions import InfeasibleProblemException	10✔
29	from flexmeasures.data.models.planning.process import ProcessScheduler	10✔
30	from flexmeasures.data.models.time_series import Sensor, TimedBelief	10✔
31	from flexmeasures.data.models.generic_assets import GenericAsset as Asset	10✔
32	from flexmeasures.data.models.data_sources import DataSource	10✔
33	from flexmeasures.data.utils import get_data_source, save_to_db	10✔
34	from flexmeasures.utils.time_utils import server_now	10✔
35	from flexmeasures.data.services.utils import (	10✔
36	job_cache,
37	get_asset_or_sensor_ref,
38	get_asset_or_sensor_from_ref,
39	get_scheduler_instance,
40	)
41
42
43	def load_custom_scheduler(scheduler_specs: dict) -> type:	10✔
44	"""
45	Read in custom scheduling spec.
46	Attempt to load the Scheduler class to use.
47
48	The scheduler class should be derived from flexmeasures.data.models.planning.Scheduler.
49	The scheduler class should have a class method named "compute".
50
51	Example specs:
52
53	{
54	"module": "/path/to/module.py", # or sthg importable, e.g. "package.module"
55	"class": "NameOfSchedulerClass",
56	}
57
58	"""
59	assert isinstance(	10✔
60	scheduler_specs, dict
61	), f"Scheduler specs is {type(scheduler_specs)}, should be a dict"
62	assert "module" in scheduler_specs, "scheduler specs have no 'module'."	10✔
63	assert "class" in scheduler_specs, "scheduler specs have no 'class'"	10✔
64
65	scheduler_name = scheduler_specs["class"]	10✔
66
67	# import module
68	module_descr = scheduler_specs["module"]	10✔
69	if os.path.exists(module_descr):	10✔
70	spec = importlib.util.spec_from_file_location(scheduler_name, module_descr)	10✔
71	assert spec, f"Could not load specs for scheduling module at {module_descr}."	10✔
72	module = importlib.util.module_from_spec(spec)	10✔
73	sys.modules[scheduler_name] = module	10✔
74	assert isinstance(spec.loader, Loader)	10✔
75	spec.loader.exec_module(module)	10✔
76	else: # assume importable module
77	try:	10✔
78	module = importlib.import_module(module_descr)	10✔
79	except TypeError as te:	×
80	current_app.logger.error(f"Cannot load {module_descr}: {te}.")	×
81	raise	×
82	except ModuleNotFoundError:	×
83	current_app.logger.error(	×
84	f"Attempted to import module {module_descr} (as it is not a valid file path), but it is not installed."
85	)
86	raise	×
87	assert module, f"Module {module_descr} could not be loaded."	10✔
88
89	# get scheduling function
90	assert hasattr(	10✔
91	module, scheduler_specs["class"]
92	), f"Module at {module_descr} has no class {scheduler_specs['class']}"
93
94	scheduler_class = getattr(module, scheduler_specs["class"])	10✔
95	schedule_function_name = "compute"	10✔
96	if not hasattr(scheduler_class, schedule_function_name):	10✔
97	raise NotImplementedError(	×
98	f"No function {schedule_function_name} in {scheduler_class}. Cannot load custom scheduler."
99	)
100	return scheduler_class	10✔
101
102
103	def trigger_optional_fallback(job, connection, type, value, traceback):	10✔
104	"""Create a fallback schedule job when the error is of type InfeasibleProblemException"""
105
106	job.meta["exception"] = value	10✔
107	job.save_meta()	10✔
108
109	if type is InfeasibleProblemException:	10✔
110	asset_or_sensor = get_asset_or_sensor_from_ref(job.meta.get("asset_or_sensor"))	10✔
111
112	scheduler_kwargs = job.meta["scheduler_kwargs"]	10✔
113
114	if ("scheduler_specs" in job.kwargs) and (	10✔
115	job.kwargs["scheduler_specs"] is not None
116	):
117	scheduler_class: Type[Scheduler] = load_custom_scheduler(	10✔
118	job.kwargs["scheduler_specs"]
119	)
120	else:
121	scheduler_class: Type[Scheduler] = find_scheduler_class(asset_or_sensor)	10✔
122
123	# only schedule a fallback schedule job if the original job has a fallback
124	# mechanism
125	if scheduler_class.fallback_scheduler_class is not None:	10✔
126	scheduler_class = scheduler_class.fallback_scheduler_class	10✔
127	scheduler_specs = {	10✔
128	"class": scheduler_class.__name__,
129	"module": inspect.getmodule(scheduler_class).__name__,
130	}
131
132	fallback_job = create_scheduling_job(	10✔
133	asset_or_sensor,
134	force_new_job_creation=True,
135	enqueue=False,
136	scheduler_specs=scheduler_specs,
137	**scheduler_kwargs,
138	)
139
140	job.meta["fallback_job_id"] = fallback_job.id	10✔
141	job.save_meta()	10✔
142	current_app.queues["scheduling"].enqueue_job(fallback_job)	10✔
143
144
145	@job_cache("scheduling")	10✔
146	def create_scheduling_job(	10✔
147	asset_or_sensor: Asset \| Sensor \| None = None,
148	sensor: Sensor \| None = None,
149	job_id: str \| None = None,
150	enqueue: bool = True,
151	requeue: bool = False,
152	force_new_job_creation: bool = False,
153	scheduler_specs: dict \| None = None,
154	**scheduler_kwargs,
155	) -> Job:
156	"""
157	Create a new Job, which is queued for later execution.
158
159	To support quick retrieval of the scheduling job, the job id is the unique entity address of the UDI event.
160	That means one event leads to one job (i.e. actions are event driven).
161
162	As a rule of thumb, keep arguments to the job simple, and deserializable.
163
164	The life cycle of a scheduling job:
165	1. A scheduling job is born here (in create_scheduling_job).
166	2. It is run in make_schedule which writes results to the db.
167	3. If an error occurs (and the worker is configured accordingly), handle_scheduling_exception comes in.
168
169	Arguments:
170	:param asset_or_sensor: asset or sensor for which the schedule is computed
171	:param job_id: optionally, set a job id explicitly
172	:param enqueue: if True, enqueues the job in case it is new
173	:param requeue: if True, requeues the job in case it is not new and had previously failed
174	(this argument is used by the @job_cache decorator)
175	:param force_new_job_creation: if True, this attribute forces a new job to be created (skipping cache)
176	(this argument is used by the @job_cache decorator)
177	:returns: the job
178
179	"""
180	# We first create a scheduler and check if deserializing works, so the flex config is checked
181	# and errors are raised before the job is enqueued (so users get a meaningful response right away).
182	# Note: We are putting still serialized scheduler_kwargs into the job!
183
184	if sensor is not None:	10✔
185	current_app.logger.warning(	×
186	"The `sensor` keyword argument is deprecated. Please, consider using the argument `asset_or_sensor`."
187	)
188	asset_or_sensor = sensor	×
189
190	if scheduler_specs:	10✔
191	scheduler_class: Type[Scheduler] = load_custom_scheduler(scheduler_specs)	10✔
192	else:
193	scheduler_class: Type[Scheduler] = find_scheduler_class(asset_or_sensor)	10✔
194
195	scheduler = get_scheduler_instance(	10✔
196	scheduler_class=scheduler_class,
197	asset_or_sensor=asset_or_sensor,
198	scheduler_params=scheduler_kwargs,
199	)
200	scheduler.deserialize_config()	10✔
201
202	asset_or_sensor = get_asset_or_sensor_ref(asset_or_sensor)	10✔
203	job = Job.create(	10✔
204	make_schedule,
205	kwargs=dict(
206	asset_or_sensor=asset_or_sensor,
207	scheduler_specs=scheduler_specs,
208	**scheduler_kwargs,
209	),
210	id=job_id,
211	connection=current_app.queues["scheduling"].connection,
212	ttl=int(
213	current_app.config.get(
214	"FLEXMEASURES_JOB_TTL", timedelta(-1)
215	).total_seconds()
216	),
217	result_ttl=int(
218	current_app.config.get(
219	"FLEXMEASURES_PLANNING_TTL", timedelta(-1)
220	).total_seconds()
221	), # NB job.cleanup docs says a negative number of seconds means persisting forever
222	on_failure=Callback(trigger_optional_fallback),
223	)
224
225	job.meta["asset_or_sensor"] = asset_or_sensor	10✔
226	job.meta["scheduler_kwargs"] = scheduler_kwargs	10✔
227	job.save_meta()	10✔
228
229	# in case the function enqueues it
230	try:	10✔
231	job_status = job.get_status(refresh=True)	10✔
NEW 232	except InvalidJobOperation:	2✔
NEW 233	job_status = None	2✔
234
235	# with job_status=None, we ensure that only fresh new jobs are enqueued (in the contrary they should be requeued)
236	if enqueue and not job_status:	10✔
237	current_app.queues["scheduling"].enqueue_job(job)	10✔
238	current_app.job_cache.add(	10✔
239	asset_or_sensor["id"],
240	job.id,
241	queue="scheduling",
242	asset_or_sensor_type=asset_or_sensor["class"].lower(),
243	)
244
245	return job	10✔
246
247
248	def make_schedule(	10✔
249	sensor_id: int \| None = None,
250	start: datetime \| None = None,
251	end: datetime \| None = None,
252	resolution: timedelta \| None = None,
253	asset_or_sensor: dict \| None = None,
254	belief_time: datetime \| None = None,
255	flex_model: dict \| None = None,
256	flex_context: dict \| None = None,
257	flex_config_has_been_deserialized: bool = False,
258	scheduler_specs: dict \| None = None,
259	) -> bool:
260	"""
261	This function computes a schedule. It returns True if it ran successfully.
262
263	It can be queued as a job (see create_scheduling_job).
264	In that case, it will probably run on a different FlexMeasures node than where the job is created.
265	In any case, this function expects flex_model and flex_context to not have been deserialized yet.
266
267	This is what this function does:
268	- Find out which scheduler should be used & compute the schedule
269	- Turn scheduled values into beliefs and save them to db
270	"""
271	# https://docs.sqlalchemy.org/en/13/faq/connections.html#how-do-i-use-engines-connections-sessions-with-python-multiprocessing-or-os-fork
272	db.engine.dispose()	10✔
273
274	if sensor_id is not None:	10✔
275	current_app.logger.warning(	×
276	"The `sensor_id` keyword argument is deprecated. Please, consider using the argument `asset_or_sensor`."
277	)
278	asset_or_sensor = {"class": "Sensor", "id": sensor_id}	×
279
280	asset_or_sensor: Asset \| Sensor = get_asset_or_sensor_from_ref(asset_or_sensor)	10✔
281
282	rq_job = get_current_job()	10✔
283	if rq_job:	10✔
284	click.echo(	10✔
285	"Running Scheduling Job %s: %s, from %s to %s"
286	% (rq_job.id, asset_or_sensor, start, end)
287	)
288
289	if scheduler_specs:	10!
290	scheduler_class: Type[Scheduler] = load_custom_scheduler(scheduler_specs)	10✔
291	else:
292	scheduler_class: Type[Scheduler] = find_scheduler_class(asset_or_sensor)	10✔
293
294	data_source_info = scheduler_class.get_data_source_info()	10✔
295
296	if belief_time is None:	10✔
297	belief_time = server_now()	10✔
298
299	scheduler_params = dict(	10✔
300	start=start,
301	end=end,
302	resolution=resolution,
303	belief_time=belief_time,
304	flex_model=flex_model,
305	flex_context=flex_context,
306	return_multiple=True,
307	)
308
309	scheduler: Scheduler = get_scheduler_instance(	10✔
310	scheduler_class=scheduler_class,
311	asset_or_sensor=asset_or_sensor,
312	scheduler_params=scheduler_params,
313	)
314
315	if flex_config_has_been_deserialized:	10✔
316	scheduler.config_deserialized = True	×
317
318	# we get the default scheduler info in case it fails in the compute step
319	if rq_job:	10✔
320	click.echo("Job %s made schedule." % rq_job.id)	10✔
321	rq_job.meta["scheduler_info"] = scheduler.info	10✔
322
323	consumption_schedule: SchedulerOutputType = scheduler.compute()	10!
324
325	# in case we are getting a custom Scheduler that hasn't implemented the multiple output return
326	# this should only be called whenever the Scheduler applies to the Sensor.
327	if isinstance(consumption_schedule, pd.Series):	10✔
328	assert isinstance(asset_or_sensor, Sensor), ""	10✔
329	consumption_schedule = [	10✔
330	{
331	"name": "consumption_schedule",
332	"data": consumption_schedule,
333	"sensor": asset_or_sensor,
334	}
335	]
336
337	if rq_job:	10✔
338	click.echo("Job %s made schedule." % rq_job.id)	10✔
339	rq_job.meta["scheduler_info"] = scheduler.info	10✔
340
341	data_source = get_data_source(	10!
342	data_source_name=data_source_info["name"],
343	data_source_model=data_source_info["model"],
344	data_source_version=data_source_info["version"],
345	data_source_type="scheduler",
346	)
347
348	# saving info on the job, so the API for a job can look the data up
349	if rq_job:	10✔
350	data_source_info["id"] = data_source.id	10✔
351	rq_job.meta["data_source_info"] = data_source_info	10✔
352	rq_job.save_meta()	10✔
353
354	for result in consumption_schedule:	10!
355	sign = 1	10✔
356
357	if result["sensor"].measures_power and result["sensor"].get_attribute(	10✔
358	"consumption_is_positive", True
359	):
360	sign = -1	10✔
361
362	ts_value_schedule = [	10!
363	TimedBelief(
364	event_start=dt,
365	belief_time=belief_time,
366	event_value=sign * value,
367	sensor=result["sensor"],
368	source=data_source,
369	)
370	for dt, value in result["data"].items()
371	] # For consumption schedules, positive values denote consumption. For the db, consumption is negative
372	bdf = tb.BeliefsDataFrame(ts_value_schedule)	10✔
373	save_to_db(bdf)	10✔
374
375	scheduler.persist_flex_model()	10✔
376	db.session.commit()	10✔
377
378	return True	10✔
379
380
381	def find_scheduler_class(asset_or_sensor: Asset \| Sensor) -> type:	10✔
382	"""
383	Find out which scheduler to use, given an asset or sensor.
384	This will morph into a logic store utility, and schedulers should be registered for asset types there,
385	instead of this fixed lookup logic.
386	"""
387
388	# Choose which algorithm to use TODO: unify loading this into a func store concept
389	# first try to look if there's a "custom-scheduler" defined
390	if "custom-scheduler" in asset_or_sensor.attributes:	10✔
391	scheduler_specs = asset_or_sensor.attributes.get("custom-scheduler")	10✔
392	scheduler_class = load_custom_scheduler(scheduler_specs)	10✔
393	return scheduler_class	10✔
394
395	if isinstance(asset_or_sensor, Sensor):	10✔
396	asset = asset_or_sensor.generic_asset	10✔
397	else:
398	asset = asset_or_sensor	×
399
400	if asset.generic_asset_type.name in (	10✔
401	"battery",
402	"one-way_evse",
403	"two-way_evse",
404	):
405	scheduler_class = StorageScheduler	10✔
406	elif asset.generic_asset_type.name in ("process", "load"):	10✔
407	scheduler_class = ProcessScheduler	×
408	else:
409	raise ValueError(	10✔
410	"Scheduling is not (yet) supported for asset type %s."
411	% asset.generic_asset_type
412	)
413
414	return scheduler_class	10✔
415
416
417	def handle_scheduling_exception(job, exc_type, exc_value, traceback):	10✔
418	"""
419	Store exception as job meta data.
420	"""
421	click.echo(	10✔
422	"HANDLING RQ SCHEDULING WORKER EXCEPTION: %s:%s\n" % (exc_type, exc_value)
423	)
424	# from traceback import print_tb
425	# print_tb(traceback)
426	job.meta["exception"] = exc_value	10✔
427	job.save_meta()	10✔
428
429
430	def get_data_source_for_job(job: Job) -> DataSource \| None:	10✔
431	"""
432	Try to find the data source linked by this scheduling job.
433
434	We expect that enough info on the source was placed in the meta dict, either:
435	- the DataSource ID itself (i.e. the normal situation), or
436	- enough info to facilitate a DataSource query (as a fallback).
437	"""
438	data_source_info = job.meta.get("data_source_info")	10✔
439	if data_source_info and "id" in data_source_info:	10✔
440	# this is the expected outcome
441	return db.session.get(DataSource, data_source_info["id"])	10✔
442	if data_source_info is None:	×
443	raise ValueError(	×
444	"Cannot look up scheduling data without knowing the full data_source_info (version)."
445	)
446	scheduler_sources = db.session.scalars(	×
447	select(DataSource)
448	.filter_by(
449	type="scheduler",
450	**data_source_info,
451	)
452	.order_by(DataSource.version.desc())
453	).all() # Might still be more than one, e.g. per user
454	if len(scheduler_sources) == 0:	×
455	return None	×
456	return scheduler_sources[0]	×

FlexMeasures / flexmeasures / 11741252103

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous