diff --git a/docs/snippets/static14.py b/docs/snippets/static14.py new file mode 100644 index 000000000..d09980979 --- /dev/null +++ b/docs/snippets/static14.py @@ -0,0 +1,127 @@ +import asyncio +import enum +import json +from dataclasses import KW_ONLY, dataclass +from pathlib import Path +from typing import TypeVar + +from fastcs.attribute_io import AttributeIO +from fastcs.attribute_io_ref import AttributeIORef +from fastcs.attributes import AttrR, AttrRW, AttrW +from fastcs.connections import IPConnection, IPConnectionSettings +from fastcs.controller import Controller +from fastcs.datatypes import Enum, Float, Int, String +from fastcs.launch import FastCS +from fastcs.logging import bind_logger, configure_logging +from fastcs.transport.epics.ca.transport import EpicsCATransport +from fastcs.transport.epics.options import EpicsGUIOptions, EpicsIOCOptions +from fastcs.wrappers import command, scan + +logger = bind_logger(__name__) + +NumberT = TypeVar("NumberT", int, float) + + +@dataclass +class TemperatureControllerAttributeIORef(AttributeIORef): + name: str + _: KW_ONLY + update_period: float | None = 0.2 + + +class TemperatureControllerAttributeIO( + AttributeIO[NumberT, TemperatureControllerAttributeIORef] +): + def __init__(self, connection: IPConnection, suffix: str = ""): + super().__init__() + + self.logger = bind_logger(__class__.__name__) + + self._connection = connection + self._suffix = suffix + + async def update(self, attr: AttrR[NumberT, TemperatureControllerAttributeIORef]): + query = f"{attr.io_ref.name}{self._suffix}?" + response = await self._connection.send_query(f"{query}\r\n") + value = response.strip("\r\n") + await attr.update(attr.dtype(value)) + + async def send( + self, attr: AttrW[NumberT, TemperatureControllerAttributeIORef], value: NumberT + ) -> None: + command = f"{attr.io_ref.name}{self._suffix}={attr.dtype(value)}" + + self.logger.info("Sending attribute value", command=command, attribute=attr) + + await self._connection.send_command(f"{command}\r\n") + + +class OnOffEnum(enum.StrEnum): + Off = "0" + On = "1" + + +class TemperatureRampController(Controller): + start = AttrRW(Int(), io_ref=TemperatureControllerAttributeIORef(name="S")) + end = AttrRW(Int(), io_ref=TemperatureControllerAttributeIORef(name="E")) + enabled = AttrRW(Enum(OnOffEnum), io_ref=TemperatureControllerAttributeIORef("N")) + target = AttrR(Float(), io_ref=TemperatureControllerAttributeIORef("T")) + actual = AttrR(Float(), io_ref=TemperatureControllerAttributeIORef("A")) + voltage = AttrR(Float()) + + def __init__(self, index: int, connection: IPConnection) -> None: + suffix = f"{index:02d}" + super().__init__( + f"Ramp{suffix}", ios=[TemperatureControllerAttributeIO(connection, suffix)] + ) + + +class TemperatureController(Controller): + device_id = AttrR(String(), io_ref=TemperatureControllerAttributeIORef("ID")) + power = AttrR(Float(), io_ref=TemperatureControllerAttributeIORef("P")) + ramp_rate = AttrRW(Float(), io_ref=TemperatureControllerAttributeIORef("R")) + + def __init__(self, ramp_count: int, settings: IPConnectionSettings): + self._ip_settings = settings + self._connection = IPConnection() + + super().__init__(ios=[TemperatureControllerAttributeIO(self._connection)]) + + self._ramp_controllers: list[TemperatureRampController] = [] + for index in range(1, ramp_count + 1): + controller = TemperatureRampController(index, self._connection) + self._ramp_controllers.append(controller) + self.add_sub_controller(f"R{index}", controller) + + async def connect(self): + await self._connection.connect(self._ip_settings) + + @scan(0.1) + async def update_voltages(self): + voltages = json.loads( + (await self._connection.send_query("V?\r\n")).strip("\r\n") + ) + for index, controller in enumerate(self._ramp_controllers): + await controller.voltage.update(float(voltages[index])) + + @command() + async def disable_all(self) -> None: + self.log_event("Disabling all ramps") + for rc in self._ramp_controllers: + await rc.enabled.put(OnOffEnum.Off, sync_setpoint=True) + # TODO: The requests all get concatenated and the sim doesn't handle it + await asyncio.sleep(0.1) + + +configure_logging() + +gui_options = EpicsGUIOptions( + output_path=Path(".") / "demo.bob", title="Demo Temperature Controller" +) +epics_ca = EpicsCATransport(gui=gui_options, epicsca=EpicsIOCOptions(pv_prefix="DEMO")) +connection_settings = IPConnectionSettings("localhost", 25565) +logger.info("Configuring connection settings", connection_settings=connection_settings) +fastcs = FastCS(TemperatureController(4, connection_settings), [epics_ca]) + +if __name__ == "__main__": + fastcs.run() diff --git a/docs/snippets/static15.py b/docs/snippets/static15.py new file mode 100644 index 000000000..ff74ab306 --- /dev/null +++ b/docs/snippets/static15.py @@ -0,0 +1,130 @@ +import asyncio +import enum +import json +from dataclasses import KW_ONLY, dataclass +from pathlib import Path +from typing import TypeVar + +from fastcs.attribute_io import AttributeIO +from fastcs.attribute_io_ref import AttributeIORef +from fastcs.attributes import AttrR, AttrRW, AttrW +from fastcs.connections import IPConnection, IPConnectionSettings +from fastcs.controller import Controller +from fastcs.datatypes import Enum, Float, Int, String +from fastcs.launch import FastCS +from fastcs.logging import LogLevel, bind_logger, configure_logging +from fastcs.transport.epics.ca.transport import EpicsCATransport +from fastcs.transport.epics.options import EpicsGUIOptions, EpicsIOCOptions +from fastcs.wrappers import command, scan + +logger = bind_logger(__name__) + +NumberT = TypeVar("NumberT", int, float) + + +@dataclass +class TemperatureControllerAttributeIORef(AttributeIORef): + name: str + _: KW_ONLY + update_period: float | None = 0.2 + + +class TemperatureControllerAttributeIO( + AttributeIO[NumberT, TemperatureControllerAttributeIORef] +): + def __init__(self, connection: IPConnection, suffix: str = ""): + super().__init__() + + self.logger = bind_logger(__class__.__name__) + + self._connection = connection + self._suffix = suffix + + async def update(self, attr: AttrR[NumberT, TemperatureControllerAttributeIORef]): + query = f"{attr.io_ref.name}{self._suffix}?" + response = await self._connection.send_query(f"{query}\r\n") + value = response.strip("\r\n") + + self.log_event("Query for attribute", query=query, response=value, topic=attr) + + await attr.update(attr.dtype(value)) + + async def send( + self, attr: AttrW[NumberT, TemperatureControllerAttributeIORef], value: NumberT + ) -> None: + command = f"{attr.io_ref.name}{self._suffix}={attr.dtype(value)}" + + self.logger.info("Sending attribute value", command=command, attribute=attr) + + await self._connection.send_command(f"{command}\r\n") + + +class OnOffEnum(enum.StrEnum): + Off = "0" + On = "1" + + +class TemperatureRampController(Controller): + start = AttrRW(Int(), io_ref=TemperatureControllerAttributeIORef(name="S")) + end = AttrRW(Int(), io_ref=TemperatureControllerAttributeIORef(name="E")) + enabled = AttrRW(Enum(OnOffEnum), io_ref=TemperatureControllerAttributeIORef("N")) + target = AttrR(Float(), io_ref=TemperatureControllerAttributeIORef("T")) + actual = AttrR(Float(), io_ref=TemperatureControllerAttributeIORef("A")) + voltage = AttrR(Float()) + + def __init__(self, index: int, connection: IPConnection) -> None: + suffix = f"{index:02d}" + super().__init__( + f"Ramp{suffix}", ios=[TemperatureControllerAttributeIO(connection, suffix)] + ) + + +class TemperatureController(Controller): + device_id = AttrR(String(), io_ref=TemperatureControllerAttributeIORef("ID")) + power = AttrR(Float(), io_ref=TemperatureControllerAttributeIORef("P")) + ramp_rate = AttrRW(Float(), io_ref=TemperatureControllerAttributeIORef("R")) + + def __init__(self, ramp_count: int, settings: IPConnectionSettings): + self._ip_settings = settings + self._connection = IPConnection() + + super().__init__(ios=[TemperatureControllerAttributeIO(self._connection)]) + + self._ramp_controllers: list[TemperatureRampController] = [] + for index in range(1, ramp_count + 1): + controller = TemperatureRampController(index, self._connection) + self._ramp_controllers.append(controller) + self.add_sub_controller(f"R{index}", controller) + + async def connect(self): + await self._connection.connect(self._ip_settings) + + @scan(0.1) + async def update_voltages(self): + voltages = json.loads( + (await self._connection.send_query("V?\r\n")).strip("\r\n") + ) + for index, controller in enumerate(self._ramp_controllers): + await controller.voltage.update(float(voltages[index])) + + @command() + async def disable_all(self) -> None: + self.log_event("Disabling all ramps") + for rc in self._ramp_controllers: + await rc.enabled.put(OnOffEnum.Off, sync_setpoint=True) + # TODO: The requests all get concatenated and the sim doesn't handle it + await asyncio.sleep(0.1) + + +configure_logging(LogLevel.TRACE) + +gui_options = EpicsGUIOptions( + output_path=Path(".") / "demo.bob", title="Demo Temperature Controller" +) +epics_ca = EpicsCATransport(gui=gui_options, epicsca=EpicsIOCOptions(pv_prefix="DEMO")) +connection_settings = IPConnectionSettings("localhost", 25565) +logger.info("Configuring connection settings", connection_settings=connection_settings) +fastcs = FastCS(TemperatureController(4, connection_settings), [epics_ca]) + +if __name__ == "__main__": + fastcs.run() diff --git a/docs/tutorials/static-drivers.md b/docs/tutorials/static-drivers.md index ebb1019af..7768908df 100644 --- a/docs/tutorials/static-drivers.md +++ b/docs/tutorials/static-drivers.md @@ -214,7 +214,7 @@ because the value is actually a string, but for `P` the value is a float, so the `update` methods needs to explicitly cast to the correct type. It can use `Attribute.dtype` to call the builtin for its datatype - e.g. `int`, `float`, `str`, etc. -:::: +::: :::{admonition} Code 8 :class: dropdown, hint @@ -405,6 +405,95 @@ New : DEMO:DisableAll DEMO:R1:Enabled_RBV Off ``` +## Logging + +FastCS has convenient logging support to provide status and metrics from the +application. To enable logging from the core framework call `configure_logging` with no +arguments (the default logging level is INFO). To log messages from a driver, either +import the singleton `logger` directly, or to provide more context to the message, call +`bind_logger` with a name (usually either the name of the module or the name of the +class). + +Create a module-level logger to log status of the application start up. Create a class +logger for `TemperatureControllerAttributeIO` to log the commands it sends. + +::::{admonition} Code 14 +:class: dropdown, hint + +:::{literalinclude} /snippets/static14.py +:emphasize-lines: 15,20-21,53-55,116,123 +::: + +:::: + +Try setting a PV and check the console for the log message it prints. + +``` +[2025-11-18 11:26:41.065+0000 I] Sending attribute value [TemperatureControllerAttributeIO] command=E01=70, attribute=AttrRW(path=R1.end, datatype=Int, io_ref=TemperatureControllerAttributeIORef(update_period=0.2, name='E')) +``` + +A similar log message could be added for the update method of the IO, but this would be +very verbose. For this use case FastCS provides the `Tracer` class, which is inherited +by `AttributeIO`, among other core FastCS classes. This adds a enables logging `TRACE` +level log messages that are disabled by default, but can be enabled at runtime. + +Update the `send` method of the IO to log a message showing the query that was sent and +the response from the device. Update the `configure_logging` call to pass +`LogLevel.TRACE` as the log level, so that when tracing is enabled the messages are +visible. + +::::{admonition} Code 15 +:class: dropdown, hint + +:::{literalinclude} /snippets/static15.py +:emphasize-lines: 15,47-49,119 +::: + +:::: + +Enable tracing on the `power` attribute by calling `enable_tracing` and then enable a +ramp so that the value updates. Check the console to see the messages. Call `disable_tracing` to disable the log messages for `power. + +``` +In [1]: controller.power.enable_tracing() +[2025-11-18 11:11:12.060+0000 T] Query for attribute [TemperatureControllerAttributeIO] query=P?, response=0.0 +[2025-11-18 11:11:12.060+0000 T] Attribute set [AttrR] attribute=AttrR(path=power, datatype=Float, io_ref=TemperatureControllerAttributeIORef(update_period=0.2, name='P')), value=0.0 +[2025-11-18 11:11:12.060+0000 T] PV set from attribute [fastcs.transport.epics.ca.ioc] pv=DEMO:Power, value=0.0 +[2025-11-18 11:11:12.194+0000 I] PV put: DEMO:R1:Enabled = 1 [fastcs.transport.epics.ca.ioc] pv=DEMO:R1:Enabled, value=1 +[2025-11-18 11:11:12.195+0000 I] Sending attribute value [TemperatureControllerAttributeIO] command=N01=1, attribute=AttrRW(path=R1.enabled, datatype=Enum, io_ref=TemperatureControllerAttributeIORef(update_period=0.2, name='N')) +[2025-11-18 11:11:12.261+0000 T] Update attribute [AttrR] +[2025-11-18 11:11:12.262+0000 T] Query for attribute [TemperatureControllerAttributeIO] query=P?, response=29.040181873093132 +[2025-11-18 11:11:12.262+0000 T] Attribute set [AttrR] attribute=AttrR(path=power, datatype=Float, io_ref=TemperatureControllerAttributeIORef(update_period=0.2, name='P')), value=29.040181873093132 +[2025-11-18 11:11:12.262+0000 T] PV set from attribute [fastcs.transport.epics.ca.ioc] pv=DEMO:Power, value=29.04 +[2025-11-18 11:11:12.463+0000 T] Update attribute [AttrR] +[2025-11-18 11:11:12.464+0000 T] Query for attribute [TemperatureControllerAttributeIO] query=P?, response=30.452524641833854 +[2025-11-18 11:11:12.464+0000 T] Attribute set [AttrR] attribute=AttrR(path=power, datatype=Float, io_ref=TemperatureControllerAttributeIORef(update_period=0.2, name='P')), value=30.452524641833854 +[2025-11-18 11:11:12.465+0000 T] PV set from attribute [fastcs.transport.epics.ca.ioc] pv=DEMO:Power, value=30.45 +In [2]: controller.power.disable_tracing() +``` + +These log messages includes other trace loggers that log messages with `power` as the +`topic`, so they also appear automatically, so the log messages show changes to the +attribute throughout the stack: the query to the device and its response, the value the +attribute is set to, and the value that the PV in the EPICS CA transport is set to. + + +:::{note} +The `Tracer` can also be used as a module-level instance for use in free functions. + +```python +from fastcs.tracer import Tracer + +tracer = Tracer(__name__) + +def handle_attribute(attr): + tracer.log_event("Handling attribute", topic=attr) +``` + +These messages can then be enabled by calling `enable_tracing` on the module-level +`Tracer`, or more likely on a specific attribute. +::: + ## Summary This demonstrates some of the simple use cases for a statically defined FastCS driver.