Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _parse_header_line(cls, line):
"""
Parses the header line in a ASDF file to obtain the ASDF version.
"""
parts = line.split()
if len(parts) != 2 or parts[0] != constants.ASDF_MAGIC:
raise ValueError("Does not appear to be a ASDF file.")
try:
version = versioning.AsdfVersion(parts[1].decode('ascii'))
except ValueError:
raise ValueError(
"Unparseable version in ASDF file: {0}".format(parts[1]))
return version
def __eq__(self, other):
# Seems like a bit of a hack...
if isinstance(other, SpecItem):
return other == self
if isinstance(other, (str, tuple, list)):
other = AsdfVersion(other)
return Version.__eq__(self, other)
def __iterate_versions(self, versions):
for v in versions:
if isinstance(v, (str, tuple, list)):
v = AsdfVersion(v)
yield v
def match(self, version):
if isinstance(version, (str, tuple, list)):
version = AsdfVersion(version)
return super(AsdfSpec, self).match(version)
self._extensions_used = set()
try:
version_map = get_version_map(self._version)
core_version_map = version_map['core']
standard_version_map = version_map['standard']
except ValueError:
raise ValueError(
"Don't know how to write out ASDF version {0}".format(
self._version))
# Process all types defined in the ASDF version map. It is important to
# make sure that tags that are associated with the core part of the
# standard are processed first in order to handle subclasses properly.
for name, _version in core_version_map.items():
self._add_by_tag(index, name, AsdfVersion(_version))
for name, _version in standard_version_map.items():
self._add_by_tag(index, name, AsdfVersion(_version))
# Now add any extension types that aren't known to the ASDF standard.
# This expects that all types defined by ASDF will be encountered
# before any types that are defined by external packages. This
# allows external packages to override types that are also defined
# by ASDF. The ordering is guaranteed due to the use of OrderedDict
# for _versions_by_type_name, and due to the fact that the built-in
# extension will always be processed first.
for name, versions in index._versions_by_type_name.items():
if name not in self._type_by_name:
self._add_by_tag(index, name, versions[-1])
for asdftype in index._unnamed_types:
self._add_all_types(index, asdftype)
if hasattr(cls, 'name'):
if isinstance(cls.name, str):
if 'yaml_tag' not in attrs:
cls.yaml_tag = cls.make_yaml_tag(cls.name)
elif isinstance(cls.name, list):
pass
elif cls.name is not None:
raise TypeError("name must be string or list")
if hasattr(cls, 'supported_versions'):
if not isinstance(cls.supported_versions, (list, set)):
cls.supported_versions = [cls.supported_versions]
supported_versions = set()
for version in cls.supported_versions:
if not isinstance(version, (AsdfVersion, AsdfSpec)):
version = AsdfVersion(version)
# This should cause an exception for invalid input
supported_versions.add(version)
# We need to convert back to a list here so that the 'in' operator
# uses actual comparison instead of hash equality
cls.supported_versions = list(supported_versions)
siblings = list()
for version in cls.supported_versions:
if version != cls.version:
new_attrs = copy(attrs)
new_attrs['version'] = version
new_attrs['supported_versions'] = set()
new_attrs['_latest_version'] = cls.version
siblings.append(
ExtensionTypeMeta. __new__(mcls, name, bases, new_attrs))
setattr(cls, '__versioned_siblings', siblings)
def __eq__(self, other):
"""Equality between Spec and Version, string, or tuple, means match"""
if isinstance(other, SpecItem):
return super(AsdfSpec, self).__eq__(other)
return self.match(other)
def __ne__(self, other):
return not self.__eq__(other)
def __hash__(self):
return super(AsdfSpec, self).__hash__()
supported_versions = [
AsdfVersion('1.0.0'),
AsdfVersion('1.1.0'),
AsdfVersion('1.2.0'),
AsdfVersion('1.3.0')
]
default_version = supported_versions[-1]
class VersionedMixin:
_version = default_version
@property
def version(self):
return self._version
@version.setter