Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Make AdsSymbol even more pythonic #192

Merged
merged 8 commits into from
Feb 11, 2021
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,11 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).

### Changed
* fixed error with source distribution not containing adslib directory

* [#192](https://github.com/stlehmann/pyads/pull/192) make AdsSymbol even more pythonic
* replace AdsSymbol.set_auto_update function by AdsSymbol.auto_update property
* make AdsSymbol.value a property
* AdsSymbol.value setter writes to plc if AdsSymbol.auto_update is True

### Removed

## 3.3.1
Expand Down
6 changes: 4 additions & 2 deletions doc/quickstart.md
Original file line number Diff line number Diff line change
Expand Up @@ -197,17 +197,19 @@ symbol.add_device_notification(my_func, attr=attr, user_handle=user_handle)
A built-in notification is available to automatically update the symbol buffer based on the remote value. This is disabled by default, enable it with:

```python
symbol.set_auto_update(True)
symbol.auto_update = True
```

This will create a new notification callback to update `symbol.value`. This can be efficient if the remote variable changes less frequently then your code runs. The number of notification callbacks will then be less than what the number of read operations would have been.

It can be disabled again with:

```python
symbol.set_auto_update(False)
symbol.auto_update = False
```

Using auto_update will also write the value immediately to the plc when `symbol.value` is changed.

Take care that `symbol.clear_notifications()` will *also* remove the auto-update notification. Like all symbol notifications, the auto-update will also be cleared automatically in the object destructor.

The connection will also be closed automatically when the object runs out of scope, making `plc.close()` optional.
Expand Down
77 changes: 51 additions & 26 deletions pyads/symbol.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,9 +8,10 @@
import re
from ctypes import sizeof
from typing import TYPE_CHECKING, Any, Optional, List, Tuple, Callable

from . import constants # To access all constants, use package notation
from .pyads_ex import adsGetSymbolInfo
from .structs import NotificationAttrib
from . import constants # To access all constants, use package notation

# ads.Connection relies on structs.AdsSymbol (but in type hints only), so use
# this 'if' to only include it when type hinting (False during execution)
Expand Down Expand Up @@ -91,8 +92,7 @@ def __init__(
self.index_group = index_group
self.symbol_type = symbol_type
self.comment = comment

self.value: Any = None
self._value: Any = None

if missing_info:
self._create_symbol_from_info() # Perform remote lookup
Expand All @@ -105,7 +105,7 @@ def __init__(
if self.symbol_type is not None:
self.plc_type = AdsSymbol.get_type_from_str(self.symbol_type)

self.set_auto_update(auto_update)
self.auto_update = auto_update

def _create_symbol_from_info(self) -> None:
"""Look up remaining info from the remote
Expand Down Expand Up @@ -143,8 +143,8 @@ def read(self) -> Any:
The new read value is also saved in the buffer.
"""
self._read_write_check()
self.value = self._plc.read(self.index_group, self.index_offset, self.plc_type)
return self.value
self._value = self._plc.read(self.index_group, self.index_offset, self.plc_type)
return self._value

def write(self, new_value: Optional[Any] = None) -> None:
"""Write a new value or the buffered value to the symbol.
Expand All @@ -156,9 +156,9 @@ def write(self, new_value: Optional[Any] = None) -> None:
"""
self._read_write_check()
if new_value is None:
new_value = self.value # Send buffered value instead
new_value = self._value # Send buffered value instead
else:
self.value = new_value # Update buffer with new value
self._value = new_value # Update buffer with new value
self._plc.write(self.index_group, self.index_offset, new_value, self.plc_type)

def __repr__(self):
Expand Down Expand Up @@ -216,30 +216,13 @@ def del_device_notification(self, handles: Tuple[int, int]) -> None:
self._plc.del_device_notification(*handles)
self._handles_list.remove(handles)

def set_auto_update(self, auto_update: bool) -> None:
"""Enable or disable auto-update of buffered value

This automatic update is done through a device notification. This
can be efficient when a remote variables changes its values less often
than your code run.
Clearing all device notifications will also disable auto-update.
Automatic update is disabled by default.
"""
if auto_update and self._auto_update_handle is None:
self._auto_update_handle = self.add_device_notification(
self._value_callback
)
elif not auto_update and self._auto_update_handle is not None:
self.del_device_notification(self._auto_update_handle)
self._auto_update_handle = None

def _value_callback(self, notification: Any, data_name: Any) -> None:
"""Internal callback used by auto-update"""

_handle, _datetime, value = self._plc.parse_notification(
notification, self.plc_type
)
self.value = value
self._value = value

@staticmethod
def get_type_from_str(type_str: str) -> Any:
Expand Down Expand Up @@ -288,3 +271,45 @@ def get_type_from_str(type_str: str) -> Any:
# error when they are being addressed

return None

@property
def auto_update(self) -> Any:
"""Return True if auto_update is enabled for this symbol."""
return self._auto_update_handle is not None

@auto_update.setter
def auto_update(self, value: bool) -> None:
"""Enable or disable auto-update of the buffered value.

This automatic update is done through a device notification. This
can be efficient when a remote variables changes its values less often
than your code run.

Clearing all device notifications will also disable auto-update.
Automatic update is disabled by default.
"""
if value and self._auto_update_handle is None:
self._auto_update_handle = self.add_device_notification(
self._value_callback
)
elif not value and self._auto_update_handle is not None:
self.del_device_notification(self._auto_update_handle)
self._auto_update_handle = None

@property
def value(self) -> Any:
"""Return the current value of the symbol."""
return self._value

@value.setter
def value(self, val: Any) -> None:
"""Set the current value of the symbol.

If auto_update is True then the the write command will be called automatically.

"""
self._value = val

# write value to plc if auto_update is enabled
if self.auto_update:
self.write(val)
23 changes: 20 additions & 3 deletions tests/test_symbol.py
Original file line number Diff line number Diff line change
Expand Up @@ -415,20 +415,37 @@ def test_auto_update(self):
symbol = self.plc.get_symbol(self.test_var.name)
self.assertIsNone(symbol._auto_update_handle)

symbol.set_auto_update(True)
symbol.auto_update = True
self.assertIsNotNone(symbol._auto_update_handle)
self.assertEqual(symbol.auto_update, True)

# Simulate value callback
notification = create_notification_struct(struct.pack("<d", 5334.1545))
symbol._value_callback(
pointer(notification),
(self.test_var.index_group, self.test_var.index_offset),
)

self.assertEqual(symbol.value, 5334.1545)

symbol.set_auto_update(False)
# test immediate writing to plc if auto_update is True
symbol.value = 123.456
r_value = self.plc.read(
symbol.index_group,
symbol.index_offset,
symbol.plc_type,
)
self.assertEqual(symbol.value, r_value)

symbol.auto_update = False
self.assertIsNone(symbol._auto_update_handle)
symbol.value = 0.0
r_value = self.plc.read(
symbol.index_group,
symbol.index_offset,
symbol.plc_type,
)
self.assertEqual(r_value, 123.456)



class TypesTestCase(unittest.TestCase):
Expand Down