Repository navigation
Pull out some methods from the cf processor to the model #185
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
base: master
Are you sure you want to change the base?
Changes from all commits
81f55dc
548e4f5
44e0f1b
8d002eb
6860558
af690ce
9db3b08
28a7753
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,6 +1,6 @@ | ||
| import enum | ||
| from dataclasses import dataclass, field | ||
| from typing import Any, Dict, List, Optional | ||
| from typing import Any, Dict, List, Optional, Set, Union | ||
|
|
||
|
|
||
| class PVStatus(enum.Enum): | ||
|
|
@@ -57,6 +57,43 @@ | |
| owner: str | ||
| properties: List[CFProperty] | ||
|
|
||
| def merged_properties( | ||
| self, new_properties: List[CFProperty], managed_properties: Optional[Set[str]] = None | ||
| ) -> List[CFProperty]: | ||
| """Return merged properties without modifying either input list. | ||
|
|
||
| New properties win on name collision. Existing properties are retained | ||
| unless replaced or omitted from the supplied managed property names. | ||
| """ | ||
| managed = managed_properties or set() | ||
| new_names = {prop.name for prop in new_properties} | ||
| return new_properties + [ | ||
| prop for prop in self.properties if prop.name not in new_names and prop.name not in managed | ||
| ] | ||
|
|
||
| def has_property(self, prop: Union[CFProperty, CFPropertyName, str]) -> bool: | ||
|
Check warning on line 74 in server/recceiver/cf/model.py
|
||
| """Check for a property name or an exact CFProperty match. | ||
|
|
||
| Name-only checks ignore owner and value, including None or empty values. | ||
| A CFProperty checks name, owner, and value using dataclass equality. | ||
| """ | ||
| if isinstance(prop, CFProperty): | ||
| return prop in self.properties | ||
| return self.property(prop) is not None | ||
|
|
||
| def property(self, name: Union[CFPropertyName, str]) -> Optional[CFProperty]: | ||
|
Check warning on line 84 in server/recceiver/cf/model.py
|
||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This API feels messy. There are three overlapping methods:
So two of them fetch a property (or its value), and the third only checks whether it exists, at about the same cost. As far as I can tell, none of the three are used by the processor. I'd rather you take this further and cut it down to one lookup method, e.g. |
||
| """Return the first property matching an enum or string name, or None.""" | ||
| property_name = name.value if isinstance(name, CFPropertyName) else name | ||
| return next((prop for prop in self.properties if prop.name == property_name), None) | ||
|
|
||
| def property_value(self, name: Union[CFPropertyName, str]) -> Optional[str]: | ||
|
Check warning on line 89 in server/recceiver/cf/model.py
|
||
| """Return the first matching property's value, or None if absent. | ||
|
|
||
| Accepts a canonical CFPropertyName or a string for custom properties. | ||
| """ | ||
| prop = self.property(name) | ||
| return prop.value if prop is not None else None | ||
|
|
||
| def as_dict(self) -> Dict[str, Any]: | ||
| """Serialise to the dict shape expected by pyCFClient.""" | ||
| return { | ||
|
|
@@ -92,6 +129,26 @@ | |
| def id(self) -> str: | ||
| return f"{self.host}:{self.port}" | ||
|
|
||
| def status_properties(self, status: PVStatus) -> List[CFProperty]: | ||
| """Build status and timestamp properties using this IOC's owner.""" | ||
| return [ | ||
| CFProperty(CFPropertyName.PV_STATUS.value, self.owner, status.value), | ||
| CFProperty(CFPropertyName.TIME.value, self.owner, self.time), | ||
| ] | ||
|
|
||
| def channel_properties(self, recceiverid: str) -> List[CFProperty]: | ||
| """Build the standard Active channel properties for this IOC.""" | ||
| return ( | ||
| [ | ||
| CFProperty(CFPropertyName.HOSTNAME.value, self.owner, self.hostname), | ||
| CFProperty(CFPropertyName.IOC_NAME.value, self.owner, self.ioc_name), | ||
| CFProperty(CFPropertyName.IOC_ID.value, self.owner, self.id), | ||
| CFProperty(CFPropertyName.IOC_IP.value, self.owner, self.ioc_ip), | ||
| ] | ||
| + self.status_properties(PVStatus.ACTIVE) | ||
| + [CFProperty(CFPropertyName.RECCEIVER_ID.value, self.owner, recceiverid)] | ||
| ) | ||
|
|
||
|
|
||
| @dataclass | ||
| class RecordInfo: | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -2,6 +2,7 @@ | |
| import logging | ||
| import time | ||
| from collections import defaultdict | ||
| from dataclasses import replace | ||
| from typing import Callable, Dict, List, Optional, Set | ||
|
|
||
| from channelfinder import ChannelFinderClient | ||
|
|
@@ -593,15 +594,7 @@ def _process_new_channels( | |
| genuinely new channels are created fresh. Appends results to channels. | ||
| """ | ||
| for channel_name in new_channels: | ||
| new_properties = create_ioc_properties( | ||
| ioc_info.owner, | ||
| ioc_info.time, | ||
| recceiverid, | ||
| ioc_info.hostname, | ||
| ioc_info.ioc_name, | ||
| ioc_info.ioc_ip, | ||
| ioc_info.id, | ||
| ) | ||
| new_properties = ioc_info.channel_properties(recceiverid) | ||
| record_info = record_info_by_name.get(channel_name) | ||
| if record_info: | ||
| if self.cf_config.record_type_enabled and record_info.record_type: | ||
|
|
@@ -680,9 +673,8 @@ def _handle_channel_is_old( | |
| last_ioc_info and must orphan the channel instead when it is absent. | ||
| """ | ||
| cf_channel.owner = last_ioc_info.owner | ||
| cf_channel.properties = _merge_property_lists( | ||
| cf_channel.properties = cf_channel.merged_properties( | ||
| create_default_properties(ioc_info, recceiverid, self.channel_ioc_ids, self.iocs, cf_channel), | ||
| cf_channel, | ||
| self.managed_properties, | ||
| ) | ||
| channels.append(cf_channel) | ||
|
|
@@ -698,11 +690,10 @@ def _handle_channel_is_old( | |
| if last_alias_ioc_info is None: | ||
| continue | ||
| alias_channel.owner = last_alias_ioc_info.owner | ||
| alias_channel.properties = _merge_property_lists( | ||
| alias_channel.properties = alias_channel.merged_properties( | ||
| create_default_properties( | ||
| ioc_info, recceiverid, self.channel_ioc_ids, self.iocs, cf_channel | ||
| ), | ||
| alias_channel, | ||
| self.managed_properties, | ||
| ) | ||
| channels.append(alias_channel) | ||
|
|
@@ -716,25 +707,15 @@ def _orphan_channel( | |
| record_info_by_name: Dict[str, RecordInfo], | ||
| ) -> None: | ||
| """Channel exists in CF but has no known IOC — mark inactive.""" | ||
| cf_channel.properties = _merge_property_lists( | ||
| [ | ||
| CFProperty(CFPropertyName.PV_STATUS.value, ioc_info.owner, PVStatus.INACTIVE.value), | ||
| CFProperty(CFPropertyName.TIME.value, ioc_info.owner, ioc_info.time), | ||
| ], | ||
| cf_channel, | ||
| ) | ||
| cf_channel.properties = cf_channel.merged_properties(ioc_info.status_properties(PVStatus.INACTIVE)) | ||
| channels.append(cf_channel) | ||
| log.debug("Add orphaned channel %s with no IOC: %s", cf_channel, ioc_info) | ||
| if self.cf_config.alias_enabled: | ||
| if cf_channel.name in record_info_by_name: | ||
| for alias_name in record_info_by_name[cf_channel.name].aliases: | ||
| alias_channel = CFChannel(alias_name, "", []) | ||
| alias_channel.properties = _merge_property_lists( | ||
| [ | ||
| CFProperty(CFPropertyName.PV_STATUS.value, ioc_info.owner, PVStatus.INACTIVE.value), | ||
| CFProperty(CFPropertyName.TIME.value, ioc_info.owner, ioc_info.time), | ||
| ], | ||
| alias_channel, | ||
| alias_channel.properties = alias_channel.merged_properties( | ||
| ioc_info.status_properties(PVStatus.INACTIVE) | ||
| ) | ||
| channels.append(alias_channel) | ||
| log.debug("Add orphaned alias %s with no IOC: %s", alias_channel, ioc_info) | ||
|
|
@@ -751,13 +732,8 @@ def _handle_channel_old_and_new( | |
| ) -> None: | ||
| """Channel exists in CF with the same iocid — mark active and update time.""" | ||
| log.debug("Channel %s exists in Channelfinder with same iocid %s", cf_channel.name, iocid) | ||
| cf_channel.properties = _merge_property_lists( | ||
| [ | ||
| CFProperty(CFPropertyName.PV_STATUS.value, ioc_info.owner, PVStatus.ACTIVE.value), | ||
| CFProperty(CFPropertyName.TIME.value, ioc_info.owner, ioc_info.time), | ||
| ], | ||
| cf_channel, | ||
| self.managed_properties, | ||
| cf_channel.properties = cf_channel.merged_properties( | ||
| ioc_info.status_properties(PVStatus.ACTIVE), self.managed_properties | ||
| ) | ||
| channels.append(cf_channel) | ||
| log.debug("Add existing channel with same IOC: %s", cf_channel) | ||
|
|
@@ -768,24 +744,15 @@ def _handle_channel_old_and_new( | |
| for alias_name in record_info_by_name[cf_channel.name].aliases: | ||
| if alias_name in old_channels: | ||
| alias_channel = CFChannel(alias_name, "", []) | ||
| alias_channel.properties = _merge_property_lists( | ||
| [ | ||
| CFProperty(CFPropertyName.PV_STATUS.value, ioc_info.owner, PVStatus.ACTIVE.value), | ||
| CFProperty(CFPropertyName.TIME.value, ioc_info.owner, ioc_info.time), | ||
| ], | ||
| alias_channel, | ||
| self.managed_properties, | ||
| alias_channel.properties = alias_channel.merged_properties( | ||
| ioc_info.status_properties(PVStatus.ACTIVE), self.managed_properties | ||
| ) | ||
| channels.append(alias_channel) | ||
| new_channels.remove(alias_name) | ||
| else: | ||
| aprops = _merge_property_lists( | ||
| [ | ||
| CFProperty(CFPropertyName.PV_STATUS.value, ioc_info.owner, PVStatus.ACTIVE.value), | ||
| CFProperty(CFPropertyName.TIME.value, ioc_info.owner, ioc_info.time), | ||
| CFProperty(CFPropertyName.ALIAS.value, ioc_info.owner, cf_channel.name), | ||
| ], | ||
| cf_channel, | ||
| aprops = cf_channel.merged_properties( | ||
| ioc_info.status_properties(PVStatus.ACTIVE) | ||
| + [CFProperty(CFPropertyName.ALIAS.value, ioc_info.owner, cf_channel.name)], | ||
| self.managed_properties, | ||
| ) | ||
| channels.append(CFChannel(alias_name, ioc_info.owner, aprops)) | ||
|
|
@@ -808,19 +775,15 @@ def _update_existing_channel_diff_iocid( | |
| ) -> None: | ||
| """Update a channel that exists in CF but is moving to a new IOC.""" | ||
| existing_channel = existing_channels[channel_name] | ||
| existing_channel.properties = _merge_property_lists( | ||
| new_properties, | ||
| existing_channel, | ||
| self.managed_properties, | ||
| ) | ||
| existing_channel.properties = existing_channel.merged_properties(new_properties, self.managed_properties) | ||
| channels.append(existing_channel) | ||
| log.debug("Add existing channel with different IOC: %s", existing_channel) | ||
| if self.cf_config.alias_enabled and channel_name in record_info_by_name: | ||
| alias_properties = [CFProperty(CFPropertyName.ALIAS.value, ioc_info.owner, channel_name)] + new_properties | ||
| for alias_name in record_info_by_name[channel_name].aliases: | ||
| if alias_name in existing_channels: | ||
| ach = existing_channels[alias_name] | ||
| ach.properties = _merge_property_lists(alias_properties, ach, self.managed_properties) | ||
| ach.properties = ach.merged_properties(alias_properties, self.managed_properties) | ||
| channels.append(ach) | ||
| else: | ||
| channels.append(CFChannel(alias_name, ioc_info.owner, alias_properties)) | ||
|
|
@@ -843,56 +806,17 @@ def _create_new_channel( | |
| log.debug("Add new alias: %s from %s", alias, channel_name) | ||
|
|
||
|
|
||
| def create_ioc_properties( | ||
| owner: str, ioc_time: str, recceiverid: str, host_name: str, ioc_name: str, ioc_ip: str, iocid: str | ||
| ) -> List[CFProperty]: | ||
| """Build the standard set of IOC-level CF properties for a channel.""" | ||
| return [ | ||
| CFProperty(CFPropertyName.HOSTNAME.value, owner, host_name), | ||
| CFProperty(CFPropertyName.IOC_NAME.value, owner, ioc_name), | ||
| CFProperty(CFPropertyName.IOC_ID.value, owner, iocid), | ||
| CFProperty(CFPropertyName.IOC_IP.value, owner, ioc_ip), | ||
| CFProperty(CFPropertyName.PV_STATUS.value, owner, PVStatus.ACTIVE.value), | ||
| CFProperty(CFPropertyName.TIME.value, owner, ioc_time), | ||
| CFProperty(CFPropertyName.RECCEIVER_ID.value, owner, recceiverid), | ||
| ] | ||
|
|
||
|
|
||
| def create_default_properties( | ||
| ioc_info: IOCInfo, | ||
| recceiverid: str, | ||
| channels_iocs: Dict[str, List[str]], | ||
| iocs: Dict[str, IOCInfo], | ||
| cf_channel: CFChannel, | ||
| ) -> List[CFProperty]: | ||
| """Build IOC properties using the last known IOC for a channel.""" | ||
| """Use the last IOC's identity with the current transaction's owner and time.""" | ||
| channel_name = cf_channel.name | ||
| last_ioc_info = iocs[channels_iocs[channel_name][-1]] | ||
| return create_ioc_properties( | ||
| ioc_info.owner, | ||
| ioc_info.time, | ||
| recceiverid, | ||
| last_ioc_info.hostname, | ||
| last_ioc_info.ioc_name, | ||
| last_ioc_info.ioc_ip, | ||
| last_ioc_info.id, | ||
| ) | ||
|
|
||
|
|
||
| def _merge_property_lists( | ||
| new_properties: List[CFProperty], channel: CFChannel, managed_properties: Optional[Set[str]] = None | ||
| ) -> List[CFProperty]: | ||
| """Merge two property lists; new_properties wins on name collision. | ||
|
|
||
| Properties in channel not in new_properties are kept unless they are | ||
| managed by this recceiver (in which case the absence is intentional). | ||
| """ | ||
| managed = managed_properties or set() | ||
| new_property_names = [p.name for p in new_properties] | ||
| for old_property in channel.properties: | ||
| if old_property.name not in new_property_names and old_property.name not in managed: | ||
| new_properties = new_properties + [old_property] | ||
| return new_properties | ||
| return replace(last_ioc_info, owner=ioc_info.owner, time=ioc_info.time).channel_properties(recceiverid) | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Nit: |
||
|
|
||
|
|
||
| def get_current_time(timezone: Optional[str] = None) -> str: | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I'm also not sure
merged_propertiesbelongs onCFChannel. As far as I can tell it isn't really a merge. It's one-way: the new properties overwrite the channel's existing ones, and an existing property whose name is inmanaged_propertiesis dropped unlessnew_propertiessupplies it. That's (maybe) fine as behaviour, but the method doesn't fit the model well:managed_propertiesis processor policy (the property names this recceiver is the authority for), not channel data. As a method, the model now takes processor config as an argument.aprops = cf_channel.merged_properties(...)uses the parent channel's properties as the base for a different (alias) channel. That's also the only reason this has to return a new list rather than update the channel.managed_properties, so every existing property is kept. I think that's intentional (an inactive channel keeps its last known properties), but it's a behaviour switch hidden behind an optional argument.I'd rather keep this in the processor as a plain function over property lists, with a name that says what it does, e.g.:
The alias case then just passes the parent's list. I know this is mostly moved code, so I don't consider it blocking. But if the goal is a cleaner model, I'd leave this one out of it.
Alternatively we just drop and inline it.