"""Seedable EUT-monitor example for simulations and custom implementations."""
from __future__ import annotations
import math
import random
import time
from .events import EUT_EVENT_STATUSES, make_eut_event
from .monitors import EUTMonitor
DEFAULT_STATUS_WEIGHTS = {
"passed": 0.85,
"degraded": 0.10,
"failed": 0.04,
"not_evaluated": 0.01,
}
DEFAULT_RECOVERY_WEIGHTS = {
"automatic": 0.70,
"operator": 0.15,
"reset": 0.10,
"failed": 0.05,
}
def _validate_weights(weights, allowed, name):
result = {key: 0.0 for key in allowed}
unknown = set(weights) - set(allowed)
if unknown:
raise ValueError("unknown %s values: %s" % (name, ", ".join(sorted(unknown))))
for key, value in weights.items():
try:
value = float(value)
except (TypeError, ValueError) as exc:
raise TypeError("%s weights must be real numbers" % name) from exc
if not math.isfinite(value) or value < 0.0:
raise ValueError("%s weights must be finite and non-negative" % name)
result[key] = value
if not any(result.values()):
raise ValueError("at least one %s weight must be positive" % name)
return result
[docs]
class RandomEUTMonitor(EUTMonitor):
"""Generate reproducible random EUT observations without hardware.
This class is both a simulation helper and a compact template for custom
monitors. Its :meth:`poll_event` method never blocks. Replace
:meth:`_make_exposure_event` with a camera, communication, or process-data
check when adapting the class to real EUT monitoring.
Parameters
----------
seed:
Seed used by an instance-local random-number generator. Equal seeds and
configurations produce equal event sequences.
status_weights:
Relative weights for ``passed``, ``degraded``, ``failed``, and
``not_evaluated`` observations during exposure.
recovery_weights:
Relative weights for ``automatic``, ``operator``, ``reset``, and
``failed`` recovery after a degraded or failed observation.
delay:
Non-negative delay in seconds before an event becomes available.
source:
Source label stored in every generated event.
operating_mode_changed, stored_data_lost:
Simulated observations needed for performance criterion B. Both
default to ``False``.
Notes
-----
This simulation must not be used as evidence of real EUT performance. The
mandatory manual monitor remains active when this monitor is passed to
``Measure_Immunity`` as an automatic ``eut_monitor``.
"""
[docs]
def __init__(self, *, seed=None, status_weights=None, recovery_weights=None,
delay=0.0, source="random_simulation",
operating_mode_changed=False, stored_data_lost=False):
self.random = random.Random(seed)
self.status_weights = _validate_weights(
DEFAULT_STATUS_WEIGHTS if status_weights is None else status_weights,
EUT_EVENT_STATUSES,
"status",
)
self.recovery_weights = _validate_weights(
DEFAULT_RECOVERY_WEIGHTS if recovery_weights is None else recovery_weights,
("automatic", "operator", "reset", "failed"),
"recovery",
)
try:
self.delay = float(delay)
except (TypeError, ValueError) as exc:
raise TypeError("delay must be a real number") from exc
if not math.isfinite(self.delay) or self.delay < 0.0:
raise ValueError("delay must be finite and non-negative")
self.source = str(source)
if not isinstance(operating_mode_changed, bool):
raise TypeError("operating_mode_changed must be bool")
if not isinstance(stored_data_lost, bool):
raise TypeError("stored_data_lost must be bool")
self.operating_mode_changed = operating_mode_changed
self.stored_data_lost = stored_data_lost
self.contexts = []
self._phase = "during_exposure"
self._event = None
self._available_at = None
self._exposure_status = None
def _weighted_choice(self, weights):
values = tuple(weights)
return self.random.choices(values, weights=[weights[value] for value in values])[0]
def _make_exposure_event(self, context):
"""Return one simulated observation for an exposure."""
status = self._weighted_choice(self.status_weights)
self._exposure_status = status
return make_eut_event(
status,
"random_simulation",
{"context": dict(context)},
source=self.source,
operating_mode_changed=self.operating_mode_changed,
stored_data_lost=self.stored_data_lost,
)
def _make_post_exposure_event(self, context):
"""Return a post-exposure observation consistent with the exposure."""
if self._exposure_status == "passed":
state, recovery = "normal", "not_required"
elif self._exposure_status == "not_evaluated":
state, recovery = "not_evaluated", "not_evaluated"
else:
recovery = self._weighted_choice(self.recovery_weights)
state = "failed" if recovery == "failed" else "normal"
status = {
"normal": "passed",
"failed": "failed",
"not_evaluated": "not_evaluated",
}[state]
return make_eut_event(
status,
"random_post_exposure_simulation",
{"context": dict(context)},
phase="post_exposure",
functional_state=state,
after_exposure_state=state,
recovery=recovery,
source=self.source,
operating_mode_changed=self.operating_mode_changed,
stored_data_lost=self.stored_data_lost,
)
def _set_pending_event(self, event):
self._event = event
self._available_at = time.monotonic() + self.delay
[docs]
def start_exposure(self, context):
"""Generate and schedule a random event for a new exposure.
Parameters
----------
context : mapping
Measurement context describing the exposure.
"""
context = dict(context)
self.contexts.append(context)
self._phase = "during_exposure"
self._set_pending_event(self._make_exposure_event(context))
[docs]
def start_phase(self, phase, context):
"""Generate a post-exposure event when that phase begins.
Parameters
----------
phase : str
Phase from :data:`~mpylab.env.eut.EUT_EVENT_PHASES`.
context : mapping
Measurement context for the phase.
"""
super().start_phase(phase, context)
self._phase = phase
if phase == "post_exposure":
self._set_pending_event(self._make_post_exposure_event(context))
[docs]
def poll_event(self):
"""Return the scheduled random event after its configured delay.
Returns
-------
dict or None
Scheduled event, or ``None`` while no event is due.
"""
if self._event is None or time.monotonic() < self._available_at:
return None
event, self._event = self._event, None
return event
[docs]
def stop_exposure(self):
"""Discard any random event still pending for the exposure."""
self._event = None
self._available_at = None