p4pillon 0.2.0rc1
“Extensions to the p4p Server API, to aid in creation and managment of pvAccess servers.”
imports Path from pathlibimports pathlib
SHA-2569b50859e1c9bd4aed40d90f40dbdb5201635c092dae615ff6529072251a7aff5
MaleculeH₂(Db₂F)
Evidence
1"""
2Specifications for various Normative Type types and fields
3"""
4
5import re
6
7###
8# NTBase_required are used for identification.
9# The key:value pairs are required_field:regex
10ntscalar_required = {"value": re.compile("[?sbBhHiIlLfd]{1}")}
11
12ntscalararray_required = {"value": re.compile("a[?sbBhHiIlLfd]{1}")}
13
14ntenum_required = {
15 "value": {
1:12… figuration from a YAML file"""
2
3from __future__ import annotations
4
5import logging
6from pathlib import Path
7from typing import TYPE_CHECKING, Any
8
9import yaml
10
11from p4pillon.definitions import PVTypes
12from p4pillon.thread.pvrecipe import PVEnumRecipe, PVScalarArrayRecipe, PVScalarRecipe
13
14if TYPE_CHECKING:
15 from p4pillon.pvrecipe import BasePVRecipe
16 from p4pillon.server.server import Server
17
18logger = logging.getLogger(__name__)
19
20# A list of rule/handler specific configs in the YAML to be passed to a non-standard rule/handler.
21# The list contains tuples of a tag name (the string used to identify the parameters for this rule)
22# and the parameter type, e.g. dictionary, integer.
23# The tag in the YAML must have the same name as the rule for it to be processed
24# correctly by SharedNT.
25rule_configs = [("cal …
31:41… f server != None
32 """
33 pvconfigs = {}
34 with Path(filename).open(encoding="utf8") as f:
35 pvconfigs = yaml.load(f, yaml.SafeLoader)
36
37 return parse_config(pvconfigs, server)
38
39
40def parse_config_string(yaml_str: str, server: Server | None = None) -> dict[str, …
64:35… v_name
65
66 def cb(self, _v: Value) -> None:
67 """This callback "cb" is part of the context.monitor() functionality.
68 See https://epics-base.github.io/p4p/client.html#monitor for further information."""
69 self._server.put_pv_value(self._pv_name, {})
70
71 def set_calc(self, calc: dict[ …
125:38… led to get pv %s", pv_name)
126 return None
127 pvs.append(val)
128 except Exception: # noqa: PERF203 -- per-item error handling around I/O, returns on first failure
129 # If there's an error getting the value of a pv return None
130 logger.exception("Failed to get pv %s", pv_name)
⋯5 lines
No evidence locations were recorded for this file. Raw result