Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use set_cb method in fMBT
Best Python code snippet using fMBT_python
test_model.py
Source:test_model.py 
1from collections import OrderedDict2from copy import deepcopy3from datetime import datetime4from enum import Enum5from unittest.mock import Mock6import pytest7from pyzkaccess.device_data.model import Field, Model, models_registry8class EnumStub(Enum):9 val1 = 12310 val2 = 45611class ModelStub(Model):12 table_name = 'table1'13 incremented_field = Field(14 'IncField', int, lambda x: int(x) + 1, lambda x: x - 1, lambda x: x > 015 )16 append_foo_field = Field(17 'FooField', str, lambda x: x + 'Foo', lambda x: x[:-3], lambda x: len(x) > 018 )19class TestField:20 def test_init__should_set_properties(self):21 field_datatype, get_cb, set_cb, validation_cb = Mock(), Mock(), Mock(), Mock()22 obj = Field('my_name', field_datatype, get_cb, set_cb, validation_cb)23 assert obj._raw_name == 'my_name'24 assert obj._field_datatype is field_datatype25 assert obj._get_cb is get_cb26 assert obj._set_cb is set_cb27 assert obj._validation_cb is validation_cb28 def test_init__should_set_default_properties(self):29 obj = Field('my_name')30 assert obj._raw_name == 'my_name'31 assert obj._field_datatype == str32 assert obj._get_cb is None33 assert obj._set_cb is None34 assert obj._validation_cb is None35 def test_raw_name_prop__should_return_raw_name(self):36 obj = Field('my_name')37 assert obj.raw_name == 'my_name'38 def test_field_datatype_prop__should_return_field_datatype(self):39 obj = Field('my_name', int)40 assert obj.field_datatype == int41 def test_to_raw_value__on_all_defaults__should_return_the_same_string(self):42 obj = Field('my_name')43 res = obj.to_raw_value('value1')44 assert res == 'value1'45 @pytest.mark.parametrize('datatype,value', (46 (str, '123'),47 (int, 123),48 (datetime, datetime(2020, 12, 12, 12, 12, 12)),49 (tuple, (1, 2, 3))50 ))51 def test_to_raw_value__if_type_set__should_return_string_representation(self, datatype, value):52 obj = Field('my_name', datatype)53 res = obj.to_raw_value(value)54 assert type(res) == str and res == str(value)55 def test_to_raw_value__if_type_is_enum__should_return_its_value_string_representation(self):56 obj = Field('my_name', EnumStub)57 res = obj.to_raw_value(EnumStub.val2)58 assert res == '456'59 @pytest.mark.parametrize('datatype,value,set_cb,expect', (60 (str, '123', int, '123'), # str=>int=>str61 (int, 123, lambda x: x + 1, '124'), # int=>[increment]=>str62 (datetime, datetime(2020, 12, 12, 12, 12, 12), lambda x: x.day, '12'), # dtime=>[.day]=>str63 (tuple, ('1', '2', '3'), lambda x: ''.join(x), '123'), # tuple=>[join to string]=>str64 (Enum, EnumStub.val2, lambda x: x + 1, '457') # Enum=>[.value]=>[increment]=>str65 ))66 def test_to_raw_value__if_set_cb_set__should_use_its_value_and_cast_to_string(67 self, datatype, value, set_cb, expect68 ):69 get_cb = Mock()70 obj = Field('my_name', datatype, get_cb, set_cb)71 res = obj.to_raw_value(value)72 assert res == expect and type(res) == type(expect)73 get_cb.assert_not_called()74 @pytest.mark.parametrize('datatype,value,set_cb,expect', (75 (str, '123', int, '123'), # str=>int=>str76 (int, 123, lambda x: x + 1, '124'), # int=>[increment]=>str77 (datetime, datetime(2020, 12, 12, 12, 12, 12), lambda x: x.day, '12'), # dtime=>[.day]=>str78 (tuple, ('1', '2', '3'), lambda x: ''.join(x), '123'), # tuple=>[join to string]=>str79 (Enum, EnumStub.val2, lambda x: x + 1, '457') # Enum=>[.value]=>[increment]=>str80 ))81 def test_to_raw_value__if_set_cb_and_validation_cb_passed__should_return_string_of_get_cb_value(82 self, datatype, value, set_cb, expect83 ):84 get_cb = Mock()85 validation_cb = Mock(return_value=True)86 obj = Field('my_name', datatype, get_cb, set_cb, validation_cb)87 res = obj.to_raw_value(value)88 assert res == expect and type(res) == type(expect)89 get_cb.assert_not_called()90 validation_cb.assert_called_once_with(value)91 @pytest.mark.parametrize('datatype,value', (92 (str, '123'),93 (int, 123),94 (datetime, datetime(2020, 12, 12, 12, 12, 12)),95 (tuple, (1, 2, 3)),96 (Enum, EnumStub.val2)97 ))98 def test_to_raw_value__if_set_cb_and_validation_cb_failed__should_raise_error(99 self, datatype, value100 ):101 get_cb = Mock()102 set_cb = Mock(return_value=555)103 validation_cb = Mock(return_value=False)104 obj = Field('my_name', datatype, get_cb, set_cb, validation_cb)105 with pytest.raises(ValueError):106 obj.to_raw_value(value)107 @pytest.mark.parametrize('datatype,value', (108 (str, 123),109 (str, None),110 (int, '123'),111 (datetime, 5),112 (tuple, datetime.now()),113 (Enum, 456)114 ))115 def test_to_raw_value__if_value_has_another_type__should_raise_error(self, datatype, value):116 get_cb = Mock()117 set_cb = Mock(return_value=555)118 validation_cb = Mock(return_value=False)119 obj = Field('my_name', datatype, get_cb, set_cb, validation_cb)120 with pytest.raises(TypeError):121 obj.to_raw_value(value)122 def test_to_field_value__on_all_defaults__should_return_the_same_string(self):123 obj = Field('my_name')124 res = obj.to_field_value('value1')125 assert res == 'value1'126 @pytest.mark.parametrize('datatype,value,expect', (127 (str, '123', '123'),128 (int, '123', 123),129 (EnumStub, 456, EnumStub.val2)130 ))131 def test_to_field_value__if_type_set__should_return_value_of_this_type(132 self, datatype, value, expect133 ):134 obj = Field('my_name', datatype)135 res = obj.to_field_value(value)136 assert res == expect and type(res) == type(expect)137 @pytest.mark.parametrize('datatype,value,get_cb,expect', (138 (139 datetime,140 '2020-12-13 14:15:16',141 lambda x: datetime.strptime(x, '%Y-%m-%d %H:%M:%S'),142 datetime(2020, 12, 13, 14, 15, 16)143 ), # str=>datetime144 (tuple, '123', lambda x: tuple(x), ('1', '2', '3')), # str=>tuple145 (EnumStub, '456', lambda x: EnumStub(int(x)), EnumStub.val2), # str=>Enum146 (bool, '1', lambda x: bool(int(x)), True), # str=>bool147 (bool, '0', lambda x: bool(int(x)), False) # str=>bool148 ))149 def test_to_field_value__if_get_cb_returns_value_of_datatype__should_convert_value(150 self, datatype, value, get_cb, expect151 ):152 set_cb = Mock()153 validation_cb = Mock()154 obj = Field('my_name', datatype, get_cb, set_cb, validation_cb)155 res = obj.to_field_value(value)156 assert res == expect and type(res) == type(expect)157 set_cb.assert_not_called()158 validation_cb.assert_not_called()159 @pytest.mark.parametrize('datatype,value', (160 (datetime, '2020-12-13 14:15:16'),161 (tuple, '123'),162 (EnumStub, '456'),163 (bool, '1'),164 (bool, '0')165 ))166 def test_to_field_value__if_get_cb_returns_none__should_return_none(self, datatype, value):167 set_cb = Mock()168 validation_cb = Mock()169 obj = Field('my_name', datatype, lambda _: None, set_cb, validation_cb)170 res = obj.to_field_value(value)171 assert res is None172 set_cb.assert_not_called()173 validation_cb.assert_not_called()174 @pytest.mark.parametrize('datatype,value,get_cb,expect', (175 (176 str,177 '2020-12-13 14:15:16',178 lambda x: datetime.strptime(x, '%Y-%m-%d %H:%M:%S'),179 '2020-12-13 14:15:16'180 ), # str=>datetime=>str181 (tuple, '123', lambda x: list(x), ('1', '2', '3')), # str=>list=>tuple182 (EnumStub, '456', int, EnumStub.val2), # str=>int=>Enum183 (bool, '1', int, True), # str=>int=>bool184 (bool, '0', int, False) # str=>int=>bool185 ))186 def test_to_field_value__if_get_cb_returns_value_not_of_datatype__should_also_cast_to_datatype(187 self, datatype, value, get_cb, expect188 ):189 set_cb = Mock()190 validation_cb = Mock()191 obj = Field('my_name', datatype, get_cb, set_cb, validation_cb)192 res = obj.to_field_value(value)193 assert res == expect and type(res) == type(expect)194 set_cb.assert_not_called()195 validation_cb.assert_not_called()196 def test_hash__should_be_hashable(self):197 obj = Field('my_name')198 assert hash(obj) == hash('my_name')199 def test_get_descriptor__should_correct_field_values(self):200 obj = ModelStub().with_raw_data({'IncField': '123', 'FooField': 'Magic'}, False)201 assert obj._dirty is False202 assert obj.append_foo_field == 'MagicFoo'203 assert obj.incremented_field == 124204 assert obj._dirty is False205 def test_get_descriptor__if_no_such_raw_field__should_return_none(self):206 obj = ModelStub().with_raw_data({'FooField': 'Magic'})207 assert obj.incremented_field is None208 def test_get_descriptor__if_class_instance__should_return_itself(self):209 assert isinstance(ModelStub.incremented_field, Field)210 def test_set_descriptor__should_set_raw_data(self):211 obj = ModelStub()212 obj.append_foo_field = "WowFoo"213 obj.incremented_field = 123214 assert obj._raw_data == {'IncField': '122', 'FooField': 'Wow'}215 def test_set_descriptor__should_set_dirty_flag(self):216 obj = ModelStub().with_raw_data({}, False)217 assert obj._dirty is False218 obj.incremented_field = 123219 assert obj._dirty is True220 def test_set_descriptor__if_none_is_given__should_delete_field_from_raw_data(self):221 obj = ModelStub().with_raw_data({'IncField': '123', 'FooField': 'Magic'})222 obj.incremented_field = None223 assert obj._raw_data == {'FooField': 'Magic'}224 def test_set_descriptor__should_consider_field_validation(self):225 obj = ModelStub()226 with pytest.raises(ValueError):227 obj.incremented_field = -1228 def test_set_descriptor__if_class_instance__should_do_nothing(self):229 class ModelStub2(Model):230 table_name = 'test'231 field1 = Field('field')232 ModelStub2.field1 = 'testvalue'233 assert object.__getattribute__(ModelStub2, 'field1') == 'testvalue'234 def test_del_descriptor__should_delete_field_from_raw_data(self):235 obj = ModelStub().with_raw_data({'IncField': '123', 'FooField': 'Magic'})236 del obj.incremented_field237 assert obj._raw_data == {'FooField': 'Magic'}238 def test_del_descriptor__should_set_dirty_flag(self):239 obj = ModelStub().with_raw_data({'IncField': '123', 'FooField': 'Magic'}, False)240 assert obj._dirty is False241 del obj.incremented_field242 assert obj._dirty is True243 def test_del_descriptor__if_class_instance__should_do_nothing(self):244 class ModelStub2(Model):245 table_name = 'test'246 field1 = Field('field')247 del ModelStub2.field1248 with pytest.raises(AttributeError):249 object.__getattribute__(ModelStub2, 'field1')250class TestModelMeta:251 @pytest.fixture252 def test_registry(self):253 orig_model_registry = deepcopy(models_registry)254 models_registry.clear()255 yield models_registry256 models_registry.clear()257 models_registry.update(orig_model_registry)258 def test_metaclass__should_add_class_to_registry(self, test_registry):259 class MyModel(Model):260 table_name = 'test'261 field1 = Field('FieldOne')262 assert test_registry == {'MyModel': MyModel}263 def test_metaclass__should_fill_fields_mapping_for_each_class(self):264 class MyModel(Model):265 table_name = 'test'266 field1 = Field('FieldOne')267 field2 = Field('FieldTwo')268 class MyModel2(Model):269 table_name = 'test'270 field3 = Field('FieldThree')271 field4 = Field('FieldFour')272 assert MyModel._fields_mapping == {'field1': 'FieldOne', 'field2': 'FieldTwo'}273 assert MyModel2._fields_mapping == {'field3': 'FieldThree', 'field4': 'FieldFour'}274 def test_metaclass__should_set_field_objects_doc_attribute(self):275 class MyModel(Model):276 table_name = 'test'277 field1 = Field('FieldOne')278 field2 = Field('FieldTwo', int)279 assert MyModel.field1.__doc__ == 'MyModel.field1'280 assert MyModel.field2.__doc__ == 'MyModel.field2'281 def test_metaclass__should_set_field_object_class_var_annotation(self):282 class MyModel(Model):283 table_name = 'test'284 field1 = Field('FieldOne')285 field2 = Field('FieldTwo', int)286 assert {'field1': str, 'field2': int}.items() <= MyModel.__annotations__.items()287class TestModel:288 def test_init__should_set_default_attributes(self):289 obj = ModelStub()290 assert obj._sdk is None291 assert obj._dirty is True292 assert obj._raw_data == {}293 def test_init__if_fields_has_passed__should_initialize_raw_data_only_with_given_fields(self):294 obj = ModelStub(incremented_field=123)295 assert obj._raw_data == {'IncField': '122'}296 def test_init__if_nones_has_passed_in_fields__should_ignore_them(self):297 obj = ModelStub(incremented_field=123, append_foo_field=None)298 assert obj._raw_data == {'IncField': '122'}299 def test_init__if_unknown_fields_has_passed__should_raise_error(self):300 with pytest.raises(TypeError):301 ModelStub(incremented_field=123, unknown_field=3)302 def test_init__should_consider_fields_validation(self):303 with pytest.raises(ValueError):304 ModelStub(incremented_field=-1)305 def test_dict__should_return_fields_value(self):306 obj = ModelStub().with_raw_data({'IncField': '123', 'FooField': 'Magic'})307 assert obj.dict == {'incremented_field': 124, 'append_foo_field': 'MagicFoo'}308 def test_dict__if_no_certain_field_in_raw_data__should_return_nones(self):309 obj = ModelStub().with_raw_data({'IncField': '123'})310 assert obj.dict == {'incremented_field': 124, 'append_foo_field': None}311 def test_raw_data__should_return_raw_data_appended_with_empty_string_for_absend_keys(self):312 obj = ModelStub().with_raw_data({'IncField': '123'})313 assert obj.raw_data == {'IncField': '123', 'FooField': ''}314 def test_fields_mapping__should_return_fields_mapping(self):315 assert ModelStub.fields_mapping() == {316 'incremented_field': 'IncField', 'append_foo_field': 'FooField'317 }318 assert ModelStub().fields_mapping() == {319 'incremented_field': 'IncField', 'append_foo_field': 'FooField'320 }321 def test_delete__should_delete_current_record_and_set_dirty_flag(322 self, generator_sends_collector323 ):324 items = []325 sdk = Mock()326 sdk.delete_device_data.side_effect = generator_sends_collector(items)327 obj = ModelStub().with_sdk(sdk).with_raw_data(328 {'IncField': '123', 'FooField': 'Magic'}, False329 )330 assert obj._dirty is False331 obj.delete()332 sdk.delete_device_data.assert_called_once_with('table1')333 assert items == [{'IncField': '123', 'FooField': 'Magic'}, None]334 assert obj._dirty is True335 def test_delete__if_manually_created_record__should_raise_error(self):336 obj = ModelStub(incremented_field=123, append_foo_field='MagicFoo')337 with pytest.raises(TypeError):338 obj.delete()339 def test_save__should_upsert_current_record_and_reset_dirty_flag(340 self, generator_sends_collector341 ):342 items = []343 sdk = Mock()344 sdk.set_device_data.side_effect = generator_sends_collector(items)345 obj = ModelStub().with_sdk(sdk).with_raw_data({'IncField': '123', 'FooField': 'Magic'})346 assert obj._dirty is True347 obj.save()348 sdk.set_device_data.assert_called_once_with('table1')349 assert items == [{'IncField': '123', 'FooField': 'Magic'}, None]350 assert obj._dirty is False351 def test_save__if_manually_created_record__should_raise_error(self):352 obj = ModelStub(incremented_field=123, append_foo_field='MagicFoo')353 with pytest.raises(TypeError):354 obj.save()355 @pytest.mark.parametrize('dirty_flag', (True, False))356 def test_with_raw_data__should_set_raw_data_and_dirty_flag(self, dirty_flag):357 obj = ModelStub().with_raw_data({'IncField': '123', 'FooField': 'Magic'}, dirty_flag)358 assert obj._raw_data == {'IncField': '123', 'FooField': 'Magic'}359 assert obj._dirty == dirty_flag360 def test_with_raw_data__should_return_self(self):361 obj = ModelStub()362 assert obj.with_raw_data({}) is obj363 def test_with_sdk__should_set_sdk(self):364 sdk = Mock()365 obj = ModelStub().with_raw_data({'A': 'val1'}, False).with_sdk(sdk)366 assert obj._sdk is sdk367 assert obj._raw_data == {'A': 'val1'}368 assert obj._dirty is False369 def test_with_sdk__should_return_self(self):370 obj = ModelStub()371 assert obj.with_sdk(Mock()) is obj372 def test_with_zk__should_set_sdk(self):373 zk = Mock()374 obj = ModelStub().with_raw_data({'A': 'val1'}, False).with_zk(zk)375 assert obj._sdk is zk.sdk376 assert obj._raw_data == {'A': 'val1'}377 assert obj._dirty is False378 def test_with_zk__should_return_self(self):379 obj = ModelStub()380 assert obj.with_zk(Mock()) is obj381 def test_eq_ne__if_raw_data_and_table_are_equal__should_return_true(self):382 obj1 = ModelStub(incremented_field=123, append_foo_field='MagicFoo')383 obj2 = ModelStub(incremented_field=123, append_foo_field='MagicFoo')384 assert obj1 == obj2385 assert not(obj1 != obj2)386 @pytest.mark.parametrize('table_name,kwargs', (387 ('table1', {'incremented_field': 1, 'append_foo_field': 'Magic'}),388 ('table2', {'incremented_field': 122, 'append_foo_field': 'Magic'}),389 ('table2', {'incremented_field': 1, 'append_foo_field': 'Magic'}),390 ))391 def test_eq_ne__if_raw_data_or_table_are_not_equal__should_return_false(392 self, table_name, kwargs393 ):394 obj1 = ModelStub(incremented_field=123, append_foo_field='MagicFoo')395 obj2 = ModelStub(**kwargs)396 obj2.table_name = table_name397 assert obj1 != obj2398 assert not(obj1 == obj2)399 def test_repr__should_return_data_table_name_and_fields_and_their_raw_values(self):400 raw_data = OrderedDict((('IncField', '123'), ('FooField', 'Magic')))401 obj = ModelStub().with_raw_data(raw_data, False)402 assert repr(obj) == 'ModelStub(append_foo_field=Magic, incremented_field=123)'403 def test_repr__if_dirty_flag_is_set__should_reflect_this_fact_in_string(self):404 raw_data = OrderedDict((('IncField', '123'), ('FooField', 'Magic')))405 obj = ModelStub().with_raw_data(raw_data, True)...
ccwi9p9215_gpio.py
Source:ccwi9p9215_gpio.py
1############################################################################2# #3# Copyright (c)2008, 2009, Digi International (Digi). All Rights Reserved. #4# #5# Permission to use, copy, modify, and distribute this software and its #6# documentation, without fee and without a signed licensing agreement, is #7# hereby granted, provided that the software is used on Digi products only #8# and that the software contain this copyright notice, and the following #9# two paragraphs appear in all copies, modifications, and distributions as #10# well. Contact Product Management, Digi International, Inc., 11001 Bren #11# Road East, Minnetonka, MN, +1 952-912-3444, for commercial licensing #12# opportunities for non-Digi products. #13# #14# DIGI SPECIFICALLY DISCLAIMS ANY WARRANTIES, INCLUDING, BUT NOT LIMITED #15# TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A #16# PARTICULAR PURPOSE. THE SOFTWARE AND ACCOMPANYING DOCUMENTATION, IF ANY, #17# PROVIDED HEREUNDER IS PROVIDED "AS IS" AND WITHOUT WARRANTY OF ANY KIND. #18# DIGI HAS NO OBLIGATION TO PROVIDE MAINTENANCE, SUPPORT, UPDATES, #19# ENHANCEMENTS, OR MODIFICATIONS. #20# #21# IN NO EVENT SHALL DIGI BE LIABLE TO ANY PARTY FOR DIRECT, INDIRECT, #22# SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES, INCLUDING LOST PROFITS, #23# ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS DOCUMENTATION, EVEN IF #24# DIGI HAS BEEN ADVISED OF THE POSSIBILITY OF SUCH DAMAGES. #25# #26############################################################################27# Imports28import digihw29from devices.device_base import DeviceBase30from settings.settings_base import SettingsBase, Setting31from channels.channel_source_device_property import *32from samples.sample import Sample33from common.shutdown import SHUTDOWN_WAIT34import threading35import time36#Main class37class ModuleGPIOs(DeviceBase, threading.Thread):38 """39 This class extends one of our base classes and is intended as an40 example of a concrete, example implementation, but it is not itself41 meant to be included as part of our developer API. Please consult the42 base class documentation for the API and the source code for this file43 for an example implementation.44 """45 46 #Class vars47 gpios = {}48 gpios_ind = {}49 setting_gpio=False50 for i in xrange (0,32):51 gpios["GPIO_"+str(i)]=0 52 for i in xrange (0,32):53 gpios_ind["GPIO_"+str(i)]=i54 55 def __init__(self, name, core_services):56 self.__name = name57 self.__core = core_services58 self.setting_gpio = False59 60 from core.tracing import get_tracer61 self.__tracer = get_tracer(name)62 63 ## Settings Table Definition:64 settings_list = [ 65 Setting(66 name='input_gpios', type=list, required=True),67 Setting(68 name='output_gpios', type=list, required=True),69 Setting( 70 name='update_rate', type=float, required=False, 71 default_value=0.1,verify_function=lambda x: x > 0.0)72 ]73 74 ## Declare the GPIOs channels75 property_list = [ 76 ChannelSourceDeviceProperty(name="GPIO_0",77 type=bool,initial=Sample(0, False), 78 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,79 options=DPROP_OPT_AUTOTIMESTAMP,80 set_cb=lambda sample: self.set_gpio(gpio="GPIO_0",81 sample=sample)),82 ChannelSourceDeviceProperty(name="GPIO_1",83 type=bool,initial=Sample(0, False), 84 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,85 options=DPROP_OPT_AUTOTIMESTAMP,86 set_cb=lambda sample: self.set_gpio(gpio="GPIO_1",87 sample=sample)),88 ChannelSourceDeviceProperty(name="GPIO_2",89 type=bool,initial=Sample(0, False), 90 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,91 options=DPROP_OPT_AUTOTIMESTAMP,92 set_cb=lambda sample: self.set_gpio(gpio="GPIO_2",93 sample=sample)),94 ChannelSourceDeviceProperty(name="GPIO_3",95 type=bool,initial=Sample(0, False), 96 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,97 options=DPROP_OPT_AUTOTIMESTAMP,98 set_cb=lambda sample: self.set_gpio(gpio="GPIO_3",99 sample=sample)),100 ChannelSourceDeviceProperty(name="GPIO_4",101 type=bool,initial=Sample(0, False), 102 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,103 options=DPROP_OPT_AUTOTIMESTAMP,104 set_cb=lambda sample: self.set_gpio(gpio="GPIO_4",105 sample=sample)),106 ChannelSourceDeviceProperty(name="GPIO_5",107 type=bool,initial=Sample(0, False), 108 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,109 options=DPROP_OPT_AUTOTIMESTAMP,110 set_cb=lambda sample: self.set_gpio(gpio="GPIO_5",111 sample=sample)),112 ChannelSourceDeviceProperty(name="GPIO_6",113 type=bool,initial=Sample(0, False), 114 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,115 options=DPROP_OPT_AUTOTIMESTAMP,116 set_cb=lambda sample: self.set_gpio(gpio="GPIO_6",117 sample=sample)),118 ChannelSourceDeviceProperty(name="GPIO_7",119 type=bool,initial=Sample(0, False), 120 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,121 options=DPROP_OPT_AUTOTIMESTAMP,122 set_cb=lambda sample: self.set_gpio(gpio="GPIO_7",123 sample=sample)),124 ChannelSourceDeviceProperty(name="GPIO_8",125 type=bool,initial=Sample(0, False), 126 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,127 options=DPROP_OPT_AUTOTIMESTAMP,128 set_cb=lambda sample: self.set_gpio(gpio="GPIO_8",129 sample=sample)),130 ChannelSourceDeviceProperty(name="GPIO_9",131 type=bool,initial=Sample(0, False), 132 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,133 options=DPROP_OPT_AUTOTIMESTAMP,134 set_cb=lambda sample: self.set_gpio(gpio="GPIO_9",135 sample=sample)),136 ChannelSourceDeviceProperty(name="GPIO_10",137 type=bool,initial=Sample(0, False), 138 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,139 options=DPROP_OPT_AUTOTIMESTAMP,140 set_cb=lambda sample: self.set_gpio(gpio="GPIO_10",141 sample=sample)),142 ChannelSourceDeviceProperty(name="GPIO_11",143 type=bool,initial=Sample(0, False), 144 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,145 options=DPROP_OPT_AUTOTIMESTAMP,146 set_cb=lambda sample: self.set_gpio(gpio="GPIO_11",147 sample=sample)),148 ChannelSourceDeviceProperty(name="GPIO_12",149 type=bool,initial=Sample(0, False), 150 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,151 options=DPROP_OPT_AUTOTIMESTAMP,152 set_cb=lambda sample: self.set_gpio(gpio="GPIO_12",153 sample=sample)),154 ChannelSourceDeviceProperty(name="GPIO_13",155 type=bool,initial=Sample(0, False), 156 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,157 options=DPROP_OPT_AUTOTIMESTAMP,158 set_cb=lambda sample: self.set_gpio(gpio="GPIO_13",159 sample=sample)),160 ChannelSourceDeviceProperty(name="GPIO_14",161 type=bool,initial=Sample(0, False), 162 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,163 options=DPROP_OPT_AUTOTIMESTAMP,164 set_cb=lambda sample: self.set_gpio(gpio="GPIO_14",165 sample=sample)),166 ChannelSourceDeviceProperty(name="GPIO_15",167 type=bool,initial=Sample(0, False), 168 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,169 options=DPROP_OPT_AUTOTIMESTAMP,170 set_cb=lambda sample: self.set_gpio(gpio="GPIO_15",171 sample=sample)),172 ChannelSourceDeviceProperty(name="GPIO_16",173 type=bool,initial=Sample(0, False), 174 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,175 options=DPROP_OPT_AUTOTIMESTAMP,176 set_cb=lambda sample: self.set_gpio(gpio="GPIO_16",177 sample=sample)),178 ChannelSourceDeviceProperty(name="GPIO_17",179 type=bool,initial=Sample(0, False), 180 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,181 options=DPROP_OPT_AUTOTIMESTAMP,182 set_cb=lambda sample: self.set_gpio(gpio="GPIO_17",183 sample=sample)),184 ChannelSourceDeviceProperty(name="GPIO_18",185 type=bool,initial=Sample(0, False), 186 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,187 options=DPROP_OPT_AUTOTIMESTAMP,188 set_cb=lambda sample: self.set_gpio(gpio="GPIO_18",189 sample=sample)),190 ChannelSourceDeviceProperty(name="GPIO_19",191 type=bool,initial=Sample(0, False), 192 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,193 options=DPROP_OPT_AUTOTIMESTAMP,194 set_cb=lambda sample: self.set_gpio(gpio="GPIO_19",195 sample=sample)),196 ChannelSourceDeviceProperty(name="GPIO_20",197 type=bool,initial=Sample(0, False), 198 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,199 options=DPROP_OPT_AUTOTIMESTAMP,200 set_cb=lambda sample: self.set_gpio(gpio="GPIO_20",201 sample=sample)),202 ChannelSourceDeviceProperty(name="GPIO_21",203 type=bool,initial=Sample(0, False), 204 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,205 options=DPROP_OPT_AUTOTIMESTAMP,206 set_cb=lambda sample: self.set_gpio(gpio="GPIO_21",207 sample=sample)),208 ChannelSourceDeviceProperty(name="GPIO_22",209 type=bool,initial=Sample(0, False), 210 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,211 options=DPROP_OPT_AUTOTIMESTAMP,212 set_cb=lambda sample: self.set_gpio(gpio="GPIO_22",213 sample=sample)),214 ChannelSourceDeviceProperty(name="GPIO_23",215 type=bool,initial=Sample(0, False), 216 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,217 options=DPROP_OPT_AUTOTIMESTAMP,218 set_cb=lambda sample: self.set_gpio(gpio="GPIO_23",219 sample=sample)),220 ChannelSourceDeviceProperty(name="GPIO_24",221 type=bool,initial=Sample(0, False), 222 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,223 options=DPROP_OPT_AUTOTIMESTAMP,224 set_cb=lambda sample: self.set_gpio(gpio="GPIO_24",225 sample=sample)),226 ChannelSourceDeviceProperty(name="GPIO_25",227 type=bool,initial=Sample(0, False), 228 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,229 options=DPROP_OPT_AUTOTIMESTAMP,230 set_cb=lambda sample: self.set_gpio(gpio="GPIO_25",231 sample=sample)),232 ChannelSourceDeviceProperty(name="GPIO_26",233 type=bool,initial=Sample(0, False), 234 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,235 options=DPROP_OPT_AUTOTIMESTAMP,236 set_cb=lambda sample: self.set_gpio(gpio="GPIO_26",237 sample=sample)),238 ChannelSourceDeviceProperty(name="GPIO_27",239 type=bool,initial=Sample(0, False), 240 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,241 options=DPROP_OPT_AUTOTIMESTAMP,242 set_cb=lambda sample: self.set_gpio(gpio="GPIO_27",243 sample=sample)),244 ChannelSourceDeviceProperty(name="GPIO_28",245 type=bool,initial=Sample(0, False), 246 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,247 options=DPROP_OPT_AUTOTIMESTAMP,248 set_cb=lambda sample: self.set_gpio(gpio="GPIO_28",249 sample=sample)),250 ChannelSourceDeviceProperty(name="GPIO_29",251 type=bool,initial=Sample(0, False), 252 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,253 options=DPROP_OPT_AUTOTIMESTAMP,254 set_cb=lambda sample: self.set_gpio(gpio="GPIO_29",255 sample=sample)),256 ChannelSourceDeviceProperty(name="GPIO_30",257 type=bool,initial=Sample(0, False), 258 perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,259 options=DPROP_OPT_AUTOTIMESTAMP,260 set_cb=lambda sample: self.set_gpio(gpio="GPIO_30",261 sample=sample))262 ]263 264 ## Initialize the DeviceBase interface:265 DeviceBase.__init__(self, self.__name, self.__core,266 settings_list, property_list)267 ## Thread initialization:268 self.__stopevent = threading.Event()269 threading.Thread.__init__(self, name=name)270 threading.Thread.setDaemon(self, True)271 ## Functions which must be implemented to conform to the DeviceBase272 ## interface:273 def apply_settings(self):274 SettingsBase.merge_settings(self)275 accepted, rejected, not_found = SettingsBase.verify_settings(self)276 if len(rejected) or len(not_found):277 self.__tracer.error("Settings rejected/not found: %s %s", 278 rejected, not_found)279 if (('update_rate' in accepted) and 280 (accepted['update_rate'] > SHUTDOWN_WAIT)):281 self.__tracer.warning("Long update_rate setting may " + 282 "interfere with shutdown of Dia")283 284 SettingsBase.commit_settings(self, accepted)285 return (accepted, rejected, not_found)286 def start(self):287 threading.Thread.start(self)288 return True289 def stop(self):290 self.__stopevent.set()291 return True 292 # Threading related functions:293 def run(self):294 295 #Get the device properties296 time_sleep = SettingsBase.get_setting(self,"update_rate")297 self.input_gpios = SettingsBase.get_setting(self,"input_gpios")298 self.output_gpios = SettingsBase.get_setting(self,"output_gpios")299 300 #Call the GPIOs initializer method301 self.initialize_gpios()302 303 #Start the refresh thread304 while 1:305 if self.__stopevent.isSet():306 self.__stopevent.clear()307 break308 try:309 while self.setting_gpio:310 pass311 self.get_GPIOs()312 except Exception, e:313 self.__tracer.error("Unable to update values: %s", str(e))314 315 #Sleep the time configured in settings (in seconds)316 time.sleep(time_sleep)317 def set_gpio(self,gpio,sample):318 319 self.setting_gpio = True320 #Update the channel and the module GPIO321 if sample.value==True: 322 self.property_set(gpio,Sample(time.time(), True))323 digihw.gpio_set_value(self.gpios_ind[gpio], 1)324 else:325 self.property_set(gpio,Sample(time.time(), False))326 digihw.gpio_set_value(self.gpios_ind[gpio], 0)327 self.setting_gpio = False328 def get_GPIOs(self):329 330 for gpio in self.input_gpios:331 val = digihw.gpio_get_value(gpio)332 #If the GPIO value has changed, update its channel333 if self.gpios["GPIO_"+str(gpio)]!=val:334 self.gpios["GPIO_"+str(gpio)]=val 335 if val==0 :336 self.property_set("GPIO_"+str(gpio),337 Sample(time.time(), False))338 else:339 self.property_set("GPIO_"+str(gpio),340 Sample(time.time(), True))341 342 def initialize_gpios(self):343 344 for gpio in self.output_gpios:345 digihw.gpio_set_value(gpio,0)346 for gpio in self.input_gpios:347 digihw.gpio_set_input(gpio)348 ...
task.py
Source:task.py
...3class Task(object):4 lock = _thread.allocate_lock()5 def __init__(self, event=(lambda args: print('task running')), args=None):6 self.alive = False7 self.set_cb(event, args)8 def set_cb(self, event, args):9 self.event = event10 self.args = args11 def run(self):12 if Task.lock.acquire():13 while self.alive:14 self.event(self.args)15 Task.lock.release()16 _thread.exit()17 def stop(self):18 if self.alive is True:19 self.alive = False20 if Task.lock.acquire():21 # print("stop")22 Task.lock.release()23 def start(self, size=2048):24 self.stop()25 self.alive = True26 if Task.lock.acquire():27 # print("start")28 import gc29 gc.collect()30 _thread.stack_size(size)31 _thread.start_new_thread(self.run, ())32 _thread.stack_size()33 Task.lock.release()34if __name__ == "__main__":35 import time36 def unit_test(args):37 print('unit_test', args)38 sleep_ms(500)39 tmp = Task(unit_test, {'name': 'unit_test_0'})40 tmp.start()41 time.sleep(1)42 tmp.set_cb(unit_test, {'name': 'unit_test_1'})43 tmp.start()44 time.sleep(2)45 tmp.set_cb(unit_test, {'name': 'unit_test_2'})46 tmp.start()47 # time.sleep(3)...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run fMBT automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
