Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def StateOne(self, axis):
return (State.On,"Device in On state")
def calculate_state_info(self, status_info=None):
if status_info is None:
status_info = self._state, self._status
state, status = status_info
if state == State.On:
state_str = "Stopped"
else:
state_str = "in " + State[state]
new_status = self._STD_STATUS.format(name=self.name, state=state_str,
ctrl_status=status)
return status_info[0], new_status
def StateOne(self, axis):
"""Get the specified counter state"""
springfield = self.springfield
state = springfield.getState(axis)
if state == 1:
return State.On, "Counter is stopped"
elif state == 2:
return State.Moving, "Counter is acquiring"
elif state == 3:
return State.Fault, "Counter has an error"
def calculate_state_info(self, state_info=None):
if state_info is None:
state = self._state
status = self._status
ls = self._limit_switches.value
else:
state, status, ls = state_info
if state == State.On:
state_str = "Stopped"
elif state == State.Moving:
state_str = "Moving"
motion = self.get_operation()
if motion is None:
state_str += " (external)"
else:
motion_state = motion._motion_info[self].motion_state
if motion_state == MotionState.MovingBacklash:
state_str += " (backlash)"
elif motion_state == MotionState.MovingInstability:
state_str += " (instability)"
else:
state_str = "in " + State[state]
limit_switches = ""
def StateOne(self, axis):
"""Get the specified motor state"""
springfield = self.springfield
state = springfield.getState(axis)
if state == 1:
return State.On, "Motor is stopped"
elif state == 2:
return State.Moving, "Motor is moving"
elif state == 3:
return State.Fault, "Motor has an error"
u_state, u_status = self._calculate_element_state(
elem, elem_state_info)
if u_state == State.Moving:
moving.add(elem)
elif u_state == State.On:
on.add(elem)
elif u_state == State.Fault:
fault.add(elem)
elif u_state == State.Alarm:
alarm.add(elem)
elif u_state is State.Unknown:
unknown.add(elem)
elif u_state is None:
none.add(elem)
status.append(u_status)
state = State.On
if none or unknown:
state = State.Unknown
if fault:
state = State.Fault
elif alarm:
state = State.Alarm
elif moving:
state = State.Moving
self._state_statistics = {State.On: on, State.Fault: fault,
State.Alarm: alarm, State.Moving: moving,
State.Unknown: unknown, None: none}
status = "\n".join(status)
return state, status
if state_info is None:
state_info = {}
for elem in user_elements:
if elem.get_type() == ElementType.External:
continue
# cannot call get_state(us) here since it may lead to dead lock!
si = elem.inspect_state(), elem.inspect_status()
state_info[elem] = si
for elem, elem_state_info in state_info.items():
elem_type = elem.get_type()
if elem_type == ElementType.External:
continue
u_state, u_status = self._calculate_element_state(elem, elem_state_info)
if u_state == State.Moving:
moving.add(elem)
elif u_state == State.On:
on.add(elem)
elif u_state == State.Fault:
fault.add(elem)
elif u_state == State.Alarm:
alarm.add(elem)
elif u_state is State.Unknown:
unknown.add(elem)
elif u_state is None:
none.add(elem)
status.append(u_status)
state = State.On
if none or unknown:
state = State.Unknown
if fault:
state = State.Fault
elif alarm:
inst, props, *args, **kwargs)
# initialize hardware communication
self.springfield = springfieldlib.SpringfieldTriggerHW()
# do some initialization
self._triggers = {}
def AddDevice(self, axis):
self._triggers[axis] = True
def DeleteDevice(self, axis):
del self._triggers[axis]
StateMap = {
1: State.On,
2: State.Moving,
3: State.Fault,
}
def StateOne(self, axis):
springfield = self.springfield
state = self.StateMap[springfield.getState(axis)]
status = springfield.getStatus(axis)
return state, status
def SynchOne(self, axis, synchronization):
self.springfield.SynchChannel(axis, synchronization)
def StartOne(self, axis, position):
self.springfield.StartChennel(axis, position)
def run(self):
self._running = True
try:
while len(self.active_events) > 0 and not self.is_stopped():
self.wait_active()
self.fire_active()
self.wait_passive()
self.fire_passive()
self._id += 1
finally:
self._started = False
self._running = False
self._stopped = False
self.fire_event(EventType("state"), State.On)
if self.iPAP.connected:
try:
register = self.attributes[axis]['status_value']
status_dict = self.iPAP.decodeStatus(register)
previous_state = self.attributes[axis]["last_state_value"]
if status_dict is None:
self.attributes[axis]["last_state_value"] = State.Alarm
return (State.Alarm, 'Status Register not available', 0)
# CHECK POWER LED
poweron = status_dict['poweron'][0]
disable, status_power = status_dict['disable']
if poweron == 1:
state = State.On
status_state = 'ON'
else:
state = State.Alarm
status_state = 'ALARM_AXIS_DISABLE'
# CHECK LIMIT SWITCHES
lower, status_limneg = status_dict['lim-']
upper, status_limpos = status_dict['lim+']
switchstate = 0
if lower == 1 and upper == 1:
switchstate = 6
elif lower == 1:
switchstate = 4
elif upper == 1:
switchstate = 2
if switchstate != 0 and state == State.On: