Skip to content
Snippets Groups Projects
Commit 609dbfd0 authored by Taya Snijder's avatar Taya Snijder
Browse files

removed SNMP and ini remnant

parent 1e24f45d
No related branches found
No related tags found
1 merge request!27Resolve #2021 "04 16 branched from master apsctl device"
Hide whitespace changes
Inline Side-by-side
devices/clients/SNMP_client.py deleted 100644 → 0
+ 0
− 156
View file @ 1e24f45d
from src.comms_client import *
import snmp
import numpy
import traceback
__all__ = ["SNMP_client"]
snmp_to_numpy_dict = {
snmp.types.INTEGER: numpy.int64,
snmp.types.TimeTicks: numpy.int64,
snmp.types.OCTET_STRING: str,
snmp.types.OID: str
}
snmp_types = {
"Integer": numpy.int64,
"Gauge": numpy.int64,
"TimeTick": numpy.int64,
"Counter32": numpy.int64,
"OctetString": str,
"IpAddress": str,
"OID": str,
}
# numpy_to_snmp_dict = {
# numpy.int64,
# numpy.int64,
# str,
# }
class SNMP_client(CommClient):
"""
messages to keep a check on the connection. On connection failure, reconnects once.
"""
def start(self):
super().start()
def __init__(self, community, host, timeout, fault_func, streams, try_interval=2):
"""
Create the SNMP and connect() to it
"""
super().__init__(fault_func, streams, try_interval)
self.community = community
self.host = host
self.manager = snmp.Manager(community=bytes(community, "utf8"))
# Explicitly connect
if not self.connect():
# hardware or infra is down -- needs fixing first
fault_func()
return
def connect(self):
"""
Try to connect to the client
"""
self.streams.debug_stream("Connecting to community: %s, host: %s", self.community, self.host)
self.connected = True
return True
def ping(self):
"""
ping the client to make sure the connection with the client is still functional.
"""
pass
def _setup_annotation(self, annotation):
"""
This class's Implementation of the get_mapping function. returns the read and write functions
"""
if isinstance(annotation, dict):
# check if required path inarg is present
if annotation.get('oids') is None:
AssertionError("SNMP get attributes require an oid")
oids = annotation.get("oids") # required
else:
TypeError("SNMP attributes require a dict with oid(s)")
return
return oids
def setup_value_conversion(self, attribute):
"""
gives the client access to the attribute_wrapper object in order to access all data it could potentially need.
"""
dim_x = attribute.dim_x
dim_y = attribute.dim_y
return dim_x, dim_y
def get_oids(self, x, y, in_oid):
if x == 0:
x = 1
if y == 0:
y = 1
nof_oids = x * y
if nof_oids == 1:
# is scalar
if type(in_oid) is str:
# for ease of handling put single oid in a 1 element list
in_oid = [in_oid]
return in_oid
elif type(in_oid) is list and len(in_oid) == nof_oids:
# already is an array and of the right length
return in_oid
elif type(in_oid) is list and len(in_oid) != nof_oids:
# already is an array but the wrong length. Unable to handle this
raise ValueError("SNMP oids need to either be a single value or an array the size of the attribute dimensions. got: {} expected: {}x{}={}".format(len(in_oid),x,y,x*y))
else:
out_oids = []
for i in range(nof_oids):
out_oids.append(in_oid + ".{}".format(i+1))
return out_oids
def setup_attribute(self, annotation, attribute):
"""
MANDATORY function: is used by the attribute wrapper to get read/write functions. must return the read and write functions
"""
# process the annotation
oids = self._setup_annotation(annotation)
# get all the necessary data to set up the read/write functions from the attribute_wrapper
dim_x, dim_y = self.setup_value_conversion(attribute)
oids = self.get_oids(dim_x, dim_y, oids)
def _read_function():
vars = self.manager.get(self.host, *oids)
value = []
for i in vars:
val = snmp_to_numpy_dict[type(i.value)](str(i.value))
value.append(val)
return value
def _write_function(value):
for i in range(len(oids)):
self.manager.set(self.host, oids[i], value[i])
# return the read/write functions
return _read_function, _write_function
devices/clients/ini_client.py deleted 100644 → 0
+ 0
− 133
View file @ 1e24f45d
from src.comms_client import *
import configparser
import numpy
numpy_to_ini_dict = {
numpy.int64: int,
numpy.double: float,
numpy.bool_: bool,
str: str
}
ini_to_numpy_dict = {
int: numpy.int64,
float: numpy.double,
bool: numpy.bool_,
str: str
}
import os
class ini_client(CommClient):
"""
this class provides an example implementation of a comms_client.
Durirng initialisation it creates a correctly shaped zero filled value. on read that value is returned and on write its modified.
"""
def start(self):
super().start()
def __init__(self, filename, fault_func, streams, try_interval=2):
"""
initialises the class and tries to connect to the client.
"""
self.config = configparser.ConfigParser()
self.filename = filename
if not filename.endswith(".ini"):
filename = filename + ".ini"
super().__init__(fault_func, streams, try_interval)
# Explicitly connect
if not self.connect():
# hardware or infra is down -- needs fixing first
fault_func()
return
def connect(self):
files_path = [os.path.abspath(x) for x in os.listdir()]
self.streams.debug_stream(" %s", files_path)
self.config_file = open(self.filename, "rw")
self.connected = True # set connected to true
return True # if succesfull, return true. otherwise return false
def disconnect(self):
self.connected = False # always force a reconnect, regardless of a successful disconnect
self.streams.debug_stream("disconnected from the 'client' ")
def _setup_annotation(self, annotation):
"""
this function gives the client access to the comm client annotation data given to the attribute wrapper.
The annotation data can be used to provide whatever extra data is necessary in order to find/access the monitor/control point.
the annotation can be in whatever format may be required. it is up to the user to handle its content
example annotation may include:
- a file path and file line/location
- COM object path
"""
# as this is an example, just print the annotation
self.streams.debug_stream("annotation: {}".format(annotation))
name = annotation.get('name')
if name is None:
AssertionError("ini client requires a variable name to set/get")
section = annotation.get('section')
if section is None:
AssertionError("requires a section to open")
return section, name
def _setup_value_conversion(self, attribute):
"""
gives the client access to the attribute_wrapper object in order to access all
necessary data such as dimensionality and data type
"""
if attribute.dim_y > 1:
dims = (attribute.dim_y, attribute.dim_x)
else:
dims = (attribute.dim_x,)
dtype = attribute.numpy_type
return dims, dtype
def _setup_mapping(self, name, section, dtype):
"""
takes all gathered data to configure and return the correct read and write functions
"""
def read_function():
value = self.config.get(section, name)
value = ini_to_numpy_dict[dtype](value)
return value
def write_function(write_value):
self.config.set(section, name, write_value)
fp = open(self.filename, 'w')
self.config.write(fp)
return read_function, write_function
def setup_attribute(self, annotation=None, attribute=None):
"""
MANDATORY function: is used by the attribute wrapper to get read/write functions.
must return the read and write functions
"""
# process the comms_annotation
section, name = self._setup_annotation(annotation)
# get all the necessary data to set up the read/write functions from the attribute_wrapper
dims, dtype = self._setup_value_conversion(attribute)
# configure and return the read/write functions
read_function, write_function = self._setup_mapping(name, section, dtype)
# return the read/write functions
return read_function, write_function
devices/example/example.ini deleted 100644 → 0
+ 0
− 18
View file @ 1e24f45d
[scalar]
double_scalar = 1.2
bool_scalar = True
int_scalar = 3
str_scalar = "hello"
[spectrum]
double_spectrum = [1.2, 3.4, 5.6, 7.8]
bool_spectrum = [True, False, False, True]
int_spectrum = [1, 2, 3, 4]
str_spectrum = ["hi", "how", "are", "you?"]
[image]
double_image = [[1.2, 3.4], [5.6, 7.8], [9.0, 1.2]]
bool_image = [[True, False], [False, False], [True, True]]
int_image = [[1, 2], [3, 4], [5,6]]
str_image = [["a", "b"], ["c", "d"], ["e", "f"]]
devices/ini_device.py deleted 100644 → 0
+ 0
− 115
View file @ 1e24f45d
# -*- coding: utf-8 -*-
#
# This file wraps around a tango device class and provides a number of abstractions useful for hardware devices. It works together
#
# Distributed under the terms of the APACHE license.
# See LICENSE.txt for more info.
"""
"""
# PyTango imports
from tango.server import run
from tango.server import device_property
from tango import DevState
# Additional import
from src.attribute_wrapper import *
from src.hardware_device import *
from clients.ini_client import *
__all__ = ["ini_device"]
class ini_device(hardware_device):
"""
This class is the minimal (read empty) implementation of a class using 'hardware_device'
"""
# ----------
# Attributes
# ----------
"""
attribute wrapper objects can be declared here. All attribute wrapper objects will get automatically put in a list (attr_list) for easy access
example = attribute_wrapper(comms_annotation="this is an example", datatype=numpy.double, dims=(8, 2), access=AttrWriteType.READ_WRITE)
...
"""
double_scalar_RW = attribute_wrapper(comms_annotation={"section": "scalar", "name": "double_scalar"}, datatype=numpy.double, access=AttrWriteType.READ_WRITE)
double_scalar_R = attribute_wrapper(comms_annotation={"section": "scalar", "name": "double_scalar"}, datatype=numpy.double)
bool_scalar_RW = attribute_wrapper(comms_annotation={"section": "scalar", "name": "bool_scalar"}, datatype=numpy.bool_, access=AttrWriteType.READ_WRITE)
bool_scalar_R = attribute_wrapper(comms_annotation={"section": "scalar", "name": "bool_scalar"}, datatype=numpy.bool_)
int_scalar_RW = attribute_wrapper(comms_annotation={"section": "scalar", "name": "int_scalar"}, datatype=numpy.int64, access=AttrWriteType.READ_WRITE)
int_scalar_R = attribute_wrapper(comms_annotation={"section": "scalar", "name": "int_scalar"}, datatype=numpy.int64)
str_scalar_RW = attribute_wrapper(comms_annotation={"section": "scalar", "name": "str_scalar"}, datatype=numpy.str, access=AttrWriteType.READ_WRITE)
str_scalar_R = attribute_wrapper(comms_annotation={"section": "scalar", "name": "str_scalar"}, datatype=numpy.str)
double_spectrum_RW = attribute_wrapper(comms_annotation={"section": "spectrum", "name": "double_spectrum"}, datatype=numpy.double, dims=(4,), access=AttrWriteType.READ_WRITE)
double_spectrum_R = attribute_wrapper(comms_annotation={"section": "spectrum", "name": "double_spectrum"}, datatype=numpy.double, dims=(4,))
bool_spectrum_RW = attribute_wrapper(comms_annotation={"section": "spectrum", "name": "bool_spectrum"}, datatype=numpy.bool_, dims=(4,), access=AttrWriteType.READ_WRITE)
bool_spectrum_R = attribute_wrapper(comms_annotation={"section": "spectrum", "name": "bool_spectrum"}, datatype=numpy.bool_, dims=(4,))
int_spectrum_RW = attribute_wrapper(comms_annotation={"section": "spectrum", "name": "int_spectrum"}, datatype=numpy.int64, dims=(4,), access=AttrWriteType.READ_WRITE)
int_spectrum_R = attribute_wrapper(comms_annotation={"section": "spectrum", "name": "int_spectrum"}, datatype=numpy.int64, dims=(4,))
str_spectrum_RW = attribute_wrapper(comms_annotation={"section": "spectrum", "name": "str_spectrum"}, datatype=numpy.str, dims=(4,), access=AttrWriteType.READ_WRITE)
str_spectrum_R = attribute_wrapper(comms_annotation={"section": "spectrum", "name": "str_spectrum"}, datatype=numpy.str, dims=(4,))
double_image_RW = attribute_wrapper(comms_annotation={"section": "image", "name": "double_image"}, datatype=numpy.double, dims=(3, 2), access=AttrWriteType.READ_WRITE)
double_image_R = attribute_wrapper(comms_annotation={"section": "image", "name": "double_image"}, datatype=numpy.double, dims=(3, 2))
bool_image_RW = attribute_wrapper(comms_annotation={"section": "image", "name": "bool_image"}, datatype=numpy.bool_, dims=(3, 2), access=AttrWriteType.READ_WRITE)
bool_image_R = attribute_wrapper(comms_annotation={"section": "image", "name": "bool_image"}, datatype=numpy.bool_, dims=(3, 2))
int_image_RW = attribute_wrapper(comms_annotation={"section": "image", "name": "int_image"}, datatype=numpy.int64, dims=(3, 2), access=AttrWriteType.READ_WRITE)
int_image_R = attribute_wrapper(comms_annotation={"section": "image", "name": "int_image"}, datatype=numpy.int64, dims=(3, 2))
str_image_RW = attribute_wrapper(comms_annotation={"section": "image", "name": "str_image"}, datatype=numpy.str, dims=(3, 2), access=AttrWriteType.READ_WRITE)
str_image_R = attribute_wrapper(comms_annotation={"section": "image", "name": "str_image"}, datatype=numpy.str, dims=(3, 2))
def always_executed_hook(self):
"""Method always executed before any TANGO command is executed."""
pass
def delete_device(self):
"""Hook to delete resources allocated in init_device.
This method allows for any memory or other resources allocated in the
init_device method to be released. This method is called by the device
destructor and by the device Init command (a Tango built-in).
"""
self.debug_stream("Shutting down...")
self.Off()
self.debug_stream("Shut down. Good bye.")
# --------
# overloaded functions
# --------
def initialise(self):
""" user code here. is called when the sate is set to INIT """
"""Initialises the attributes and properties of the PCC."""
self.set_state(DevState.INIT)
# set up the OPC ua client
self.ini_client = ini_client("example/example.ini", self.Fault, self)
# map an access helper class
for i in self.attr_list():
i.set_comm_client(self.ini_client)
self.ini_client.start()
# ----------
# Run server
# ----------
def main(args=None, **kwargs):
"""Main function of the hardware device module."""
return run((ini_device,), args=args, **kwargs)
if __name__ == '__main__':
main()
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment