Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use push_namespace method in Slash

Best Python code snippet using slash

system_template_py3.py

Source:system_template_py3.py

1import json2import redis3import redis_graph_py34from .  import redis_graph_functions5from   .redis_graph_functions  import Build_Configuration6from    .redis_graph_functions  import Query_Configuration7class Construct_System(Build_Configuration):8   def __init__( self, db = 14 ):9         print("db",db)10         redis_handle  = redis.StrictRedis( host = "localhost", port=6379, db=db , decode_responses=True)11      12         super().__init__(redis_handle )13    14   def construct_system( self,name,properties={}):15   16       self.construct_node( push_namespace = True,  relationship="SYSTEM", label = name,  17          properties=properties)18       19   def end_system( self):20       self.pop_namespace()21   def construct_site( self,name, address, properties={}):22       properties["address"] = address23       self.construct_node(  push_namespace=True,relationship="SITE", label=name,  24               properties =properties)25   def end_site( self ):26      self.pop_namespace()27   def add_redis_data_store( self, name, ip, port=6379, properties = {} ):28       29       properties["ip"] = ip30       properties["port"] = port31       self.construct_node( push_namespace=True,relationship="DATA_STORE", label=name,32                               properties= properties )33   def start_moisture_store( self ):34       self.construct_node( push_namespace=True,relationship="MOISTURE_STORE", label="MOISTURE_STORE",35                               properties= {} )36   def end_moisture_store( self ):37      self.pop_namespace()38      39   def add_moisture_sensor_store( self, name, description, description_map, depth_map, update_time ):40       properties = {}41       properties["description"] = description42       properties["description_map"] = json.dumps(description_map)43       properties["update_time"] = update_time44       properties["depth_map"] = json.dumps(depth_map)45       self.construct_node( push_namespace=True,relationship="MOISTURE_DATA", label=name,46                               properties= properties )47   48   def add_status_store( self, name, queue_name):49       properties = {}50       properties["queue_name"] = queue_name51       self.construct_node( push_namespace=True,relationship="STATUS_STORE", label=name,52                               properties= properties )53   def start_info_store( self ):54       self.construct_node( push_namespace=True,relationship="INFO_STORE", label="INFO_STORE",55                               properties= {} )56  57   def add_eto_store(self ):58       self.construct_node( push_namespace=False,relationship="ETO_STORE", label="ETO_STORE",59                               properties= {} )60   def add_air_temperature_humidity_store(self):61       self.construct_node( push_namespace=False,relationship="TEMP_HUMIDITY", label="TEMP_HUMIDITY",62                               properties= {} )63   def add_air_temperature_humidity_daily_log(self):64       self.construct_node( push_namespace=False,relationship="TEMP_HUMIDITY_DAILY", label="TEMP_HUMIDITY_DAILY",65                               properties= {} )66       self.construct_node( push_namespace=False,relationship="TEMP_HUMIDITY_DAILY_ETO", label="TEMP_HUMIDITY_DAILY_ETO",67                               properties= {} )68   def end_info_store(self):   69       self.pop_namespace()70   def end_redis_data_store( self):71       self.pop_namespace()72   def add_udp_io_sever(self, name, ip,remote_type, port, properties={} ):73       properties["ip"] = ip74       properties["remote_type"] = remote_type75       properties["port"] = port76       return self.construct_node(  push_namespace=True,relationship="UDP_IO_SERVER", 77               label=name, properties = properties )78   def end_udp_io_server(self ):79       self.pop_namespace()80   def add_rtu_interface(self, name ,protocol, baud_rate, properties={} ):81       properties["protocol"]= protocol82       properties["baud_rate"] = baud_rate83       return self.construct_node(  push_namespace=True,relationship="RTU_INTERFACE", 84                 label=name,properties = properties)85   def end_rtu_interface( self ):86       self.pop_namespace()87   def add_remote( self, name,modbus_address,type, function, properties = {}):88 89       properties["modbus_address"] = modbus_address90       properties["type"]           = type91       properties["function"]       = function92       self.construct_node(  push_namespace=True,relationship="REMOTE", label=name, 93               properties = properties )94   def construct_controller( self,name, ip,type,properties={} ):95       properties["name"] = name96       properties["ip"]   = ip97       self.construct_node(  push_namespace=True,relationship="CONTROLLER", label=name, 98               properties = properties)99   def end_controller( self ):100       self.pop_namespace()101   def start_service( self, properties = {} ):102       self.construct_node(  push_namespace=TRUE,relationship="SERVICES", label=name, 103               properties = properties)104   def construct_web_server( self, name,url,properties = {} ):105       properties["url"]   = url106       self.construct_node(  push_namespace=False,relationship="WEB_SERVER", label=name, 107               properties = properties)108   def add_rabbitmq_command_rpc_queue( self,name, properties = {} ):109       110       self.construct_node(  push_namespace=False,relationship="COMMAND_RPC_QUEUE", label=name, 111                                properties = properties)112   def add_rabbitmq_web_rpc_queue( self,name, properties = {} ):113 114       115       self.construct_node(  push_namespace=False,relationship="WEB_RPC_QUEUE", label=name, 116                                properties = properties)117   def add_rabbitmq_event_queue( self,name, properties = {} ):118       119       self.construct_node(  push_namespace=False,relationship="RABBITMQ_EVENT_QUEUE", label=name, 120                                 properties = properties)121   def add_rabbitmq_status_queue( self,name,vhost,queue,port,server  ):122       properties          = {}123  124       properties["vhost"]    = vhost125       properties["queue"]    = queue126       properties["port"]     = port127       properties["server"]   = server128       129       self.construct_node(  push_namespace=False,relationship="RABBITMQ_STATUS_QUEUE", label=name, 130                                 properties = properties)131   def add_ntpd_server( self,name, properties = {} ):132       133       self.construct_node(  push_namespace=False,relationship="NTPD_SERVER", label=name, 134                                properties = properties)135   def start_eto_server( self,name, properties = {} ):136 137       self.construct_node(  push_namespace=False,relationship="ETO_SERVER", label=name, properties = properties)138   def add_eto_setup_code( self, access_codes, altitude , properties = {} ):139       properties["messo_eto"]     = json.dumps( access_codes["messo_eto"] )140       properties["messo_precp"]   = json.dumps( access_codes["messo_precp"] )141       properties["cimis_eto"]     = json.dumps( access_codes["cimis_eto"] )142       properties["cimis_spatial"] = json.dumps( access_codes["cimis_spatial"])143       properties["altitude"]      = altitude144       145       self.construct_node(  push_namespace=False,relationship="ETO_SETUP_DATA", label="ETO_SETUP_DATA", 146                                 properties = properties)147   def end_eto_server(self):148         self.pop_namespace()149   150   def add_linux_server_monitor( self, name,properties = {} ):151       properties["name"]  = "Linux Server Monitor"152       153       self.construct_node(  push_namespace=False,relationship="LINUX_SERVER_MONITOR", label=name, properties = properties)154   def add_schedule_monitoring( self, name,properties = {} ):155       156       self.construct_node(  push_namespace=False,relationship="NTPD_SERVER", label=name, properties = properties)157   def add_moisture_monitoring( self, name, properties = {} ):158       159       self.construct_node(  push_namespace=False,relationship="NTPD_SERVER", label=name, properties = properties)160   def irrigation_monitoring( self, name,properties = {} ):161       162       self.construct_node(  push_namespace=False,relationship="IRRIGATION_MONITOR", label=name, properties = properties)163   def add_device_monitoring( self, name, properties = {} ):164 165       166       self.construct_node(  push_namespace=False,relationship="DEVICE_MONITOR", label=name, properties = properties)167   def add_process_monitoring( self,name, properties = {}  ):      168       self.construct_node(  push_namespace=False,relationship="PROCESS_MONITOR", label=name, properties = properties)169   def add_watch_dog_monitoring( self,name, properties = {}  ):170       171       self.construct_node(  push_namespace=False,relationship="WATCH_DOG_MONITORING", label=name, properties = properties)172   def add_io_collection( self, name, properties = {} ):173       174       self.construct_node(  push_namespace=False,relationship="PROCESS_MONITOR", label=name, properties = properties)175   def add_local_ai( self, name, properties = {} ):       176       self.construct_node(  push_namespace=False,relationship="PROCESS_MONITOR", label=name, properties = properties)177   178class Graph_Management(Query_Configuration):179   def __init__( self , controller_name , io_server_name, data_store_name,db = 14 ):180      self.redis_handle  = redis.StrictRedis( host = "localhost", port=6379, db =db , decode_responses=True)181   182      super().__init__( self.redis_handle)183      184      self.controller_name = controller_name185      self.io_server_name  = io_server_name186      self.data_store_name = data_store_name187      self.initialize_cb_handlers()188   def find_remotes( self  ):189      data = self.match_terminal_relationship( "REMOTE_UNIT", label= None , starting_set = None )190 191      return data192   def find_data_stores( self ):193       data = self.match_terminal_relationship( "DATA_STORE", label= None , starting_set = None )194       return data195   def find_io_servers( self ):196      data = self.match_terminal_relationship( "UDP_IO_SERVER", label= None , starting_set = None )197      return data198   199   def initialize_cb_handlers( self ):200       self.cb_handlers = {}201   def add_cb_handler( self, tag, function ):202       self.cb_handlers[ tag ] = function203   def verify_handler( self, tag ):204       try:205           return tag in self.cb_handlers206       except:207          #print "handlers:", type(self.cb_handlers)208          #print "tag", tag209          raise        210   def execute_cb_handlers( self, tag, value, parameters ):  # parameters is a list211       function = self.cb_handlers[tag]...

farm_template.py

Source:farm_template.py

1import json2import redis3from redis_graph_populate import Build_Configuration4from redis_graph_common import Redis_Graph_Common5import copy6class Construct_Farm():7   def __init__( self, bc):8      self.bc = bc # Build configuration in graph_functions9   def construct_system( self,name=None):10       self.bc.construct_node( push_namespace = True,  relationship="SYSTEM", label = "SYSTEM", name = name, properties={})11       12   def end_system( self):13       self.bc.pop_namespace()14   def construct_site( self,name=None,wired=True,address=None):15       self.bc.construct_node(  push_namespace=True,relationship="SITE", label="SITE", name=name, 16               properties ={"wired":wired,"address":address})17   def end_site( self ):18      self.bc.pop_namespace()19   def construct_controller( self,name,web_queue,rpc_queue,local_ip,controller_type,vhost,card_dict,redis_controller_key ):20       card_dict_json = json.dumps( card_dict )21       self.bc.construct_node(  push_namespace=True,relationship="CONTROLLER", label="CONTROLLER", name=name, 22               properties ={"web_queue":web_queue, "rpc_queue":rpc_queue,"local_ip":local_ip,"controller_type":controller_type,"vhost":vhost,"card_dict":card_dict_json, 23                             "irrigation_resets":0,"system_resets":0, "ping_loss":0, "ping_counts":0,"temperature":0 ,"redis_key":redis_controller_key })24   def end_controller( self ):25       self.bc.pop_namespace()26   def add_event_queue( self,name, events ):27       self.bc.construct_node(  push_namespace=False,relationship="EVENT_QUEUE", label="EVENT_QUEUE", name=name,28                                    properties = {  "timestamp":0, "events":json.dumps(events) } )29   def add_diagnostic_card_header( self, *args):30       self.bc.construct_node(  push_namespace=True,31                                    relationship="DIAGNOSTIC_CARD_HEADER", 32                                    label="DIAGNOSTIC_CARD_HEADER", 33                                    name="DIAGNOSTIC_CARD_HEADER",34                                    properties = {} )35   def end_diagnostic_card_header( self, *args):36       self.bc.pop_namespace()  37   def add_diagnostic_card( self, org_name, board_name, list_name, card_name,description=None ):38       if description == None:39           description = card_name40       self.bc.construct_node(  push_namespace=False,41                                relationship="DIAGNOSTIC_CARD", 42                                label="DIAGNOSTIC_CARD", 43                                name = card_name,44                                 properties = { "org_name":org_name, "board_name":board_name, "list_name":list_name, "description":description,"label":"green","new_commit":[]  } )45   def add_schedule_header( self ):46       return self.bc.construct_node(  push_namespace=True,relationship="Schedule_Header", label="Schedule_Header", name="Schedule_Header", 47               properties ={})48   def end_schedule_header( self ):49       self.bc.pop_namespace()   50   def add_schedule( self,name,number,flow_sensor_names ,card_link ):51       schedule_node = self.bc.construct_node(  push_namespace=True,relationship="IRRIGATION_SCHEDULE", label="IRRIGATION_SCHEDULE", name=name, 52                       properties ={"number":number})53       for i in range(0,number):54           self.bc.construct_node(  push_namespace=True,relationship="STEP", label="STEP", name=str(i+1),  properties ={ "card":card_link+str(i+1) } )55           self.bc.construct_node(  push_namespace=True,relationship="FLOW_SENSOR_HEADERS", label="FLOW_SENSOR_HEADERS", name="FLOW_SENSOR_HEADERS", 56                       properties ={  })57           for j in flow_sensor_names:58               self.bc.construct_node( push_namespace = True,  relationship="FLOW_SENSOR_HEADER", label = "FLOW_SENSOR_HEADER", name = j, properties={} )59               self.bc.construct_node( push_namespace = False, relationship="FLOW_SENSOR_LIMIT", label = "FLOW_SENSOR_LIMIT", name = j, properties={} )60               self.bc.construct_node( push_namespace = False, relationship="FLOW_SENSOR_VALUE", label = "FLOW_SENSOR_VALUE", name = j, properties={} )61               self.bc.pop_namespace()62           self.bc.pop_namespace()63           self.bc.construct_node(  push_namespace=False,relationship="COIL_CURRENT", label="COIL_CURRENT", name= "COIL_CURRENT", 64                       properties ={ })65           self.bc.construct_node(  push_namespace=False,relationship="COIL_CURRENT_LIMIT", label="COIL_CURRENT_LIMIT", name= "COIL_CURRENT_LIMIT",66                       properties ={ })67           68       69           for j in flow_sensor_names:70               self.bc.construct_node( push_namespace = False, relationship="FLOW_SENSOR_LIMIT", label = "FLOW_SENSOR_LIMIT", name = j, properties={} )71           72           self.bc.pop_namespace()73  74       self.bc.pop_namespace()75   def add_flow_sensor_header( self ):76       return self.bc.construct_node(  push_namespace=True,relationship="FLOW_SENSOR_HEADER", label="FLOW_SENSOR_HEADER", name="flow_sensor_header", 77               properties ={})78   def end_flow_sensor_header( self ):79       self.bc.pop_namespace()   80   def add_flow_sensor( self,name,controller,io,conversion_factor):81       return self.bc.construct_node(  push_namespace=False,relationship="FLOW_SENSOR", label="FLOW_SENSOR", name=name, 82               properties ={"name":name,"controller":controller,"io":io,"conversion_factor":conversion_factor})83   84   def add_udp_io_sever(self, name, ip,remote_type, port, redis_key ):85       return self.bc.construct_node(  push_namespace=True,relationship="UDP_IO_SERVER", label="UDP_IO_SERVER", name=name, 86               properties ={"name":name,"ip":ip,"remote_type":remote_type,"port":port,"redis_key":redis_key })87   def end_udp_io_server(self ):88       self.bc.pop_namespace()89   def add_rtu_interface(self, name ,protocol,baud_rate ):90       return self.bc.construct_node(  push_namespace=True,relationship="RTU_INTERFACE", label="RTU_INTERFACE", name=name, 91               properties ={"name":name,"protocol":protocol,"baud_rate":baud_rate })92   def add_remote( self, name,modbus_address,irrigation_station_number, card_dict):93       card_dict_json = json.dumps(card_dict)94       self.bc.construct_node(  push_namespace=True,relationship="REMOTE", label="REMOTE", name=name, 95               properties ={"name":name,"modbus_address":modbus_address,"irrigation_station_number":irrigation_station_number, "card_dict":card_dict_json})96       self.bc.construct_node(  push_namespace=True,relationship="IRRIGATION_VALVE_CURRENT_HEADER", label="IRRIGATION_VALVE_CURRENT_HEADER", name = "valve_current_header", 97           properties ={ })           98       for i in range(0,irrigation_station_number):99           self.bc.construct_node(  push_namespace=False,relationship="IRRIGATION_VALVE_CURRENT", label="IRRIGATION_VALVE_CURRENT", name = str(i+1), 100           properties ={ "active":False })         101       self.bc.pop_namespace()102       self.bc.construct_node(  push_namespace=True,relationship="IRRIGATION_VALVE_CURRENT_HEADER", label="IRRIGATION_VALVE_CURRENT_LIMIT_HEADER", name = "valve_current_limit_header", 103           properties ={ })           104       for i in range(0,irrigation_station_number):  105           self.bc.construct_node(  push_namespace=False,relationship="IRRIGATION_VALVE_CURRENT_LIMIT", label="IRRIGATION_VALVE_CURRENT_LIMIT", name= str(i+1), 106           properties ={ "active":False  })107  108       self.bc.pop_namespace()109       self.bc.pop_namespace()110   def end_rtu_interface( self ):...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.