Source code for galaxy.datatypes.flow

"""
Flow analysis datatypes.
"""

import logging

from galaxy.datatypes.binary import Binary
from galaxy.datatypes.sniff import (
    build_sniff_from_prefix,
    FilePrefix,
)
from . import data

log = logging.getLogger(__name__)


[docs]@build_sniff_from_prefix
class FCS(Binary):
    """Class describing an FCS binary file"""
    file_ext = "fcs"

[docs]    def set_peek(self, dataset):
        if not dataset.dataset.purged:
            dataset.peek = "Binary FCS file"
            dataset.blurb = data.nice_size(dataset.get_size())
        else:
            dataset.peek = 'file does not exist'
            dataset.blurb = 'file purged from disk'

[docs]    def display_peek(self, dataset):
        try:
            return dataset.peek
        except Exception:
            return "Binary FCS file"

[docs]    def sniff_prefix(self, file_prefix: FilePrefix):
        """
        Checking if the file is in FCS format. Should read FCS2.0, FCS3.0
        and FCS3.1

        Based on flowcore:
        https://github.com/RGLab/flowCore/blob/27141b792ad65ae8bd0aeeef26e757c39cdaefe7/R/IO.R#L667
        """
        content = file_prefix.contents_header_bytes[:42].decode()
        version = content[:6]
        if version not in ["FCS2.0", "FCS3.0", "FCS3.1"]:
            return False
        if content[6:10] != '    ':
            return False
        # we only need to check ioffs 2 to 5
        int(content[10:42].replace(' ', ''))
        return True