# SPDX-License-Identifier: BSD-3-Clause # # Copyright 2020 Raritan Inc. All rights reserved. # # This is an auto-generated file. # # Section generated by IdlC from "UserEvent.idl" # import raritan.rpc from raritan.rpc import Interface, Structure, ValueObject, Enumeration, typecheck, DecodeException import raritan.rpc.idl # value object class UserEvent(raritan.rpc.idl.Event): idlType = "event.UserEvent:1.0.0" def __init__(self, actUserName, actIpAddr, source): super(raritan.rpc.event.UserEvent, self).__init__(source) typecheck.is_string(actUserName, AssertionError) typecheck.is_string(actIpAddr, AssertionError) self.actUserName = actUserName self.actIpAddr = actIpAddr def encode(self): json = super(raritan.rpc.event.UserEvent, self).encode() json['actUserName'] = self.actUserName json['actIpAddr'] = self.actIpAddr return json @classmethod def decode(cls, json, agent): obj = cls( actUserName = json['actUserName'], actIpAddr = json['actIpAddr'], # for idl.Event source = Interface.decode(json['source'], agent), ) return obj def listElements(self): elements = ["actUserName", "actIpAddr"] elements = elements + super(raritan.rpc.event.UserEvent, self).listElements() return elements # # Section generated by IdlC from "DataPushService.idl" # import raritan.rpc from raritan.rpc import Interface, Structure, ValueObject, Enumeration, typecheck, DecodeException import raritan.rpc.event import raritan.rpc.idl # interface class DataPushService(Interface): idlType = "event.DataPushService:1.0.1" # enumeration class EntryType(Enumeration): idlType = "event.DataPushService_1_0_1.EntryType:1.0.0" values = ["SENSORLIST", "SENSORLOG", "AMSLIST", "AMSLOG", "AUDITLOG"] EntryType.SENSORLIST = EntryType(0) EntryType.SENSORLOG = EntryType(1) EntryType.AMSLIST = EntryType(2) EntryType.AMSLOG = EntryType(3) EntryType.AUDITLOG = EntryType(4) ERR_NO_SUCH_ID = 1 ERR_INVALID_PARAMS = 2 ERR_MAX_ENTRIES_REACHED = 3 # structure class EntrySettings(Structure): idlType = "event.DataPushService_1_0_1.EntrySettings:1.0.0" elements = ["url", "allowOffTimeRangeCerts", "caCertChain", "useAuth", "username", "password", "type", "items"] def __init__(self, url, allowOffTimeRangeCerts, caCertChain, useAuth, username, password, type, items): typecheck.is_string(url, AssertionError) typecheck.is_bool(allowOffTimeRangeCerts, AssertionError) typecheck.is_string(caCertChain, AssertionError) typecheck.is_bool(useAuth, AssertionError) typecheck.is_string(username, AssertionError) typecheck.is_string(password, AssertionError) typecheck.is_enum(type, raritan.rpc.event.DataPushService.EntryType, AssertionError) for x0 in items: typecheck.is_string(x0, AssertionError) self.url = url self.allowOffTimeRangeCerts = allowOffTimeRangeCerts self.caCertChain = caCertChain self.useAuth = useAuth self.username = username self.password = password self.type = type self.items = items @classmethod def decode(cls, json, agent): obj = cls( url = json['url'], allowOffTimeRangeCerts = json['allowOffTimeRangeCerts'], caCertChain = json['caCertChain'], useAuth = json['useAuth'], username = json['username'], password = json['password'], type = raritan.rpc.event.DataPushService.EntryType.decode(json['type']), items = [x0 for x0 in json['items']], ) return obj def encode(self): json = {} json['url'] = self.url json['allowOffTimeRangeCerts'] = self.allowOffTimeRangeCerts json['caCertChain'] = self.caCertChain json['useAuth'] = self.useAuth json['username'] = self.username json['password'] = self.password json['type'] = raritan.rpc.event.DataPushService.EntryType.encode(self.type) json['items'] = [x0 for x0 in self.items] return json # structure class EntryStatus(Structure): idlType = "event.DataPushService_1_0_1.EntryStatus:1.0.0" elements = ["busy", "rescheduled", "lastAttemptTime", "lastSuccessTime"] def __init__(self, busy, rescheduled, lastAttemptTime, lastSuccessTime): typecheck.is_bool(busy, AssertionError) typecheck.is_bool(rescheduled, AssertionError) typecheck.is_time(lastAttemptTime, AssertionError) typecheck.is_time(lastSuccessTime, AssertionError) self.busy = busy self.rescheduled = rescheduled self.lastAttemptTime = lastAttemptTime self.lastSuccessTime = lastSuccessTime @classmethod def decode(cls, json, agent): obj = cls( busy = json['busy'], rescheduled = json['rescheduled'], lastAttemptTime = raritan.rpc.Time.decode(json['lastAttemptTime']), lastSuccessTime = raritan.rpc.Time.decode(json['lastSuccessTime']), ) return obj def encode(self): json = {} json['busy'] = self.busy json['rescheduled'] = self.rescheduled json['lastAttemptTime'] = raritan.rpc.Time.encode(self.lastAttemptTime) json['lastSuccessTime'] = raritan.rpc.Time.encode(self.lastSuccessTime) return json # value object class EntryAddedEvent(UserEvent): idlType = "event.DataPushService_1_0_1.EntryAddedEvent:1.0.0" def __init__(self, entryId, settings, actUserName, actIpAddr, source): super(raritan.rpc.event.DataPushService.EntryAddedEvent, self).__init__(actUserName, actIpAddr, source) typecheck.is_int(entryId, AssertionError) typecheck.is_struct(settings, raritan.rpc.event.DataPushService.EntrySettings, AssertionError) self.entryId = entryId self.settings = settings def encode(self): json = super(raritan.rpc.event.DataPushService.EntryAddedEvent, self).encode() json['entryId'] = self.entryId json['settings'] = raritan.rpc.event.DataPushService.EntrySettings.encode(self.settings) return json @classmethod def decode(cls, json, agent): obj = cls( entryId = json['entryId'], settings = raritan.rpc.event.DataPushService.EntrySettings.decode(json['settings'], agent), # for event.UserEvent actUserName = json['actUserName'], actIpAddr = json['actIpAddr'], # for idl.Event source = Interface.decode(json['source'], agent), ) return obj def listElements(self): elements = ["entryId", "settings"] elements = elements + super(raritan.rpc.event.DataPushService.EntryAddedEvent, self).listElements() return elements # value object class EntryModifiedEvent(UserEvent): idlType = "event.DataPushService_1_0_1.EntryModifiedEvent:1.0.0" def __init__(self, entryId, oldSettings, newSettings, actUserName, actIpAddr, source): super(raritan.rpc.event.DataPushService.EntryModifiedEvent, self).__init__(actUserName, actIpAddr, source) typecheck.is_int(entryId, AssertionError) typecheck.is_struct(oldSettings, raritan.rpc.event.DataPushService.EntrySettings, AssertionError) typecheck.is_struct(newSettings, raritan.rpc.event.DataPushService.EntrySettings, AssertionError) self.entryId = entryId self.oldSettings = oldSettings self.newSettings = newSettings def encode(self): json = super(raritan.rpc.event.DataPushService.EntryModifiedEvent, self).encode() json['entryId'] = self.entryId json['oldSettings'] = raritan.rpc.event.DataPushService.EntrySettings.encode(self.oldSettings) json['newSettings'] = raritan.rpc.event.DataPushService.EntrySettings.encode(self.newSettings) return json @classmethod def decode(cls, json, agent): obj = cls( entryId = json['entryId'], oldSettings = raritan.rpc.event.DataPushService.EntrySettings.decode(json['oldSettings'], agent), newSettings = raritan.rpc.event.DataPushService.EntrySettings.decode(json['newSettings'], agent), # for event.UserEvent actUserName = json['actUserName'], actIpAddr = json['actIpAddr'], # for idl.Event source = Interface.decode(json['source'], agent), ) return obj def listElements(self): elements = ["entryId", "oldSettings", "newSettings"] elements = elements + super(raritan.rpc.event.DataPushService.EntryModifiedEvent, self).listElements() return elements # value object class EntryDeletedEvent(UserEvent): idlType = "event.DataPushService_1_0_1.EntryDeletedEvent:1.0.0" def __init__(self, entryId, actUserName, actIpAddr, source): super(raritan.rpc.event.DataPushService.EntryDeletedEvent, self).__init__(actUserName, actIpAddr, source) typecheck.is_int(entryId, AssertionError) self.entryId = entryId def encode(self): json = super(raritan.rpc.event.DataPushService.EntryDeletedEvent, self).encode() json['entryId'] = self.entryId return json @classmethod def decode(cls, json, agent): obj = cls( entryId = json['entryId'], # for event.UserEvent actUserName = json['actUserName'], actIpAddr = json['actIpAddr'], # for idl.Event source = Interface.decode(json['source'], agent), ) return obj def listElements(self): elements = ["entryId"] elements = elements + super(raritan.rpc.event.DataPushService.EntryDeletedEvent, self).listElements() return elements # value object class EntryStatusChangedEvent(raritan.rpc.idl.Event): idlType = "event.DataPushService_1_0_1.EntryStatusChangedEvent:1.0.0" def __init__(self, entryId, newStatus, source): super(raritan.rpc.event.DataPushService.EntryStatusChangedEvent, self).__init__(source) typecheck.is_int(entryId, AssertionError) typecheck.is_struct(newStatus, raritan.rpc.event.DataPushService.EntryStatus, AssertionError) self.entryId = entryId self.newStatus = newStatus def encode(self): json = super(raritan.rpc.event.DataPushService.EntryStatusChangedEvent, self).encode() json['entryId'] = self.entryId json['newStatus'] = raritan.rpc.event.DataPushService.EntryStatus.encode(self.newStatus) return json @classmethod def decode(cls, json, agent): obj = cls( entryId = json['entryId'], newStatus = raritan.rpc.event.DataPushService.EntryStatus.decode(json['newStatus'], agent), # for idl.Event source = Interface.decode(json['source'], agent), ) return obj def listElements(self): elements = ["entryId", "newStatus"] elements = elements + super(raritan.rpc.event.DataPushService.EntryStatusChangedEvent, self).listElements() return elements class _addEntry(Interface.Method): name = 'addEntry' @staticmethod def encode(entrySettings): typecheck.is_struct(entrySettings, raritan.rpc.event.DataPushService.EntrySettings, AssertionError) args = {} args['entrySettings'] = raritan.rpc.event.DataPushService.EntrySettings.encode(entrySettings) return args @staticmethod def decode(rsp, agent): _ret_ = rsp['_ret_'] entryId = rsp['entryId'] typecheck.is_int(_ret_, DecodeException) typecheck.is_int(entryId, DecodeException) return (_ret_, entryId) class _modifyEntry(Interface.Method): name = 'modifyEntry' @staticmethod def encode(entryId, entrySettings): typecheck.is_int(entryId, AssertionError) typecheck.is_struct(entrySettings, raritan.rpc.event.DataPushService.EntrySettings, AssertionError) args = {} args['entryId'] = entryId args['entrySettings'] = raritan.rpc.event.DataPushService.EntrySettings.encode(entrySettings) return args @staticmethod def decode(rsp, agent): _ret_ = rsp['_ret_'] typecheck.is_int(_ret_, DecodeException) return _ret_ class _deleteEntry(Interface.Method): name = 'deleteEntry' @staticmethod def encode(entryId): typecheck.is_int(entryId, AssertionError) args = {} args['entryId'] = entryId return args @staticmethod def decode(rsp, agent): _ret_ = rsp['_ret_'] typecheck.is_int(_ret_, DecodeException) return _ret_ class _getEntry(Interface.Method): name = 'getEntry' @staticmethod def encode(entryId): typecheck.is_int(entryId, AssertionError) args = {} args['entryId'] = entryId return args @staticmethod def decode(rsp, agent): _ret_ = rsp['_ret_'] entrySettings = raritan.rpc.event.DataPushService.EntrySettings.decode(rsp['entrySettings'], agent) typecheck.is_int(_ret_, DecodeException) typecheck.is_struct(entrySettings, raritan.rpc.event.DataPushService.EntrySettings, DecodeException) return (_ret_, entrySettings) class _listEntries(Interface.Method): name = 'listEntries' @staticmethod def encode(): args = {} return args @staticmethod def decode(rsp, agent): _ret_ = dict([( elem['key'], raritan.rpc.event.DataPushService.EntrySettings.decode(elem['value'], agent)) for elem in rsp['_ret_']]) return _ret_ class _pushData(Interface.Method): name = 'pushData' @staticmethod def encode(entryId): typecheck.is_int(entryId, AssertionError) args = {} args['entryId'] = entryId return args @staticmethod def decode(rsp, agent): _ret_ = rsp['_ret_'] typecheck.is_int(_ret_, DecodeException) return _ret_ class _getEntryStatus(Interface.Method): name = 'getEntryStatus' @staticmethod def encode(entryId): typecheck.is_int(entryId, AssertionError) args = {} args['entryId'] = entryId return args @staticmethod def decode(rsp, agent): _ret_ = rsp['_ret_'] entryStatus = raritan.rpc.event.DataPushService.EntryStatus.decode(rsp['entryStatus'], agent) typecheck.is_int(_ret_, DecodeException) typecheck.is_struct(entryStatus, raritan.rpc.event.DataPushService.EntryStatus, DecodeException) return (_ret_, entryStatus) def __init__(self, target, agent): super(DataPushService, self).__init__(target, agent) self.addEntry = DataPushService._addEntry(self) self.modifyEntry = DataPushService._modifyEntry(self) self.deleteEntry = DataPushService._deleteEntry(self) self.getEntry = DataPushService._getEntry(self) self.listEntries = DataPushService._listEntries(self) self.pushData = DataPushService._pushData(self) self.getEntryStatus = DataPushService._getEntryStatus(self) # # Section generated by IdlC from "AlarmManager.idl" # import raritan.rpc from raritan.rpc import Interface, Structure, ValueObject, Enumeration, typecheck, DecodeException import raritan.rpc.event import raritan.rpc.idl # interface class AlarmManager(Interface): idlType = "event.AlarmManager:1.0.0" NO_ERROR = 0 ERR_UNKNOWN_ALARM_ID = 1 ERR_EXECUTING_ACTIONS = 2 # structure class Alert(Structure): idlType = "event.AlarmManager.Alert:1.0.0" elements = ["eventCondition", "message", "firstAppearance", "lastAppearance", "numberAlerts"] def __init__(self, eventCondition, message, firstAppearance, lastAppearance, numberAlerts): typecheck.is_string(eventCondition, AssertionError) typecheck.is_string(message, AssertionError) typecheck.is_time(firstAppearance, AssertionError) typecheck.is_time(lastAppearance, AssertionError) typecheck.is_int(numberAlerts, AssertionError) self.eventCondition = eventCondition self.message = message self.firstAppearance = firstAppearance self.lastAppearance = lastAppearance self.numberAlerts = numberAlerts @classmethod def decode(cls, json, agent): obj = cls( eventCondition = json['eventCondition'], message = json['message'], firstAppearance = raritan.rpc.Time.decode(json['firstAppearance']), lastAppearance = raritan.rpc.Time.decode(json['lastAppearance']), numberAlerts = json['numberAlerts'], ) return obj def encode(self): json = {} json['eventCondition'] = self.eventCondition json['message'] = self.message json['firstAppearance'] = raritan.rpc.Time.encode(self.firstAppearance) json['lastAppearance'] = raritan.rpc.Time.encode(self.lastAppearance) json['numberAlerts'] = self.numberAlerts return json # structure class Alarm(Structure): idlType = "event.AlarmManager.Alarm:1.0.0" elements = ["id", "name", "actionId", "alerts"] def __init__(self, id, name, actionId, alerts): typecheck.is_string(id, AssertionError) typecheck.is_string(name, AssertionError) typecheck.is_string(actionId, AssertionError) for x0 in alerts: typecheck.is_struct(x0, raritan.rpc.event.AlarmManager.Alert, AssertionError) self.id = id self.name = name self.actionId = actionId self.alerts = alerts @classmethod def decode(cls, json, agent): obj = cls( id = json['id'], name = json['name'], actionId = json['actionId'], alerts = [raritan.rpc.event.AlarmManager.Alert.decode(x0, agent) for x0 in json['alerts']], ) return obj def encode(self): json = {} json['id'] = self.id json['name'] = self.name json['actionId'] = self.actionId json['alerts'] = [raritan.rpc.event.AlarmManager.Alert.encode(x0) for x0 in self.alerts] return json # value object class AlarmAddedEvent(raritan.rpc.idl.Event): idlType = "event.AlarmManager.AlarmAddedEvent:1.0.0" def __init__(self, alarm, source): super(raritan.rpc.event.AlarmManager.AlarmAddedEvent, self).__init__(source) typecheck.is_struct(alarm, raritan.rpc.event.AlarmManager.Alarm, AssertionError) self.alarm = alarm def encode(self): json = super(raritan.rpc.event.AlarmManager.AlarmAddedEvent, self).encode() json['alarm'] = raritan.rpc.event.AlarmManager.Alarm.encode(self.alarm) return json @classmethod def decode(cls, json, agent): obj = cls( alarm = raritan.rpc.event.AlarmManager.Alarm.decode(json['alarm'], agent), # for idl.Event source = Interface.decode(json['source'], agent), ) return obj def listElements(self): elements = ["alarm"] elements = elements + super(raritan.rpc.event.AlarmManager.AlarmAddedEvent, self).listElements() return elements # value object class AlarmUpdatedEvent(raritan.rpc.idl.Event): idlType = "event.AlarmManager.AlarmUpdatedEvent:1.0.0" def __init__(self, alarm, source): super(raritan.rpc.event.AlarmManager.AlarmUpdatedEvent, self).__init__(source) typecheck.is_struct(alarm, raritan.rpc.event.AlarmManager.Alarm, AssertionError) self.alarm = alarm def encode(self): json = super(raritan.rpc.event.AlarmManager.AlarmUpdatedEvent, self).encode() json['alarm'] = raritan.rpc.event.AlarmManager.Alarm.encode(self.alarm) return json @classmethod def decode(cls, json, agent): obj = cls( alarm = raritan.rpc.event.AlarmManager.Alarm.decode(json['alarm'], agent), # for idl.Event source = Interface.decode(json['source'], agent), ) return obj def listElements(self): elements = ["alarm"] elements = elements + super(raritan.rpc.event.AlarmManager.AlarmUpdatedEvent, self).listElements() return elements # value object class AlarmAcknowledgedEvent(raritan.rpc.idl.Event): idlType = "event.AlarmManager.AlarmAcknowledgedEvent:1.0.0" def __init__(self, alarmId, source): super(raritan.rpc.event.AlarmManager.AlarmAcknowledgedEvent, self).__init__(source) typecheck.is_string(alarmId, AssertionError) self.alarmId = alarmId def encode(self): json = super(raritan.rpc.event.AlarmManager.AlarmAcknowledgedEvent, self).encode() json['alarmId'] = self.alarmId return json @classmethod def decode(cls, json, agent): obj = cls( alarmId = json['alarmId'], # for idl.Event source = Interface.decode(json['source'], agent), ) return obj def listElements(self): elements = ["alarmId"] elements = elements + super(raritan.rpc.event.AlarmManager.AlarmAcknowledgedEvent, self).listElements() return elements class _acknowledgeAlarm(Interface.Method): name = 'acknowledgeAlarm' @staticmethod def encode(alarmId): typecheck.is_string(alarmId, AssertionError) args = {} args['alarmId'] = alarmId return args @staticmethod def decode(rsp, agent): _ret_ = rsp['_ret_'] typecheck.is_int(_ret_, DecodeException) return _ret_ class _listAlarms(Interface.Method): name = 'listAlarms' @staticmethod def encode(): args = {} return args @staticmethod def decode(rsp, agent): _ret_ = [raritan.rpc.event.AlarmManager.Alarm.decode(x0, agent) for x0 in rsp['_ret_']] for x0 in _ret_: typecheck.is_struct(x0, raritan.rpc.event.AlarmManager.Alarm, DecodeException) return _ret_ def __init__(self, target, agent): super(AlarmManager, self).__init__(target, agent) self.acknowledgeAlarm = AlarmManager._acknowledgeAlarm(self) self.listAlarms = AlarmManager._listAlarms(self) # # Section generated by IdlC from "TimerEventManager.idl" # import raritan.rpc from raritan.rpc import Interface, Structure, ValueObject, Enumeration, typecheck, DecodeException import raritan.rpc.event # interface class TimerEventManager(Interface): idlType = "event.TimerEventManager:2.0.0" NO_ERROR = 0 ERR_INVALID_SCHEDULE = 1 ERR_UNKNOWN_EVENT_ID = 2 ERR_CREATE_EVENT_ID_FAILED = 3 ERR_MAX_TIMERS_CREATED = 4 JAN = 1 FEB = 2 MAR = 3 APR = 4 MAY = 5 JUN = 6 JUL = 7 AUG = 8 SEP = 9 OCT = 10 NOV = 11 DEC = 12 SUN = 0 MON = 1 TUE = 2 WED = 3 THU = 4 FRI = 5 SAT = 6 # structure class Range(Structure): idlType = "event.TimerEventManager_2_0_0.Range:1.0.0" elements = ["start", "end", "step"] def __init__(self, start, end, step): typecheck.is_int(start, AssertionError) typecheck.is_int(end, AssertionError) typecheck.is_int(step, AssertionError) self.start = start self.end = end self.step = step @classmethod def decode(cls, json, agent): obj = cls( start = json['start'], end = json['end'], step = json['step'], ) return obj def encode(self): json = {} json['start'] = self.start json['end'] = self.end json['step'] = self.step return json # structure class Schedule(Structure): idlType = "event.TimerEventManager_2_0_0.Schedule:1.0.0" elements = ["enabled", "minute", "hour", "dayOfMonth", "month", "dayOfWeek"] def __init__(self, enabled, minute, hour, dayOfMonth, month, dayOfWeek): typecheck.is_bool(enabled, AssertionError) for x0 in minute: typecheck.is_struct(x0, raritan.rpc.event.TimerEventManager.Range, AssertionError) for x0 in hour: typecheck.is_struct(x0, raritan.rpc.event.TimerEventManager.Range, AssertionError) for x0 in dayOfMonth: typecheck.is_struct(x0, raritan.rpc.event.TimerEventManager.Range, AssertionError) for x0 in month: typecheck.is_struct(x0, raritan.rpc.event.TimerEventManager.Range, AssertionError) for x0 in dayOfWeek: typecheck.is_struct(x0, raritan.rpc.event.TimerEventManager.Range, AssertionError) self.enabled = enabled self.minute = minute self.hour = hour self.dayOfMonth = dayOfMonth self.month = month self.dayOfWeek = dayOfWeek @classmethod def decode(cls, json, agent): obj = cls( enabled = json['enabled'], minute = [raritan.rpc.event.TimerEventManager.Range.decode(x0, agent) for x0 in json['minute']], hour = [raritan.rpc.event.TimerEventManager.Range.decode(x0, agent) for x0 in json['hour']], dayOfMonth = [raritan.rpc.event.TimerEventManager.Range.decode(x0, agent) for x0 in json['dayOfMonth']], month = [raritan.rpc.event.TimerEventManager.Range.decode(x0, agent) for x0 in json['month']], dayOfWeek = [raritan.rpc.event.TimerEventManager.Range.decode(x0, agent) for x0 in json['dayOfWeek']], ) return obj def encode(self): json = {} json['enabled'] = self.enabled json['minute'] = [raritan.rpc.event.TimerEventManager.Range.encode(x0) for x0 in self.minute] json['hour'] = [raritan.rpc.event.TimerEventManager.Range.encode(x0) for x0 in self.hour] json['dayOfMonth'] = [raritan.rpc.event.TimerEventManager.Range.encode(x0) for x0 in self.dayOfMonth] json['month'] = [raritan.rpc.event.TimerEventManager.Range.encode(x0) for x0 in self.month] json['dayOfWeek'] = [raritan.rpc.event.TimerEventManager.Range.encode(x0) for x0 in self.dayOfWeek] return json # structure class TimerEvent(Structure): idlType = "event.TimerEventManager_2_0_0.TimerEvent:1.0.0" elements = ["eventId", "executionTime"] def __init__(self, eventId, executionTime): for x0 in eventId: typecheck.is_string(x0, AssertionError) typecheck.is_struct(executionTime, raritan.rpc.event.TimerEventManager.Schedule, AssertionError) self.eventId = eventId self.executionTime = executionTime @classmethod def decode(cls, json, agent): obj = cls( eventId = [x0 for x0 in json['eventId']], executionTime = raritan.rpc.event.TimerEventManager.Schedule.decode(json['executionTime'], agent), ) return obj def encode(self): json = {} json['eventId'] = [x0 for x0 in self.eventId] json['executionTime'] = raritan.rpc.event.TimerEventManager.Schedule.encode(self.executionTime) return json class _addTimerEvent(Interface.Method): name = 'addTimerEvent' @staticmethod def encode(schedule): typecheck.is_struct(schedule, raritan.rpc.event.TimerEventManager.Schedule, AssertionError) args = {} args['schedule'] = raritan.rpc.event.TimerEventManager.Schedule.encode(schedule) return args @staticmethod def decode(rsp, agent): _ret_ = rsp['_ret_'] eventId = [x0 for x0 in rsp['eventId']] typecheck.is_int(_ret_, DecodeException) for x0 in eventId: typecheck.is_string(x0, DecodeException) return (_ret_, eventId) class _modifyTimerEvent(Interface.Method): name = 'modifyTimerEvent' @staticmethod def encode(eventId, schedule): for x0 in eventId: typecheck.is_string(x0, AssertionError) typecheck.is_struct(schedule, raritan.rpc.event.TimerEventManager.Schedule, AssertionError) args = {} args['eventId'] = [x0 for x0 in eventId] args['schedule'] = raritan.rpc.event.TimerEventManager.Schedule.encode(schedule) return args @staticmethod def decode(rsp, agent): _ret_ = rsp['_ret_'] typecheck.is_int(_ret_, DecodeException) return _ret_ class _deleteTimerEvent(Interface.Method): name = 'deleteTimerEvent' @staticmethod def encode(eventId): for x0 in eventId: typecheck.is_string(x0, AssertionError) args = {} args['eventId'] = [x0 for x0 in eventId] return args @staticmethod def decode(rsp, agent): _ret_ = rsp['_ret_'] typecheck.is_int(_ret_, DecodeException) return _ret_ class _listTimerEvents(Interface.Method): name = 'listTimerEvents' @staticmethod def encode(): args = {} return args @staticmethod def decode(rsp, agent): _ret_ = [raritan.rpc.event.TimerEventManager.TimerEvent.decode(x0, agent) for x0 in rsp['_ret_']] for x0 in _ret_: typecheck.is_struct(x0, raritan.rpc.event.TimerEventManager.TimerEvent, DecodeException) return _ret_ def __init__(self, target, agent): super(TimerEventManager, self).__init__(target, agent) self.addTimerEvent = TimerEventManager._addTimerEvent(self) self.modifyTimerEvent = TimerEventManager._modifyTimerEvent(self) self.deleteTimerEvent = TimerEventManager._deleteTimerEvent(self) self.listTimerEvents = TimerEventManager._listTimerEvents(self) # # Section generated by IdlC from "EventService.idl" # import raritan.rpc from raritan.rpc import Interface, Structure, ValueObject, Enumeration, typecheck, DecodeException import raritan.rpc.event import raritan.rpc.idl # interface class Consumer(Interface): idlType = "event.Consumer:1.0.0" class _pushEvents(Interface.Method): name = 'pushEvents' @staticmethod def encode(events): for x0 in events: typecheck.is_valobj(x0, raritan.rpc.idl.Event, AssertionError) args = {} args['events'] = [ValueObject.encode(x0) for x0 in events] return args @staticmethod def decode(rsp, agent): return None def __init__(self, target, agent): super(Consumer, self).__init__(target, agent) self.pushEvents = Consumer._pushEvents(self) # interface class Channel(Interface): idlType = "event.Channel:1.0.1" class _demandEventType(Interface.Method): name = 'demandEventType' @staticmethod def encode(type): typecheck.is_typeinfo(type, AssertionError) args = {} args['type'] = raritan.rpc.TypeInfo.encode(type) return args @staticmethod def decode(rsp, agent): return None class _cancelEventType(Interface.Method): name = 'cancelEventType' @staticmethod def encode(type): typecheck.is_typeinfo(type, AssertionError) args = {} args['type'] = raritan.rpc.TypeInfo.encode(type) return args @staticmethod def decode(rsp, agent): return None class _demandEventTypes(Interface.Method): name = 'demandEventTypes' @staticmethod def encode(types): for x0 in types: typecheck.is_typeinfo(x0, AssertionError) args = {} args['types'] = [raritan.rpc.TypeInfo.encode(x0) for x0 in types] return args @staticmethod def decode(rsp, agent): return None class _cancelEventTypes(Interface.Method): name = 'cancelEventTypes' @staticmethod def encode(types): for x0 in types: typecheck.is_typeinfo(x0, AssertionError) args = {} args['types'] = [raritan.rpc.TypeInfo.encode(x0) for x0 in types] return args @staticmethod def decode(rsp, agent): return None class _demandEvent(Interface.Method): name = 'demandEvent' @staticmethod def encode(type, src): typecheck.is_typeinfo(type, AssertionError) typecheck.is_remote_obj(src, AssertionError) args = {} args['type'] = raritan.rpc.TypeInfo.encode(type) args['src'] = Interface.encode(src) return args @staticmethod def decode(rsp, agent): return None class _cancelEvent(Interface.Method): name = 'cancelEvent' @staticmethod def encode(type, src): typecheck.is_typeinfo(type, AssertionError) typecheck.is_remote_obj(src, AssertionError) args = {} args['type'] = raritan.rpc.TypeInfo.encode(type) args['src'] = Interface.encode(src) return args @staticmethod def decode(rsp, agent): return None # structure class EventSelect(Structure): idlType = "event.Channel_1_0_1.EventSelect:1.0.0" elements = ["type", "src"] def __init__(self, type, src): typecheck.is_typeinfo(type, AssertionError) typecheck.is_remote_obj(src, AssertionError) self.type = type self.src = src @classmethod def decode(cls, json, agent): obj = cls( type = raritan.rpc.TypeInfo.decode(json['type']), src = Interface.decode(json['src'], agent), ) return obj def encode(self): json = {} json['type'] = raritan.rpc.TypeInfo.encode(self.type) json['src'] = Interface.encode(self.src) return json class _demandEvents(Interface.Method): name = 'demandEvents' @staticmethod def encode(events): for x0 in events: typecheck.is_struct(x0, raritan.rpc.event.Channel.EventSelect, AssertionError) args = {} args['events'] = [raritan.rpc.event.Channel.EventSelect.encode(x0) for x0 in events] return args @staticmethod def decode(rsp, agent): return None class _cancelEvents(Interface.Method): name = 'cancelEvents' @staticmethod def encode(events): for x0 in events: typecheck.is_struct(x0, raritan.rpc.event.Channel.EventSelect, AssertionError) args = {} args['events'] = [raritan.rpc.event.Channel.EventSelect.encode(x0) for x0 in events] return args @staticmethod def decode(rsp, agent): return None class _subscribe(Interface.Method): name = 'subscribe' @staticmethod def encode(consumer): typecheck.is_interface(consumer, raritan.rpc.event.Consumer, AssertionError) args = {} args['consumer'] = Interface.encode(consumer) return args @staticmethod def decode(rsp, agent): return None class _unsubscribe(Interface.Method): name = 'unsubscribe' @staticmethod def encode(consumer): typecheck.is_interface(consumer, raritan.rpc.event.Consumer, AssertionError) args = {} args['consumer'] = Interface.encode(consumer) return args @staticmethod def decode(rsp, agent): _ret_ = rsp['_ret_'] typecheck.is_int(_ret_, DecodeException) return _ret_ class _pollEvents(Interface.Method): name = 'pollEvents' @staticmethod def encode(): args = {} return args @staticmethod def decode(rsp, agent): _ret_ = rsp['_ret_'] events = [ValueObject.decode(x0, agent) for x0 in rsp['events']] typecheck.is_bool(_ret_, DecodeException) for x0 in events: typecheck.is_valobj(x0, raritan.rpc.idl.Event, DecodeException) return (_ret_, events) class _pollEventsNb(Interface.Method): name = 'pollEventsNb' @staticmethod def encode(): args = {} return args @staticmethod def decode(rsp, agent): _ret_ = rsp['_ret_'] events = [ValueObject.decode(x0, agent) for x0 in rsp['events']] typecheck.is_bool(_ret_, DecodeException) for x0 in events: typecheck.is_valobj(x0, raritan.rpc.idl.Event, DecodeException) return (_ret_, events) def __init__(self, target, agent): super(Channel, self).__init__(target, agent) self.demandEventType = Channel._demandEventType(self) self.cancelEventType = Channel._cancelEventType(self) self.demandEventTypes = Channel._demandEventTypes(self) self.cancelEventTypes = Channel._cancelEventTypes(self) self.demandEvent = Channel._demandEvent(self) self.cancelEvent = Channel._cancelEvent(self) self.demandEvents = Channel._demandEvents(self) self.cancelEvents = Channel._cancelEvents(self) self.subscribe = Channel._subscribe(self) self.unsubscribe = Channel._unsubscribe(self) self.pollEvents = Channel._pollEvents(self) self.pollEventsNb = Channel._pollEventsNb(self) # interface class Service(Interface): idlType = "event.Service:1.0.1" INVALID_CHANNEL = 1 class _createChannel(Interface.Method): name = 'createChannel' @staticmethod def encode(): args = {} return args @staticmethod def decode(rsp, agent): _ret_ = Interface.decode(rsp['_ret_'], agent) typecheck.is_interface(_ret_, raritan.rpc.event.Channel, DecodeException) return _ret_ class _destroyChannel(Interface.Method): name = 'destroyChannel' @staticmethod def encode(channel): typecheck.is_interface(channel, raritan.rpc.event.Channel, AssertionError) args = {} args['channel'] = Interface.encode(channel) return args @staticmethod def decode(rsp, agent): _ret_ = rsp['_ret_'] typecheck.is_int(_ret_, DecodeException) return _ret_ class _pushEvent(Interface.Method): name = 'pushEvent' @staticmethod def encode(event): typecheck.is_valobj(event, raritan.rpc.idl.Event, AssertionError) args = {} args['event'] = ValueObject.encode(event) return args @staticmethod def decode(rsp, agent): return None class _pushEvents(Interface.Method): name = 'pushEvents' @staticmethod def encode(events): for x0 in events: typecheck.is_valobj(x0, raritan.rpc.idl.Event, AssertionError) args = {} args['events'] = [ValueObject.encode(x0) for x0 in events] return args @staticmethod def decode(rsp, agent): return None def __init__(self, target, agent): super(Service, self).__init__(target, agent) self.createChannel = Service._createChannel(self) self.destroyChannel = Service._destroyChannel(self) self.pushEvent = Service._pushEvent(self) self.pushEvents = Service._pushEvents(self) # # Section generated by IdlC from "EventEngine.idl" # import raritan.rpc from raritan.rpc import Interface, Structure, ValueObject, Enumeration, typecheck, DecodeException import raritan.rpc.event # structure class KeyValue(Structure): idlType = "event.KeyValue:1.0.0" elements = ["key", "value"] def __init__(self, key, value): typecheck.is_string(key, AssertionError) typecheck.is_string(value, AssertionError) self.key = key self.value = value @classmethod def decode(cls, json, agent): obj = cls( key = json['key'], value = json['value'], ) return obj def encode(self): json = {} json['key'] = self.key json['value'] = self.value return json # structure class Event(Structure): idlType = "event.Event:1.0.0" elements = ["type", "id", "asserted", "timeStamp", "context"] def __init__(self, type, id, asserted, timeStamp, context): typecheck.is_enum(type, raritan.rpc.event.Event.Type, AssertionError) for x0 in id: typecheck.is_string(x0, AssertionError) typecheck.is_bool(asserted, AssertionError) typecheck.is_time(timeStamp, AssertionError) for x0 in context: typecheck.is_struct(x0, raritan.rpc.event.KeyValue, AssertionError) self.type = type self.id = id self.asserted = asserted self.timeStamp = timeStamp self.context = context @classmethod def decode(cls, json, agent): obj = cls( type = raritan.rpc.event.Event.Type.decode(json['type']), id = [x0 for x0 in json['id']], asserted = json['asserted'], timeStamp = raritan.rpc.Time.decode(json['timeStamp']), context = [raritan.rpc.event.KeyValue.decode(x0, agent) for x0 in json['context']], ) return obj def encode(self): json = {} json['type'] = raritan.rpc.event.Event.Type.encode(self.type) json['id'] = [x0 for x0 in self.id] json['asserted'] = self.asserted json['timeStamp'] = raritan.rpc.Time.encode(self.timeStamp) json['context'] = [raritan.rpc.event.KeyValue.encode(x0) for x0 in self.context] return json # enumeration class Type(Enumeration): idlType = "event.Event.Type:1.0.0" values = ["STATE", "TRIGGER"] Type.STATE = Type(0) Type.TRIGGER = Type(1) # interface class Engine(Interface): idlType = "event.Engine:1.0.1" # structure class EventDesc(Structure): idlType = "event.Engine_1_0_1.EventDesc:1.0.0" elements = ["eventDescType", "eventType", "dynNodeContext", "idComp", "name", "entries"] def __init__(self, eventDescType, eventType, dynNodeContext, idComp, name, entries): typecheck.is_enum(eventDescType, raritan.rpc.event.Engine.EventDesc.Type, AssertionError) typecheck.is_enum(eventType, raritan.rpc.event.Event.Type, AssertionError) typecheck.is_string(dynNodeContext, AssertionError) typecheck.is_string(idComp, AssertionError) typecheck.is_string(name, AssertionError) for x0 in entries: typecheck.is_struct(x0, raritan.rpc.event.Engine.EventDesc, AssertionError) self.eventDescType = eventDescType self.eventType = eventType self.dynNodeContext = dynNodeContext self.idComp = idComp self.name = name self.entries = entries @classmethod def decode(cls, json, agent): obj = cls( eventDescType = raritan.rpc.event.Engine.EventDesc.Type.decode(json['eventDescType']), eventType = raritan.rpc.event.Event.Type.decode(json['eventType']), dynNodeContext = json['dynNodeContext'], idComp = json['idComp'], name = json['name'], entries = [raritan.rpc.event.Engine.EventDesc.decode(x0, agent) for x0 in json['entries']], ) return obj def encode(self): json = {} json['eventDescType'] = raritan.rpc.event.Engine.EventDesc.Type.encode(self.eventDescType) json['eventType'] = raritan.rpc.event.Event.Type.encode(self.eventType) json['dynNodeContext'] = self.dynNodeContext json['idComp'] = self.idComp json['name'] = self.name json['entries'] = [raritan.rpc.event.Engine.EventDesc.encode(x0) for x0 in self.entries] return json # enumeration class Type(Enumeration): idlType = "event.Engine_1_0_1.EventDesc.Type:1.0.0" values = ["NODE", "DYN_NODE", "LEAF"] Type.NODE = Type(0) Type.DYN_NODE = Type(1) Type.LEAF = Type(2) class _listEventDescs(Interface.Method): name = 'listEventDescs' @staticmethod def encode(eventIdPrefix): for x0 in eventIdPrefix: typecheck.is_string(x0, AssertionError) args = {} args['eventIdPrefix'] = [x0 for x0 in eventIdPrefix] return args @staticmethod def decode(rsp, agent): _ret_ = rsp['_ret_'] eventDescs = [raritan.rpc.event.Engine.EventDesc.decode(x0, agent) for x0 in rsp['eventDescs']] typecheck.is_int(_ret_, DecodeException) for x0 in eventDescs: typecheck.is_struct(x0, raritan.rpc.event.Engine.EventDesc, DecodeException) return (_ret_, eventDescs) class _listActionTypes(Interface.Method): name = 'listActionTypes' @staticmethod def encode(): args = {} return args @staticmethod def decode(rsp, agent): _ret_ = [x0 for x0 in rsp['_ret_']] for x0 in _ret_: typecheck.is_string(x0, DecodeException) return _ret_ # structure class Action(Structure): idlType = "event.Engine_1_0_1.Action:1.0.0" elements = ["id", "name", "isSystem", "type", "arguments"] def __init__(self, id, name, isSystem, type, arguments): typecheck.is_string(id, AssertionError) typecheck.is_string(name, AssertionError) typecheck.is_bool(isSystem, AssertionError) typecheck.is_string(type, AssertionError) for x0 in arguments: typecheck.is_struct(x0, raritan.rpc.event.KeyValue, AssertionError) self.id = id self.name = name self.isSystem = isSystem self.type = type self.arguments = arguments @classmethod def decode(cls, json, agent): obj = cls( id = json['id'], name = json['name'], isSystem = json['isSystem'], type = json['type'], arguments = [raritan.rpc.event.KeyValue.decode(x0, agent) for x0 in json['arguments']], ) return obj def encode(self): json = {} json['id'] = self.id json['name'] = self.name json['isSystem'] = self.isSystem json['type'] = self.type json['arguments'] = [raritan.rpc.event.KeyValue.encode(x0) for x0 in self.arguments] return json class _addAction(Interface.Method): name = 'addAction' @staticmethod def encode(action): typecheck.is_struct(action, raritan.rpc.event.Engine.Action, AssertionError) args = {} args['action'] = raritan.rpc.event.Engine.Action.encode(action) return args @staticmethod def decode(rsp, agent): _ret_ = rsp['_ret_'] actionId = rsp['actionId'] typecheck.is_int(_ret_, DecodeException) typecheck.is_string(actionId, DecodeException) return (_ret_, actionId) class _modifyAction(Interface.Method): name = 'modifyAction' @staticmethod def encode(action): typecheck.is_struct(action, raritan.rpc.event.Engine.Action, AssertionError) args = {} args['action'] = raritan.rpc.event.Engine.Action.encode(action) return args @staticmethod def decode(rsp, agent): _ret_ = rsp['_ret_'] typecheck.is_int(_ret_, DecodeException) return _ret_ class _deleteAction(Interface.Method): name = 'deleteAction' @staticmethod def encode(actionId): typecheck.is_string(actionId, AssertionError) args = {} args['actionId'] = actionId return args @staticmethod def decode(rsp, agent): _ret_ = rsp['_ret_'] typecheck.is_int(_ret_, DecodeException) return _ret_ class _listActions(Interface.Method): name = 'listActions' @staticmethod def encode(): args = {} return args @staticmethod def decode(rsp, agent): _ret_ = [raritan.rpc.event.Engine.Action.decode(x0, agent) for x0 in rsp['_ret_']] for x0 in _ret_: typecheck.is_struct(x0, raritan.rpc.event.Engine.Action, DecodeException) return _ret_ class _triggerAction(Interface.Method): name = 'triggerAction' @staticmethod def encode(actionId, context): typecheck.is_string(actionId, AssertionError) for x0 in context: typecheck.is_struct(x0, raritan.rpc.event.KeyValue, AssertionError) args = {} args['actionId'] = actionId args['context'] = [raritan.rpc.event.KeyValue.encode(x0) for x0 in context] return args @staticmethod def decode(rsp, agent): _ret_ = rsp['_ret_'] errMsg = rsp['errMsg'] typecheck.is_int(_ret_, DecodeException) typecheck.is_string(errMsg, DecodeException) return (_ret_, errMsg) class _testAction(Interface.Method): name = 'testAction' @staticmethod def encode(action, context): typecheck.is_struct(action, raritan.rpc.event.Engine.Action, AssertionError) for x0 in context: typecheck.is_struct(x0, raritan.rpc.event.KeyValue, AssertionError) args = {} args['action'] = raritan.rpc.event.Engine.Action.encode(action) args['context'] = [raritan.rpc.event.KeyValue.encode(x0) for x0 in context] return args @staticmethod def decode(rsp, agent): _ret_ = rsp['_ret_'] errMsg = rsp['errMsg'] typecheck.is_int(_ret_, DecodeException) typecheck.is_string(errMsg, DecodeException) return (_ret_, errMsg) # structure class Condition(Structure): idlType = "event.Engine_1_0_1.Condition:1.0.0" elements = ["negate", "operation", "matchType", "eventId", "conditions"] def __init__(self, negate, operation, matchType, eventId, conditions): typecheck.is_bool(negate, AssertionError) typecheck.is_enum(operation, raritan.rpc.event.Engine.Condition.Op, AssertionError) typecheck.is_enum(matchType, raritan.rpc.event.Engine.Condition.MatchType, AssertionError) for x0 in eventId: typecheck.is_string(x0, AssertionError) for x0 in conditions: typecheck.is_struct(x0, raritan.rpc.event.Engine.Condition, AssertionError) self.negate = negate self.operation = operation self.matchType = matchType self.eventId = eventId self.conditions = conditions @classmethod def decode(cls, json, agent): obj = cls( negate = json['negate'], operation = raritan.rpc.event.Engine.Condition.Op.decode(json['operation']), matchType = raritan.rpc.event.Engine.Condition.MatchType.decode(json['matchType']), eventId = [x0 for x0 in json['eventId']], conditions = [raritan.rpc.event.Engine.Condition.decode(x0, agent) for x0 in json['conditions']], ) return obj def encode(self): json = {} json['negate'] = self.negate json['operation'] = raritan.rpc.event.Engine.Condition.Op.encode(self.operation) json['matchType'] = raritan.rpc.event.Engine.Condition.MatchType.encode(self.matchType) json['eventId'] = [x0 for x0 in self.eventId] json['conditions'] = [raritan.rpc.event.Engine.Condition.encode(x0) for x0 in self.conditions] return json # enumeration class Op(Enumeration): idlType = "event.Engine_1_0_1.Condition.Op:1.0.0" values = ["AND", "OR", "XOR"] Op.AND = Op(0) Op.OR = Op(1) Op.XOR = Op(2) # enumeration class MatchType(Enumeration): idlType = "event.Engine_1_0_1.Condition.MatchType:1.0.0" values = ["ASSERTED", "DEASSERTED", "BOTH"] MatchType.ASSERTED = MatchType(0) MatchType.DEASSERTED = MatchType(1) MatchType.BOTH = MatchType(2) # structure class Rule(Structure): idlType = "event.Engine_1_0_1.Rule:1.0.0" elements = ["id", "name", "isSystem", "isEnabled", "isAutoRearm", "hasMatched", "condition", "actionIds", "arguments"] def __init__(self, id, name, isSystem, isEnabled, isAutoRearm, hasMatched, condition, actionIds, arguments): typecheck.is_string(id, AssertionError) typecheck.is_string(name, AssertionError) typecheck.is_bool(isSystem, AssertionError) typecheck.is_bool(isEnabled, AssertionError) typecheck.is_bool(isAutoRearm, AssertionError) typecheck.is_bool(hasMatched, AssertionError) typecheck.is_struct(condition, raritan.rpc.event.Engine.Condition, AssertionError) for x0 in actionIds: typecheck.is_string(x0, AssertionError) for x0 in arguments: typecheck.is_struct(x0, raritan.rpc.event.KeyValue, AssertionError) self.id = id self.name = name self.isSystem = isSystem self.isEnabled = isEnabled self.isAutoRearm = isAutoRearm self.hasMatched = hasMatched self.condition = condition self.actionIds = actionIds self.arguments = arguments @classmethod def decode(cls, json, agent): obj = cls( id = json['id'], name = json['name'], isSystem = json['isSystem'], isEnabled = json['isEnabled'], isAutoRearm = json['isAutoRearm'], hasMatched = json['hasMatched'], condition = raritan.rpc.event.Engine.Condition.decode(json['condition'], agent), actionIds = [x0 for x0 in json['actionIds']], arguments = [raritan.rpc.event.KeyValue.decode(x0, agent) for x0 in json['arguments']], ) return obj def encode(self): json = {} json['id'] = self.id json['name'] = self.name json['isSystem'] = self.isSystem json['isEnabled'] = self.isEnabled json['isAutoRearm'] = self.isAutoRearm json['hasMatched'] = self.hasMatched json['condition'] = raritan.rpc.event.Engine.Condition.encode(self.condition) json['actionIds'] = [x0 for x0 in self.actionIds] json['arguments'] = [raritan.rpc.event.KeyValue.encode(x0) for x0 in self.arguments] return json class _addRule(Interface.Method): name = 'addRule' @staticmethod def encode(rule): typecheck.is_struct(rule, raritan.rpc.event.Engine.Rule, AssertionError) args = {} args['rule'] = raritan.rpc.event.Engine.Rule.encode(rule) return args @staticmethod def decode(rsp, agent): _ret_ = rsp['_ret_'] ruleId = rsp['ruleId'] typecheck.is_int(_ret_, DecodeException) typecheck.is_string(ruleId, DecodeException) return (_ret_, ruleId) class _modifyRule(Interface.Method): name = 'modifyRule' @staticmethod def encode(rule): typecheck.is_struct(rule, raritan.rpc.event.Engine.Rule, AssertionError) args = {} args['rule'] = raritan.rpc.event.Engine.Rule.encode(rule) return args @staticmethod def decode(rsp, agent): _ret_ = rsp['_ret_'] typecheck.is_int(_ret_, DecodeException) return _ret_ class _enableRule(Interface.Method): name = 'enableRule' @staticmethod def encode(ruleId): typecheck.is_string(ruleId, AssertionError) args = {} args['ruleId'] = ruleId return args @staticmethod def decode(rsp, agent): _ret_ = rsp['_ret_'] typecheck.is_int(_ret_, DecodeException) return _ret_ class _disableRule(Interface.Method): name = 'disableRule' @staticmethod def encode(ruleId): typecheck.is_string(ruleId, AssertionError) args = {} args['ruleId'] = ruleId return args @staticmethod def decode(rsp, agent): _ret_ = rsp['_ret_'] typecheck.is_int(_ret_, DecodeException) return _ret_ class _deleteRule(Interface.Method): name = 'deleteRule' @staticmethod def encode(ruleId): typecheck.is_string(ruleId, AssertionError) args = {} args['ruleId'] = ruleId return args @staticmethod def decode(rsp, agent): _ret_ = rsp['_ret_'] typecheck.is_int(_ret_, DecodeException) return _ret_ class _listRules(Interface.Method): name = 'listRules' @staticmethod def encode(): args = {} return args @staticmethod def decode(rsp, agent): _ret_ = [raritan.rpc.event.Engine.Rule.decode(x0, agent) for x0 in rsp['_ret_']] for x0 in _ret_: typecheck.is_struct(x0, raritan.rpc.event.Engine.Rule, DecodeException) return _ret_ class _deliverEvent(Interface.Method): name = 'deliverEvent' @staticmethod def encode(event): typecheck.is_struct(event, raritan.rpc.event.Event, AssertionError) args = {} args['event'] = raritan.rpc.event.Event.encode(event) return args @staticmethod def decode(rsp, agent): _ret_ = rsp['_ret_'] typecheck.is_int(_ret_, DecodeException) return _ret_ class _rearmRule(Interface.Method): name = 'rearmRule' @staticmethod def encode(ruleId): typecheck.is_string(ruleId, AssertionError) args = {} args['ruleId'] = ruleId return args @staticmethod def decode(rsp, agent): _ret_ = rsp['_ret_'] typecheck.is_int(_ret_, DecodeException) return _ret_ def __init__(self, target, agent): super(Engine, self).__init__(target, agent) self.listEventDescs = Engine._listEventDescs(self) self.listActionTypes = Engine._listActionTypes(self) self.addAction = Engine._addAction(self) self.modifyAction = Engine._modifyAction(self) self.deleteAction = Engine._deleteAction(self) self.listActions = Engine._listActions(self) self.triggerAction = Engine._triggerAction(self) self.testAction = Engine._testAction(self) self.addRule = Engine._addRule(self) self.modifyRule = Engine._modifyRule(self) self.enableRule = Engine._enableRule(self) self.disableRule = Engine._disableRule(self) self.deleteRule = Engine._deleteRule(self) self.listRules = Engine._listRules(self) self.deliverEvent = Engine._deliverEvent(self) self.rearmRule = Engine._rearmRule(self)