smart-energy-monitor/box-pc/plc-connector/inputs/rockwell/allen_bradley_connect.py

import logging

from pylogix import PLC
from threading import Thread
import time
from datetime import datetime

from structures.plant import *
from inputs.common import Input

localtz = datetime.now().astimezone().tzinfo
logger = logging.getLogger(__name__)

class AllenBradleyCPU(Input):
 
  def __init__(self, host):
    super().__init__(self.read_handler)
    self.comm = PLC()
    self.comm.IPAddress = host
    self.interval = 0.02
    self.tags = [
      "B14[31]", # ejector_move_down
      "B14[32]", # ejector_move_up
      "B14[34]", # carriage_move_out
      "B14[35]", # carriage_move_in
      "B14[37]", # side_clamps_open
      "B14[38]", # side_clamps_close
      "B14[42]", # table_move_down
      "B14[43]", # table_move_up
      "B14[45]", # gassing_platemove_out
      "B14[46]", # gassing_plate_move_in
      "B14[48]", # cope_eject_plate_move_out
      "B14[49]", # cope_eject_plate_move_in
      "B14[51]", # top_part_move_up
      "B14[52]", # top_part_move_down
      "B14[54]", # front_door_open
      "B14[55]", # front_door_close
      "B14[57]", # pneumatic_loose_part_1_move_out
      "B14[58]", # pneumatic_loose_part_1_move_in
      "B14[60]", # hydraulic_loose_part_2_move_out
      "B14[61]", # hydraulic_loose_part_2_move_in
      "B14[63]", # hydraulic_loose_part_3_move_out
      "B14[64]", # hydraulic_loose_part_3_move_in
      "B14[69]", # clamping_device_side_clamp_left_clamp
      "B14[70]", # clamping_device_side_clamp_left_loose
      "B14[72]", # clamping_device_side_clamp_right_clamp
      "B14[73]", # clamping_device_side_clamp_right_loose
      "B14[81]", # clamping_device_shoot_plate_clamp
      "B14[82]", # clamping_device_shoot_plate_loose
      "B14[83]", # sand_refill
      "B14[84]", # sand_gate_close
      "B14[85]", # sand_gate_open
      "B14[93]", # shoot
      "B14[103]", # clamping_device_gassing_plate_clamp
      "B16[12]", # central_amine_supply_refill
      "B16[13]", # gassing
      "B16[15]", # gas_generator_process_coldbox_betaset
      "B18[31]", # mixer_lid_move_up
      "B18[32]", # mixer_lid_move_down
      "B18[35]", # mixer_wing_motor_on
      "B18[37]", # mixer_move_up
      "B18[38]", # mixer_move_down
      "B18[40]", # sand_dosing_unit_inlet
      "B18[42]", # sand_dosing_unit_outlet
      "B18[44]", # binder_1_sucking
      "B18[45]", # binder_1_blowing
      "B18[47]", # binder_2_suction
      "B18[48]", # binder_2_blowing
      "B18[50]", # binder_3_sucking
      "B18[51]", # binder_3_blowing
      "B18[53]", # binder_4_sucking
      "B18[54]", # binder_4_blowing
      "B18[59]", # additive_1_dosing
      "B18[60]", # additive_2_dosing
      "B18[62]", # mixer_bowl_direction_eject_1_machine
      "B18[63]", # mixer_bowl_direction_eject_2_scrap
      "B18[65]", # cleaning_cylinder_move_up
      "B18[66]", # cleaning_cylinder_move_down
      "B18[68]", # mixer_sand_slide_move_to_machine
      "B18[69]", # mixer_sand_slide_move_to_scrap
      "B18[71]", # vertical_mixersand_slide_gateclose
      "B18[72]", # vertical_mixer_sand_slide_gate_open
      "B18[73]", # sand_sender
    ]
    
  def read_handler(self):
    timestamp = datetime.now(localtz)
    ret = self.comm.Read(self.tags)
    if ret[0].Status == "Success":
      values = [r.Value for r in ret]
      self._q.put(PlantState(timestamp, "AB", *values))
    else:
      logger.error("CPU read: " + ret[0].Status)