Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_integer_tag(self):
if _debug: TestInteger._debug("test_integer_tag")
tag = Tag(Tag.applicationTagClass, Tag.integerAppTag, 1, xtob('01'))
obj = Integer(tag)
assert obj.value == 1
tag = Tag(Tag.applicationTagClass, Tag.booleanAppTag, 0, xtob(''))
with self.assertRaises(InvalidTag):
Integer(tag)
tag = Tag(Tag.contextTagClass, 0, 1, xtob('ff'))
with self.assertRaises(InvalidTag):
Integer(tag)
tag = Tag(Tag.openingTagClass, 0)
with self.assertRaises(InvalidTag):
Integer(tag)
def IntegerTag(v):
"""Return an application encoded integer tag with the appropriate value.
"""
obj = Integer(v)
tag = Tag()
obj.encode(tag)
return tag
if _debug: TestConsoleCmd._debug(" - datatype: %r", datatype)
# change atomic values into something encodeable, null is a special case
if (value == 'null'):
value = Null()
elif issubclass(datatype, Atomic):
if datatype is Integer:
value = int(value)
elif datatype is Real:
value = float(value)
elif datatype is Unsigned:
value = int(value)
value = datatype(value)
elif issubclass(datatype, Array) and (indx is not None):
if indx == 0:
value = Integer(value)
elif issubclass(datatype.subtype, Atomic):
value = datatype.subtype(value)
elif not isinstance(value, datatype.subtype):
raise TypeError, "invalid result datatype, expecting %s" % (datatype.subtype.__name__,)
elif not isinstance(value, datatype):
raise TypeError, "invalid result datatype, expecting %s" % (datatype.__name__,)
if _debug: TestConsoleCmd._debug(" - encodeable value: %r %s", value, type(value))
# build a request
request = WritePropertyRequest(
objectIdentifier=(obj_type, obj_inst),
propertyIdentifier=prop_id
)
request.pduDestination = Address(addr)
# save the value
def integer_decode(tag):
"""Decode an integer application tag into an integer."""
if _debug: integer_decode._debug("integer_decode %r", tag)
obj = Integer(tag)
if _debug: integer_decode._debug(" - obj: %r", obj)
return obj
if value == "null":
value = Null()
elif issubclass(datatype, Atomic):
if datatype is Integer:
value = int(value)
elif datatype is Real:
value = float(value)
elif datatype is Unsigned:
value = int(value)
elif datatype is Enumerated:
value = int(value)
value = datatype(value)
elif issubclass(datatype, Array) and (indx is not None):
if indx == 0:
value = Integer(value)
elif issubclass(datatype.subtype, Atomic):
value = datatype.subtype(value)
elif not isinstance(value, datatype.subtype):
raise TypeError(
"invalid result datatype, expecting {}".format(
(datatype.subtype.__name__,)
)
)
elif not isinstance(value, datatype):
raise TypeError(
"invalid result datatype, expecting {}".format((datatype.__name__,))
)
self._log.debug("{:<20} {!r} {}".format("Encodeable value", value, type(value)))
_value = Any()
),
),
listOfTimeValues=[
TimeValue(time=(2,0,0,0), value=Integer(2)),
TimeValue(time=(9,0,0,0), value=Null()),
],
eventPriority=4,
),
SpecialEvent(
period=SpecialEventPeriod(
calendarEntry=CalendarEntry(
weekNDay=xtob("FF.FF.FF"),
),
),
listOfTimeValues=[
TimeValue(time=(1,0,0,0), value=Integer(1)),
],
eventPriority=5,
),
]),
scheduleDefault=Integer(0),
)
_log.debug(" - so: %r", so)
this_application.add_object(so)
schedule_objects.append(so)
# list of time values for every five minutes
ltv = []
for hr in range(24):
for mn in range(0, 60, 5):
ltv.append(TimeValue(time=(hr,mn,0,0), value=Integer(hr * 100 + mn)))
# being identical.
#
so = LocalScheduleObject(
objectIdentifier=('schedule', 1),
objectName='Schedule 1',
presentValue=Integer(8),
effectivePeriod=DateRange(
startDate=(0, 1, 1, 1),
endDate=(254, 12, 31, 2),
),
weeklySchedule=ArrayOf(DailySchedule)([
DailySchedule(
daySchedule=[
TimeValue(time=(8,0,0,0), value=Integer(8)),
TimeValue(time=(14,0,0,0), value=Null()),
TimeValue(time=(17,0,0,0), value=Integer(42)),
# TimeValue(time=(0,0,0,0), value=Null()),
]
),
] * 7),
scheduleDefault=Integer(0),
)
_log.debug(" - so: %r", so)
this_application.add_object(so)
schedule_objects.append(so)
#
# A special schedule when the Year 2000 problem was supposed to collapse
# systems, the panic clears ten minutes later when it didn't.
#
so = LocalScheduleObject(
objectIdentifier=('schedule', 2),
@RPC.export
def write_property(self, target_address, value, object_type, instance_number, property_name, priority=None, index=None):
"""Write to a property."""
request = WritePropertyRequest(
objectIdentifier=(object_type, instance_number),
propertyIdentifier=property_name)
datatype = get_datatype(object_type, property_name)
if (value == 'null'):
bac_value = Null()
elif issubclass(datatype, Atomic):
if datatype is Integer:
value = int(value)
elif datatype is Real:
value = float(value)
elif datatype is Unsigned:
value = int(value)
bac_value = datatype(value)
elif issubclass(datatype, Array) and (index is not None):
if index == 0:
bac_value = Integer(value)
elif issubclass(datatype.subtype, Atomic):
bac_value = datatype.subtype(value)
elif not isinstance(value, datatype.subtype):
raise TypeError("invalid result datatype, expecting %s" % (datatype.subtype.__name__,))
elif not isinstance(value, datatype):
raise TypeError("invalid result datatype, expecting %s" % (datatype.__name__,))