Skip to content
Merged
2 changes: 2 additions & 0 deletions .github/workflows/deploy-pypi.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@ jobs:

steps:
- uses: actions/checkout@v4
with:
submodules: recursive
- name: Set up Python
uses: actions/setup-python@v5
with:
Expand Down
24 changes: 24 additions & 0 deletions .github/workflows/ruff_formatting.yml
Original file line number Diff line number Diff line change
@@ -0,0 +1,24 @@
name: Ruff Format Check

on:
push:
branches: [ main ]
pull_request:
branches: [ main ]

jobs:
ruff-check:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4

- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: "3.11" # or your preferred version

- name: Install Ruff
run: pip install ruff

- name: Run Ruff Check
run: ruff check --diff .
49 changes: 49 additions & 0 deletions .github/workflows/tests.yml
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
# This workflow will install Python dependencies, run tests and lint with a single version of Python
# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python

name: run tests

on:
push:
branches: [ "main" ]
pull_request:
branches: [ "main" ]

permissions:
contents: read

jobs:
build:

runs-on: ubuntu-latest

steps:
- uses: actions/checkout@v4
with:
submodules: recursive
- name: Set up Python 3.12
uses: actions/setup-python@v3
with:
python-version: ">=3.12"
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install pytango
pip install .
pip install numpy
pip install scipy
pip install pydantic
pip install accelerator-toolbox
pip install matplotlib
pip install h5py
pip install accelerator-middle-layer
pip install flake8 pytest
if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
- name: Lint with flake8
run: |
# stop the build if there are Python syntax errors or undefined names
flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
# exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
- name: Test with pytest
run: python -m pytest tests
21 changes: 21 additions & 0 deletions .pre-commit-config.yaml
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
repos:
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v6.0.0
hooks:
- id: trailing-whitespace
- id: end-of-file-fixer

- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.14.5
hooks:
- id: ruff
args: [--fix]
- id: ruff-format

# - repo: https://github.com/pre-commit/mirrors-mypy
# rev: v1.18.2
# hooks:
# - id: mypy
# additional_dependencies: [
# "pydantic>=2.0",
# ]
3 changes: 2 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ classifiers = [

dependencies = [
"PyTango>=10.1.1",
"accelerator-middle-layer<=0.1.1",
"accelerator-middle-layer<=0.2.2",
"pydantic>=2.0"
]

Expand All @@ -53,6 +53,7 @@ dev = [
"ruff", # Linter (optionnel)
"mypy", # Typage statique (optionnel)
"ipython", # Débogage interactif
"pre-commit",
]

[project.urls]
Expand Down
4 changes: 2 additions & 2 deletions tango/pyaml/__init__.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
__version__ = "0.3.1"
__version__ = "0.3.2"

import logging.config
import os
Expand All @@ -10,5 +10,5 @@

logger = logging.getLogger("tango.pyaml")
level = os.getenv("TANGO_PYAML_LOG_LEVEL", "").upper()
if len(level)>0:
if len(level) > 0:
logger.setLevel(getattr(logging, level, logging.WARNING))
87 changes: 68 additions & 19 deletions tango/pyaml/attribute.py
Original file line number Diff line number Diff line change
@@ -1,18 +1,20 @@
import logging
from typing import Optional, Tuple

from pydantic import BaseModel

from pyaml.control.deviceaccess import DeviceAccess
from pyaml.control.readback_value import Value, Quality

from .controlsystem import TangoControlSystem
from .initializable_element import InitializableElement
from .device_factory import DeviceFactory
from .tango_pyaml_utils import *

PYAMLCLASS : str = "Attribute"
PYAMLCLASS: str = "Attribute"

logger = logging.getLogger(__name__)


class ConfigModel(BaseModel):
"""
Configuration model for Tango attributes.
Expand All @@ -23,9 +25,14 @@ class ConfigModel(BaseModel):
Full path of the Tango attribute (e.g., 'my/ps/device/current').
unit : str, optional
The unit of the attribute.
range : tuple(min, max), optional
Range of valid values. Use null for -∞ or +∞.
"""

attribute: str
unit: str = ""
range: Optional[Tuple[Optional[float], Optional[float]]] = None


class Attribute(DeviceAccess, InitializableElement):
"""
Expand All @@ -41,31 +48,43 @@ class Attribute(DeviceAccess, InitializableElement):
pyaml.PyAMLException
If the Tango attribute is not writable.
"""

def __init__(self, cfg: ConfigModel, writable=True):
super().__init__()
self._cfg = cfg
self._writable = writable
self._attribute_dev:tango.DeviceProxy = None
self._attribute_dev: tango.DeviceProxy = None
self._attr_config: tango.AttributeConfig = None
self._attribute_dev_name: str = None
self._attr_name: str = None

def initialize(self):
super().initialize()
try:
self._attribute_dev_name, self._attr_name = self._cfg.attribute.rsplit("/", 1)
self._attribute_dev_name, self._attr_name = self._cfg.attribute.rsplit(
"/", 1
)
self._attribute_dev = DeviceFactory().get_device(self._attribute_dev_name)
except tango.DevFailed as df:
raise tango_to_PyAMLException(df)

self._attr_config = self._attribute_dev.get_attribute_config(self._attr_name, wait=True)

self._attr_config: tango.AttributeConfig = (
self._attribute_dev.get_attribute_config(self._attr_name, wait=True)
)

if self._writable:
if self._attr_config .writable not in [tango._tango.AttrWriteType.READ_WRITE,
tango._tango.AttrWriteType.WRITE,
tango._tango.AttrWriteType.READ_WITH_WRITE]:
raise pyaml.PyAMLException(f"Tango attribute {self._cfg.attribute} is not writable.")

if self._attr_config.writable not in [
tango.AttrWriteType.READ_WRITE,
tango.AttrWriteType.WRITE,
tango.AttrWriteType.READ_WITH_WRITE,
]:
raise pyaml.PyAMLException(
f"Tango attribute {self._cfg.attribute} is not writable."
)

def is_writable(self):
return self._writable

def set(self, value: float):
"""
Write a value asynchronously to the Tango attribute.
Expand All @@ -81,13 +100,14 @@ def set(self, value: float):
If the Tango write fails.
"""
self._ensure_initialized()
logger.log(logging.DEBUG, f"Setting asynchronously {self._cfg.attribute} to {value}")
logger.log(
logging.DEBUG, f"Setting asynchronously {self._cfg.attribute} to {value}"
)
try:
self._attribute_dev.write_attribute_asynch(self._attr_name, value)
except tango.DevFailed as df:
raise tango_to_PyAMLException(df)


def set_and_wait(self, value: float):
"""
Write a value synchronously to the Tango attribute.
Expand Down Expand Up @@ -127,8 +147,10 @@ def readback(self) -> Value:
logger.log(logging.DEBUG, f"Reading {self._cfg.attribute}")
try:
attr_value = self._attribute_dev.read_attribute(self._attr_name)
quality = Quality[attr_value.quality.name.rsplit('_', 1)[1]] # AttrQuality.ATTR_VALID gives Quality.VALID
value = Value(attr_value.value, quality, attr_value.time.todatetime() )
quality = Quality[
attr_value.quality.name.rsplit("_", 1)[1]
] # AttrQuality.ATTR_VALID gives Quality.VALID
value = Value(attr_value.value, quality, attr_value.time.todatetime())
except tango.DevFailed as df:
raise tango_to_PyAMLException(df)
return value
Expand Down Expand Up @@ -164,7 +186,7 @@ def measure_name(self) -> str:
str
The attribute name (e.g., 'current').
"""
return self._cfg.attribute.rsplit('/', 1)[1]
return self._cfg.attribute.rsplit("/", 1)[1]

def get(self) -> float:
"""
Expand All @@ -185,6 +207,33 @@ def get(self) -> float:
return self._attribute_dev.read_attribute(self._attr_name).w_value
except tango.DevFailed as df:
raise tango_to_PyAMLException(df)


def get_range(self) -> list[float]:
attr_range: list[float] = [None, None]
if self._cfg.range is not None:
attr_range[0] = (
self._cfg.range[0] if self._cfg.range[0] is not None else None
)
attr_range[1] = (
self._cfg.range[1] if self._cfg.range[1] is not None else None
)
else:
self._ensure_initialized()
min_value = self._attr_config.min_value
max_value = self._attr_config.max_value
attr_range[0] = to_float_or_none(min_value)
attr_range[1] = to_float_or_none(max_value)

return attr_range

def check_device_availability(self) -> bool:
available = True
try:
self._ensure_initialized()
self._attribute_dev.ping()
except tango.DevFailed | pyaml.PyAMLException:
available = False
return available

def __repr__(self):
return repr(self._cfg).replace("ConfigModel",self.__class__.__name__)
return repr(self._cfg).replace("ConfigModel", self.__class__.__name__)
Loading