-
|
Hello, i am using the following approach to create a generic class ConnectionState(Enum):
UNINITIALIZED = -1
INIT = 0
CONNECTED = 1
UNREACHABLE = 2
INACCESSIBLE = 3
UNKNOWN_ERROR = 255
class Device:
""" Holds information on a device and its connection state
"""
def __init__(self, name: str, machine: Optional[Machine] = None):
self.name = name
if machine is not None:
self.machine = machine
else:
# Initialize the state machine with default states
# later bind of machine by attribute assignment will not trigger magic methods
self.machine = Machine(model=self,
states=ConnectionState,
initial=ConnectionState.UNINITIALIZED,
after_state_change=self.report_state_change,
send_event=True)I am creating multiple I read some of the issues which contained "Skip binding" but i am not sure if i understood the root cause for this message. Why do i get these messages in my use case? I am just creating new instance of the model and each of the model instances has its own machine. So why does the model already contain these attributes? |
Beta Was this translation helpful? Give feedback.
Replies: 2 comments
-
|
Hello @mahadi, I cannot reproduce your observations with the info provided. Could you assemble an MRE? The warnings you see are usually emitted if you attempt to bind the same model twice by passing it to two different machines with similar or identical state configurations or by calling class Model:
...
m = Model()
m1 = Machine(model=m, states=states)
m2 = Machine(model=m, states=states) # <-- emits warnings
m1.add_model(m) # <-- emits warningAlso while adding and removing models their decoration is not removed. See API doc states:
This means if you re-add a previously removed model again it will emit warnings as well. |
Beta Was this translation helpful? Give feedback.
-
|
I found the issue. Please see the MRE below (it contains a pit of boiler plate code the get the transitions logger - just some copy and paste code). The issue is that there is not only the import logging
from enum import Enum
from typing import Optional
from loguru import logger
from transitions import Machine
class InterceptHandler(logging.Handler):
""" Redirects standard logging to loguru's logger """
def __init__(self, logger, *args, **kwargs):
super().__init__(*args, **kwargs)
self.logger = logger
def emit(self, record):
# Get corresponding Loguru level if it exists
try:
level = logger.level(record.levelname).name
except ValueError:
level = record.levelno
# Find caller from where originated the logged message
frame, depth = logging.currentframe(), 2
while frame.f_code.co_filename == logging.__file__:
frame = frame.f_back
depth += 1
self.logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())
# intercept ALL logs
logging.basicConfig(handlers=[InterceptHandler(logger)], level=logging.INFO)
class ServiceState(Enum):
UNINITIALIZED = -1
INIT = 0
CONNECTED = 1
UNREACHABLE = 2
INACCESSIBLE = 3
UNKNOWN_ERROR = 255
class ConnectionState(Enum):
UNINITIALIZED = -1
INIT = 0
CONNECTED = 1
UNREACHABLE = 2
INACCESSIBLE = 3
UNKNOWN_ERROR = 255
class Device:
""" Holds information on a device and its connection state
"""
def __init__(self, name: str, machine: Optional[Machine] = None):
self.name = name
if machine is not None:
self.machine = machine
else:
# Initialize the state machine with default states
# later bind of machine by attribute assignment will not trigger magic methods
self.machine = Machine(model=self,
states=ConnectionState,
initial=ConnectionState.UNINITIALIZED,
after_state_change=self.report_state_change,
send_event=True)
def report_state_change(self, event_data):
""" Report the current state """
if event_data.transition.source != event_data.transition.dest:
logger.warning(f"{self.name}.connection_state has changed from {event_data.transition.source} to: "
f"[{self.state.value:02d}] {self.state.name}")
class Service(Device):
def __init__(self, name: str):
super().__init__(name)
# Initialize the state machine with service states instead of device states
self.machine = Machine(model=self,
states=ServiceState,
initial=ServiceState.UNINITIALIZED,
after_state_change=self.report_state_change,
send_event=True)
if __name__ == '__main__':
logger.info("Init devices")
d1 = Device("d1")
d2 = Device("d2")
d3 = Device("d3")
logger.info("Init services")
s1 = Service("s1") # <-- warnings are raised hereBut the warning revealed that the code was actually not correct and indeed the model was bound twice 🎉. So for the original purpose to work the machine should be actually passed into the parent constructor: class Service(Device):
def __init__(self, name: str):
# Initialize the state machine with service states instead of device states
machine = Machine(model=self,
states=ServiceState,
initial=ServiceState.UNINITIALIZED,
after_state_change=self.report_state_change,
send_event=True)
super().__init__(name, machine) |
Beta Was this translation helpful? Give feedback.
Hello @mahadi,
I cannot reproduce your observations with the info provided. Could you assemble an MRE? The warnings you see are usually emitted if you attempt to bind the same model twice by passing it to two different machines with similar or identical state configurations or by calling
Machine.add_modelwith an already bound model.Also while adding and removing models their decoration is not removed. See API doc states: