class MachineLogs(list):
    """Read in machine logs from a directory. Inherits from list. Batch methods are also provided."""

    def __init__(self, folder: str, recursive: bool = True):
        """
        Parameters
        ----------
        folder : str
            The directory of interest. Will walk through and process any logs, Trajectory or dynalog, it finds.
            Non-log files will be skipped.
        recursive : bool
            Whether to walk through subfolders of passed directory. Only used if ``folder`` is a valid log directory.

        Examples
        --------
        Load a directory upon initialization::

            >>> log_folder = r'C:\path\log\directory'
            >>> logs = MachineLogs(log_folder)

        Batch methods include determining the average gamma and average gamma pass value::

            >>> logs.avg_gamma()
            >>> 0.05 # or whatever it is
            >>> logs.avg_gamma_pct()
            >>> 97.2
        """
        super().__init__()
        self.load_folder(folder, recursive)

    @classmethod
    def from_zip(cls, zfile: str):
        """Instantiate from a ZIP archive.

        Parameters
        ----------
        zfile : str
            Path to the zip archive.
        """
        with io.TemporaryZipDirectory(zfile) as tzd:
            logs = cls(tzd)
        return logs

    @property
    def num_logs(self) -> int:
        """The number of logs currently loaded."""
        return len(self)

    @property
    def num_tlogs(self) -> int:
        """The number of Trajectory logs currently loaded."""
        return sum(isinstance(log, TrajectoryLog) for log in self)

    @property
    def num_dlogs(self) -> int:
        """The number of Trajectory logs currently loaded."""
        return sum(isinstance(log, Dynalog) for log in self)

    def load_folder(self, directory: str, recursive: bool = True):
        """Load log files from a directory and append to existing list.

        Parameters
        ----------
        directory : str, None
            The directory of interest.
            If a string, will walk through and process any logs, Trajectory or dynalog, it finds.
            Non-log files will be skipped.
            If None, files must be loaded later using .load_dir() or .append().
        recursive : bool
            If True (default), will walk through subfolders of passed directory.
            If False, will only search root directory.
        """
        # get log files from directory
        log_files = _get_log_filenames(directory, recursive=recursive)
        if len(log_files) == 0:
            print("No logs found.")
            return

        # actual log loading
        print(f"{len(log_files)} logs found.")
        for idx, file in enumerate(log_files):
            self.append(file)
            print(f"Log loaded: {idx+1} of {len(log_files)}", end='\r')
        print('')

    def _check_empty(self) -> None:
        """Check if any logs have been loaded."""
        if len(self) == 0:
            raise ValueError("No logs have been loaded yet.")

    def report_basic_parameters(self) -> None:
        """Report basic parameters of the logs.

        - Number of logs
        - Average gamma value of all logs
        - Average gamma pass percent of all logs
        """
        print(f"Number of logs: {len(self)}")
        print(f"Average gamma: {self.avg_gamma():3.2f}")
        print(f"Average gamma pass percent: {self.avg_gamma_pct():3.1f}")

    def append(self, obj, recursive: bool = True) -> None:
        """Append a log. Overloads list method.

        Parameters
        ----------
        obj : str, Dynalog, TrajectoryLog
            If a string, must point to a log file.
            If a directory, must contain log files.
            If a Dynalog or Trajectory log instance, then simply appends.
        recursive : bool
            Whether to walk through subfolders of passed directory. Only applicable if obj was a directory.
        """
        if isinstance(obj, str):
            if is_log(obj):
                log = load_log(obj)
                super().append(log)
            elif osp.isdir(obj):
                files = io.retrieve_filenames(obj)
                for file in files:
                    self.append(file)
        elif isinstance(obj, (Dynalog, TrajectoryLog)):
            super().append(obj)
        else:
            raise TypeError("Can only append MachineLog or string pointing to a log or log directory.")

    def avg_gamma(self, doseTA: Union[int, float] = 1, distTA: Union[int, float] = 1,
                  threshold: Union[int, float] = 0.1, resolution: Union[int, float] = 0.1) -> float:
        """Calculate and return the average gamma of all logs. See :meth:`~pylinac.log_analyzer.GammaFluence.calc_map()`
        for further parameter info."""
        self._check_empty()
        gamma_list = np.zeros(self.num_logs)

        for num, log in enumerate(self):
            log.fluence.gamma.calc_map(doseTA, distTA, threshold, resolution)
            gamma_list[num] = log.fluence.gamma.avg_gamma
            print(f'Calculating gammas: {num+1} of {self.num_logs}', end='\r')
        print('')
        return gamma_list.mean()

    def avg_gamma_pct(self, doseTA: Union[int, float] = 1, distTA: Union[int, float] = 1,
                      threshold: Union[int, float] = 0.1, resolution: Union[int, float] = 0.1) -> float:
        """Calculate and return the average gamma pass percent of all logs. See :meth:`~pylinac.log_analyzer.GammaFluence.calc_map()`
        for further parameter info."""
        self._check_empty()
        gamma_list = np.zeros(self.num_logs)

        for num, log in enumerate(self):
            log.fluence.gamma.calc_map(doseTA, distTA, threshold, resolution)
            gamma_list[num] = log.fluence.gamma.pass_prcnt
            print(f"Calculating gamma pass percent: {num+1} of {self.num_logs}", end='\r')
        print('')
        return gamma_list.mean()

    def to_csv(self) -> List[str]:
        """Write trajectory logs to CSV. If there are both dynalogs and trajectory logs,
        only the trajectory logs will be written. File names will be the same as the original log file names.

        Returns
        -------
        list
            A list of all the filenames of the newly created CSV files.
        """
        tlogs_written = False
        files = []
        for log in self:
            if is_tlog(log.filename):
                file = log.to_csv()
                tlogs_written = True
                files.append(file)
        if tlogs_written:
            print('\nAll trajectory logs written to CSV files!')
        else:
            print('\nNo files written. Either no logs are loaded or all logs were dynalogs.')
        return files

    def anonymize(self, inplace: bool = False, suffix: Optional[str] = None):
        """Save anonymized versions of the logs.

        For dynalogs, this replaces the patient ID in the filename(s) and the second line of the log with 'Anonymous<suffix>`.
        This will rename both A* and B* logs if both are present in the same directory.

        For trajectory logs, the patient ID in the filename is replaced with `Anonymous<suffix>` for the .bin file. If the
        associated .txt file is in the same directory it will similarly replace the patient ID in the filename with
        `Anonymous<suffix>`. Additionally, the `Patient ID` row will be replaced with `Patient ID: Anonymous<suffix>`.

        .. note::
            Anonymization is only available for logs loaded locally (i.e. not from a URL or a data stream). To
            anonymize such a log it must be first downloaded or written to a file, then loaded in.

        .. note::
            Anonymization is done to the log *file* itself. The current instance(s) of `MachineLog` will not be anonymized.

        Parameters
        ----------
        inplace : bool
            If False (default), creates an anonymized *copy* of the log(s).
            If True, *renames and replaces* the content of the log file.
        suffix : str, optional
            An optional suffix that is added after `Anonymous` to give specificity to the log.

        Returns
        -------
        list
            A list containing the paths to the newly written files.
        """
        file_list = []
        for log in self:
            files = log.anonymize(inplace=inplace, suffix=suffix)
            file_list += files
        print("\n\nDone anonymizing!")
        return file_list class TreatmentType(enum.Enum):
    STATIC_IMRT = 'Static IMRT'  #:
    DYNAMIC_IMRT = 'Dynamic IMRT'  #:
    VMAT = 'VMAT'  #:
    IMAGING = 'Imaging'  #: def anonymize(self, inplace: bool = False, suffix: Optional[str] = None):
    """Save anonymized versions of the logs.

    For dynalogs, this replaces the patient ID in the filename(s) and the second line of the log with 'Anonymous<suffix>`.
    This will rename both A* and B* logs if both are present in the same directory.

    For trajectory logs, the patient ID in the filename is replaced with `Anonymous<suffix>` for the .bin file. If the
    associated .txt file is in the same directory it will similarly replace the patient ID in the filename with
    `Anonymous<suffix>`. Additionally, the `Patient ID` row will be replaced with `Patient ID: Anonymous<suffix>`.

    .. note::
        Anonymization is only available for logs loaded locally (i.e. not from a URL or a data stream). To
        anonymize such a log it must be first downloaded or written to a file, then loaded in.

    .. note::
        Anonymization is done to the log *file* itself. The current instance(s) of `MachineLog` will not be anonymized.

    Parameters
    ----------
    inplace : bool
        If False (default), creates an anonymized *copy* of the log(s).
        If True, *renames and replaces* the content of the log file.
    suffix : str, optional
        An optional suffix that is added after `Anonymous` to give specificity to the log.

    Returns
    -------
    list
        A list containing the paths to the newly written files.
    """
    file_list = []
    for log in self:
        files = log.anonymize(inplace=inplace, suffix=suffix)
        file_list += files
    print("\n\nDone anonymizing!")
    return file_list class TrajectoryLog(LogBase):
    """A class for loading and analyzing the data of a Trajectory log.

    Attributes
    ----------
    header : `~pylinac.log_analyzer.TrajectoryLogHeader`, which has the following attributes:
    axis_data : `~pylinac.log_analyzer.TrajectoryLogAxisData`
    fluence : `~pylinac.log_analyzer.FluenceStruct`
    subbeams : `~pylinac.log_analyzer.SubbeamManager`
    """
    ANON_LINE = 0

    def __init__(self, filename: Union[str, BinaryIO], exclude_beam_off: bool = True):
        super().__init__(filename, exclude_beam_off)

        self._read_txt_file()

        # load from file object
        if isinstance(filename, (BinaryIO, BufferedReader)):
            filename.seek(0)
            self._read_it(filename)
        # load from disk
        else:
            with open(self.filename, mode='rb') as tlogfile:
                self._read_it(tlogfile)

        self.subbeams.post_hoc_metadata(self.axis_data)
        if not self.treatment_type == TreatmentType.IMAGING.value:
            self.fluence = FluenceStruct(self.axis_data.mlc, self.axis_data.mu, self.axis_data.jaws)

    def _read_it(self, tlogfile: BinaryIO):
        """Read the file object"""
        self.header = TrajectoryLogHeader(tlogfile)
        self.subbeams = SubbeamManager(tlogfile, self.header)
        self.axis_data = TrajectoryLogAxisData(self, tlogfile, self.subbeams)

    @property
    def txt_filename(self) -> str:
        """The name of the associated .txt file for the .bin file. The file may or may not be available."""
        if self.txt is not None:
            return self.filename.replace('.bin', '.txt')

    def anon_file_renames(self, destination: str, suffix: str) -> dict:
        base_filename = osp.basename(self.filename)
        anonymous_base_filename = 'Anonymous' + suffix + base_filename[self._underscore_idx:]
        anonymous_filename = osp.join(destination, anonymous_base_filename)
        filenames = collections.OrderedDict()
        filenames[self.filename] = anonymous_filename
        if self.txt_filename is not None:
            anonymous_txtfilename = anonymous_filename.replace('.bin', '.txt')
            filenames[self.txt_filename] = anonymous_txtfilename
        return filenames

    def anon_files(self, destination: str, suffix: str) -> list:
        renames = self.anon_file_renames(destination, suffix)
        if self.txt is not None:
            return [file for file in renames.values() if '.txt' in file]
        else:
            return []

    def _read_txt_file(self) -> None:
        """Read a Tlog's associated .txt file and put in under the 'txt' attribute."""
        self.txt = None
        if '.bin' in self.filename:  # files downloaded via URL may not have .bin ending
            txt_filename = self.filename.replace('.bin', '.txt')
            if osp.isfile(txt_filename):
                self.txt = {}
                with open(txt_filename) as txtfile:
                    txtdata = txtfile.readlines()
                for line in txtdata:
                    items = line.split(':')
                    if len(items) == 2:
                        self.txt[items[0].strip()] = items[1].strip()

    @classmethod
    def from_demo(cls, exclude_beam_off: bool = True):
        """Load and instantiate from the demo trajetory log file included with the package."""
        demo_file = io.retrieve_demo_file(url='Tlog.bin')
        return cls(demo_file, exclude_beam_off)

    @staticmethod
    def run_demo():
        """Run the Trajectory log demo."""
        tlog = TrajectoryLog.from_demo()
        tlog.report_basic_parameters()
        tlog.plot_summary()

    def to_csv(self, filename: Optional[str] = None) -> str:
        """Write the log to a CSV file.

        Parameters
        ----------
        filename : None, str
            If None (default), the CSV filename will be the same as the filename of the log.
            If a string, the filename will be named so.

        Returns
        -------
        str
            The full filename of the newly created CSV file.
        """
        if filename is None:
            filename = self.filename.replace('bin', 'csv')
        elif not filename.endswith('.csv'):
            filename += '.csv'

        csv_file = open(filename, mode='w')
        writer = csv.writer(csv_file, lineterminator='\n')
        # write header info
        header_titles = ('Tlog File:', 'Signature:', 'Version:', 'Header Size:', 'Sampling Inteval:',
                         'Number of Axes:', 'Axis Enumeration:', 'Samples per Axis:', 'Axis Scale:',
                         'Number of Subbeams:', 'Is Truncated?', 'Number of Snapshots:', 'MLC Model:')
        h = self.header
        header_values = (self.filename, h.header, h.version, h.header_size, h.sampling_interval,
                         h.num_axes, h.axis_enum, h.samples_per_axis, h.axis_scale, h.num_subbeams, h.is_truncated,
                         h.num_snapshots, h.mlc_model)
        for title, value in zip(header_titles, header_values):
            write_single_value(writer, title, value)

        # write axis data
        data_titles = ('Gantry', 'Collimator', 'Couch Lat', 'Couch Lng', 'Couch Rtn', 'MU',
                       'Beam Hold', 'Control Point', 'Carriage A', 'Carriage B')
        ad = self.axis_data
        data_values = (ad.gantry, ad.collimator, ad.couch.latl, ad.couch.long, ad.couch.rotn,
                       ad.mu, ad.beam_hold, ad.control_point, ad.carriage_A, ad.carriage_B)
        data_units = ('degrees', 'degrees', 'cm', 'cm', 'degrees', 'MU', None, None, 'cm',
                      'cm')
        for title, value, unit in zip(data_titles, data_values, data_units):
            write_array(writer, title, value, unit)

        # write leaf data
        for leaf_num, leaf in self.axis_data.mlc.leaf_axes.items():
            write_array(writer, 'Leaf ' + str(leaf_num), leaf, 'cm')

        print("CSV file written to: " + filename)
        return filename

    def publish_pdf(self, filename: Union[str, BinaryIO], metadata: dict = None, notes: Union[str, list] = None,
                    open_file: bool = False):
        """Publish (print) a PDF containing the analysis and quantitative results.

        Parameters
        ----------
        filename : (str, file-like object}
            The file to write the results to.
        notes : str, list of strings
            Text; if str, prints single line.
            If list of strings, each list item is printed on its own line.
        open_file : bool
            Whether to open the file using the default program after creation.
        metadata : dict
            Extra data to be passed and shown in the PDF. The key and value will be shown with a colon.
            E.g. passing {'Author': 'James', 'Unit': 'TrueBeam'} would result in text in the PDF like:
            --------------
            Author: James
            Unit: TrueBeam
            --------------
        """
        if self.treatment_type == TreatmentType.IMAGING.value:
            raise ValueError("Log is of imaging type (e.g. kV setup) and does not contain relevant gamma/leaf data")
        self.fluence.gamma.calc_map()
        canvas = pdf.PylinacCanvas(filename, page_title="Trajectory Log Analysis", metadata=metadata)
        canvas.add_text(
                      text=['Trajectory Log results:',
                            f'Average RMS (mm): {self.axis_data.mlc.get_RMS_avg()*10:2.2f}',
                            f'Max RMS (mm): {self.axis_data.mlc.get_RMS_max()*10:2.2f}',
                            f'95th Percentile error (mm): {self.axis_data.mlc.get_error_percentile(95)*10:2.2f}',
                            f'Number of beam holdoffs: {self.num_beamholds}',
                            f'Gamma pass (%): {self.fluence.gamma.pass_prcnt:2.1f}',
                            f'Gamma average: {self.fluence.gamma.avg_gamma:2.2f}',
                            ],
                        location=(10, 25.5))
        for idx, (x, y, graph) in enumerate(zip((2, 11, 2, 11), (14, 14, 6, 6), (Fluence.ACTUAL, Fluence.EXPECTED, Fluence.GAMMA, ''))):
            data = BytesIO()
            if graph != '':
                self.save_subimage(data, graph, fontsize=20)
            else:
                self.save_subgraph(data, Graph.GAMMA, fontsize=20, labelsize=12)
            canvas.add_image(data, location=(x, y), dimensions=(9, 9))
        if notes is not None:
            canvas.add_text(location=(1, 5.5), font_size=14, text="Notes:")
            canvas.add_text(location=(1, 5), text=notes)
        canvas.add_new_page()
        for idx, (x, y, graph) in enumerate(zip((5, 5), (13, 2), (Graph.HISTOGRAM, Graph.RMS))):
            data = BytesIO()
            self.save_subgraph(data, graph, fontsize=20, labelsize=12)
            canvas.add_image(location=(x, y), dimensions=(13, 13), image_data=data)
        canvas.finish()

        if open_file:
            open_path(filename)

    @property
    def num_beamholds(self) -> int:
        """Return the number of times the beam was held."""
        diffmatrix = np.diff(self.axis_data.beam_hold.actual)
        num_holds = int(np.sum(diffmatrix > 0))
        return num_holds

    @property
    def is_hdmlc(self) -> bool:
        """Whether the machine has an HDMLC or not."""
        return self.header.mlc_model == 3 class Dynalog(LogBase):
    """Class for loading, analyzing, and plotting data within a Dynalog file.

    Attributes
    ----------
    header : :class:`~pylinac.log_analyzer.DynalogHeader`
    axis_data : :class:`~pylinac.log_analyzer.DynalogAxisData`
    fluence : :class:`~pylinac.log_analyzer.FluenceStruct`
    """
    ANON_LINE = 1
    HEADER_LINE_LENGTH = 6

    def __init__(self, filename, exclude_beam_off: bool = True):
        super().__init__(filename, exclude_beam_off)
        if not is_dlog(self.filename):
            raise NotADynalogError(f"{self.filename} was not a valid Dynalog file")
        if not self._has_other_file:
            raise DynalogMatchError("Didn't find the matching dynalog file")  # TODO: clean up

        with open(self.a_logfile) as a_log:
            dlgdata = [line for line in csv.reader(a_log, delimiter=',')]
        self.header = DynalogHeader(dlgdata)
        self.axis_data = DynalogAxisData(self, dlgdata)
        self.fluence = FluenceStruct(self.axis_data.mlc, self.axis_data.mu, self.axis_data.jaws)

    def anon_file_renames(self, destination: str, suffix: str) -> dict:
        base_a = osp.basename(self.a_logfile)
        base_b = osp.basename(self.b_logfile)
        anonymous_base_a = base_a[:self._underscore_idx] + '_Anonymous' + suffix + '.dlg'
        anonymous_base_b = base_b[:self._underscore_idx] + '_Anonymous' + suffix + '.dlg'
        anonymous_a = osp.join(destination, anonymous_base_a)
        anonymous_b = osp.join(destination, anonymous_base_b)
        filenames = collections.OrderedDict()
        filenames[self.a_logfile] = anonymous_a
        filenames[self.b_logfile] = anonymous_b
        return filenames

    def anon_files(self, destination: str, suffix: str):
        return self.anon_file_renames(destination, suffix).values()

    def snapshot_idx(self, axis_data) -> list:
        if self.exclude_beam_off:
            hold_idx = np.where(axis_data.beam_hold.actual == 0)[0]
            beamon_idx = np.where(axis_data.beam_on.actual == 1)[0]
            snapshot_idx = np.intersect1d(hold_idx, beamon_idx)
        else:
            snapshot_idx = list(range(self.num_snapshots))
        return snapshot_idx

    @property
    def _has_other_file(self) -> bool:
        """Whether the companion file (A* for B-file or vic versa)."""
        return True if self.identify_other_file(self.filename, raise_find_error=False) is not None else False

    @cached_property
    def a_logfile(self) -> str:
        """Path of the A* dynalog file."""
        other_dlg_file = self.identify_other_file(self.filename)
        return self.filename if osp.basename(self.filename).startswith('A') else other_dlg_file

    @cached_property
    def b_logfile(self) -> str:
        """Path of the B* dynalog file."""
        other_dlg_file = self.identify_other_file(self.filename)
        return self.filename if osp.basename(self.filename).startswith('B') else other_dlg_file

    @property
    def num_beamholds(self) -> int:
        """Return the number of times the beam was held."""
        diffmatrix = np.diff(self.axis_data.beam_hold.actual)
        num_holds = int(np.sum(diffmatrix > 0))
        return num_holds

    @classmethod
    def from_demo(cls, exclude_beam_off: bool = True):
        """Load and instantiate from the demo dynalog file included with the package."""
        demo_file = io.retrieve_demo_file(url='AQA.dlg')
        io.retrieve_demo_file(url='BQA.dlg')  # also download "B" dynalog
        return cls(demo_file, exclude_beam_off)

    @staticmethod
    def run_demo():
        """Run the Dynalog demo."""
        dlog = Dynalog.from_demo()
        dlog.report_basic_parameters()
        dlog.plot_summary()

    def publish_pdf(self, filename: str, notes: str = None, metadata: dict = None, open_file: bool = False):
        """Publish (print) a PDF containing the analysis and quantitative results.

        Parameters
        ----------
        filename : (str, file-like object}
            The file to write the results to.
        notes : str, list of strings
            Text; if str, prints single line.
            If list of strings, each list item is printed on its own line.
        open_file : bool
            Whether to open the file using the default program after creation.
        metadata : dict
            Extra data to be passed and shown in the PDF. The key and value will be shown with a colon.
            E.g. passing {'Author': 'James', 'Unit': 'TrueBeam'} would result in text in the PDF like:
            --------------
            Author: James
            Unit: TrueBeam
            --------------
        """
        self.fluence.gamma.calc_map()
        canvas = pdf.PylinacCanvas(filename, page_title="Dynalog Analysis", metadata=metadata)
        canvas.add_text(
                      text=['Dynalog results:',
                            f'Average RMS (mm): {self.axis_data.mlc.get_RMS_avg()*10:2.2f}',
                            f'Max RMS (mm): {self.axis_data.mlc.get_RMS_max()*10:2.2f}',
                            f'95th Percentile error (mm): {self.axis_data.mlc.get_error_percentile(95)*10:2.2f}',
                            f'Number of beam holdoffs: {self.num_beamholds}',
                            f'Gamma pass (%): {self.fluence.gamma.pass_prcnt:2.1f}',
                            f'Gamma average: {self.fluence.gamma.avg_gamma:2.2f}',
                            ],
                        location=(10, 25.5))
        for idx, (x, y, graph) in enumerate(zip((2, 11, 2, 11), (14, 14, 6, 6), ('actual', 'expected', 'gamma', ''))):
            data = BytesIO()
            if idx != 3:
                self.save_subimage(data, graph, fontsize=20)
            else:
                self.save_subgraph(data, 'gamma', fontsize=20, labelsize=12)
            canvas.add_image(data, location=(x, y), dimensions=(9, 9))
        if notes is not None:
            canvas.add_text(location=(1, 5.5), font_size=14, text="Notes:")
            canvas.add_text(location=(1, 5), text=notes)
        canvas.add_new_page()
        for idx, (x, y, graph) in enumerate(zip((5, 5), (13, 2), (Graph.HISTOGRAM, Graph.RMS))):
            data = BytesIO()
            self.save_subgraph(data, graph, fontsize=20, labelsize=12)
            canvas.add_image(location=(x, y), dimensions=(13, 13), image_data=data)
        canvas.finish()

        if open_file:
            open_path(filename)

    @staticmethod
    def identify_other_file(first_dlg_file: str, raise_find_error: bool = True) -> str:
        """Return the filename of the corresponding dynalog file.

        For example, if the A*.dlg file was passed in, return the corresponding B*.dlg filename.
        Can find both A- and B-files.

        Parameters
        ----------
        first_dlg_file : str
            The absolute file path of the dynalog file.
        raise_find_error : bool
            Whether to raise an error if the file isn't found.

        Returns
        -------
        str
            The absolute file path to the corresponding dynalog file.
        """
        dlg_dir, dlg_file = osp.split(first_dlg_file)
        if dlg_file.startswith('A'):
            file2get = dlg_file.replace("A", "B", 1)
        elif dlg_file.startswith('B'):
            file2get = dlg_file.replace("B", "A", 1)
        else:
            raise ValueError("Unable to decipher log names; ensure dynalogs start with 'A' and 'B'")
        other_filename = osp.join(dlg_dir, file2get)

        if osp.isfile(other_filename):
            return other_filename
        elif raise_find_error:
            raise FileNotFoundError("Complementary dlg file not found; ensure A and B-file are in same directory.") def load_log(file_or_dir: str, exclude_beam_off: bool = True, recursive: bool = True) -> Union[
             TrajectoryLog, Dynalog, MachineLogs]:
    """Load a log file or directory of logs, either dynalogs or Trajectory logs.

    Parameters
    ----------
    file_or_dir : str
        String pointing to a single log file or a directory that contains log files.
    exclude_beam_off : bool
        Whether to include snapshots where the beam was off.
    recursive : bool
        Whether to recursively search a directory. Irrelevant for single log files.

    Returns
    -------
    One of :class:`~pylinac.log_analyzer.Dynalog`, :class:`~pylinac.log_analyzer.TrajectoryLog`,
        :class:`~pylinac.log_analyzer.MachineLogs`.
    """
    if io.is_url(file_or_dir):
        file_or_dir = io.get_url(file_or_dir)
    if osp.isfile(file_or_dir):
        if io.is_zipfile(file_or_dir):
            return MachineLogs.from_zip(file_or_dir)
        if not is_log(file_or_dir):
            raise NotALogError("Not a valid log")
        elif is_tlog(file_or_dir):
            return TrajectoryLog(file_or_dir, exclude_beam_off)
        else:
            return Dynalog(file_or_dir, exclude_beam_off)
    elif osp.isdir(file_or_dir):
        return MachineLogs(file_or_dir, recursive)
    else:
        raise NotALogError(f"'{file_or_dir}' did not point to a valid file, directory, or ZIP archive") class DynalogMatchError(IOError):
    """Dynalog error. Indicates that the associated file of the dynalog passed in
    (A file if B passed in & vic versa) cannot be found. Ensure associated file is in the same folder
    and has the same name as the passed file, except the first letter."""
    pass class NotADynalogError(IOError):
    """Dynalog error. Indicates that the passed file is not a valid dynalog file."""
    pass class NotALogError(IOError):
    """Machine log error. Indicates that the passed file is not a valid machine log file."""
    pass def save_file(method, *args, as_file_object=None, **kwargs):
    """Save a file using the passed method and assert it exists after saving.
    Also deletes the file after checking for existence."""
    if as_file_object is None:  # regular file
        with TemporaryDirectory() as tmpdir:
            tmpfile = osp.join(tmpdir, "myfile")
            method(tmpfile, *args, **kwargs)
    else:
        if "b" in as_file_object:
            temp = BytesIO
        elif "s" in as_file_object:
            temp = StringIO
        with temp() as t:
            method(t, *args, **kwargs) class CloudFileMixin:
    """A mixin that provides attrs and a method to get the absolute location of the
    file using syntax.

    Usage
    -----
    0. Override ``dir_location`` with the directory of the destination.
    1. Override ``file_path`` with a list that contains the subfolder(s) and file name.
    """

    file_name: str
    dir_path: Sequence[str]
    delete_file = True

    @classmethod
    def get_filename(cls) -> str:
        """Return the canonical path to the file on disk. Download if it doesn't exist."""
        full_path = Path(LOCAL_TEST_DIR, *cls.dir_path, cls.file_name).absolute()
        if full_path.is_file():
            return str(full_path)
        else:
            return get_file_from_cloud_test_repo([*cls.dir_path, cls.file_name])

    @classmethod
    def tearDownClass(cls):
        if cls.delete_file and DELETE_FILES:
            file = cls.get_filename()
            if osp.isfile(file):
                os.remove(file)
            elif osp.isdir(file):
                shutil.rmtree(file) def get_file_from_cloud_test_repo(path: List[str], force: bool = False) -> str:
    """Get a single file from GCP storage. Returns the path to disk it was downloaded to"""
    local_filename = osp.join(osp.dirname(__file__), LOCAL_TEST_DIR, *path)
    if osp.isfile(local_filename) and not force:
        return local_filename
    with access_gcp() as client:
        bucket = client.bucket(GCP_BUCKET_NAME)
        blob = bucket.blob(
            str(PurePosixPath(*path))
        )  # posix because google storage is on unix and won't find path w/ windows path
        # make any necessary subdirs leading up to the file
        if len(path) > 1:
            for idx in range(1, len(path)):
                local_dir = osp.join(osp.dirname(__file__), LOCAL_TEST_DIR, *path[:idx])
                if not osp.isdir(local_dir):
                    os.mkdir(local_dir)

        blob.download_to_filename(local_filename)
        return local_filename def get_folder_from_cloud_test_repo(folder: List[str]) -> str:
    """Get a folder from GCP"""
    with access_gcp() as storage_client:

        # get the folder data
        all_blobs = list(storage_client.list_blobs(GCP_BUCKET_NAME))
        blobs = (
            Enumerable(all_blobs)
            .where(lambda b: len(b.name.split("/")) >= len(folder))
            .where(lambda b: b.name.split("/")[1] != "")
            .where(
                lambda b: all(
                    f in b.name.split("/")[idx] for idx, f in enumerate(folder)
                )
            )
            .to_list()
        )

        # make root folder if need be
        dest_folder = osp.join(osp.dirname(__file__), LOCAL_TEST_DIR, *folder)
        if not osp.isdir(dest_folder):
            os.makedirs(dest_folder)

        # make subfolders if need be
        subdirs = [b.name.split("/")[1:-2] for b in blobs if len(b.name.split("/")) > 2]
        dest_sub_folders = [
            osp.join(osp.dirname(__file__), LOCAL_TEST_DIR, *folder, *f)
            for f in subdirs
        ]
        for sdir in dest_sub_folders:
            if not osp.isdir(sdir):
                os.makedirs(sdir)

        for blob in blobs:
            # download file
            path = osp.join(dest_folder, blob.name.split("/")[-1])
            if not os.path.exists(path):
                blob.download_to_filename(path)

        return osp.join(osp.dirname(__file__), LOCAL_TEST_DIR, *folder) class FromDemoImageTesterMixin(MixinTesterBase):
    demo_load_method: str = "from_demo_image"

    def test_from_demo(self):
        inst = getattr(self.klass, self.demo_load_method)()
        self.assertIsInstance(inst, self.klass) class FromURLTesterMixin(MixinTesterBase):
    url: str
    url_kwargs: dict = {}

    @property
    def full_url(self):
        return r"https://storage.googleapis.com/pylinac_demo_files/" + self.url

    def test_from_url(self):
        self.klass.from_url(self.full_url, **self.url_kwargs)
import io
import os.path as osp
import os
import shutil
import tempfile
import numpy as np
from unittest import TestCase
from pylinac.log_analyzer import MachineLogs, TreatmentType, \
    anonymize, TrajectoryLog, Dynalog, load_log, DynalogMatchError, NotADynalogError, NotALogError
from tests_basic.utils import save_file, CloudFileMixin, get_file_from_cloud_test_repo, \
    get_folder_from_cloud_test_repo, FromDemoImageTesterMixin, FromURLTesterMixin


class TestPublishPDF(TestCase):

    @classmethod
    def setUpClass(cls):
        cls.tlog = TrajectoryLog.from_demo()
        cls.dlog = Dynalog.from_demo()

    def test_publish_pdf(self):
        # normal publish; shouldn't raise
        with tempfile.TemporaryFile() as t:
            self.dlog.publish_pdf(t)
        with tempfile.TemporaryFile() as t:
            self.tlog.publish_pdf(t)

    def test_publish_pdf_w_imaging_log(self):
        imaging_tlog = TrajectoryLog(get_file_from_cloud_test_repo(['mlc_logs', 'tlogs', 'imaging.bin']))
        with self.assertRaises(ValueError), tempfile.NamedTemporaryFile() as t:
            imaging_tlog.publish_pdf(t.name)

    def test_publish_pdf_w_metadata_and_notes(self):
        with tempfile.TemporaryFile() as t:
            self.dlog.publish_pdf(t, metadata={'unit': 'TB1'}, notes='extra string')

        with tempfile.TemporaryFile() as t:
            self.tlog.publish_pdf(t, notes=['stuff', 'to', 'list'])


class LogPlottingSavingMixin:
    """Test the plotting methods and plot saving methods."""

    def test_plot_axes(self):
        for methodname in ('plot_actual', 'plot_expected', 'plot_difference'):
            method = getattr(self.log.axis_data.mlc.leaf_axes[10], methodname)
            method()  # shouldn't raise

    def test_save_axes(self):
        for methodname in ('save_plot_actual', 'save_plot_expected', 'save_plot_difference'):
            # save matplotlib figures
            method = getattr(self.log.axis_data.mlc.leaf_axes[10], methodname)
            save_file(method)

    def test_fluence_plotting(self):
        self.log.fluence.actual.calc_map()
        self.log.fluence.actual.plot_map()
        self.log.fluence.gamma.calc_map()
        self.log.fluence.gamma.histogram()
        self.log.fluence.gamma.plot_histogram()
        self.log.fluence.gamma.plot_passfail_map()

    def test_saving_fluence_plots(self):
        self.log.fluence.gamma.calc_map()
        save_file(self.log.fluence.gamma.save_map)
        save_file(self.log.fluence.gamma.save_histogram)

    def test_save_summary(self):
        self.log.fluence.gamma.calc_map()
        save_file(self.log.save_summary)


class TestTrajectoryTreatmentTypes(TestCase):

    def test_imaging_log(self):
        tlog = TrajectoryLog(get_file_from_cloud_test_repo(['mlc_logs', 'tlogs', 'imaging.bin']))
        self.assertTrue(tlog.treatment_type, TreatmentType.IMAGING.value)

    def test_vmat_log(self):
        tlog = TrajectoryLog(get_file_from_cloud_test_repo(['mlc_logs', 'tlogs', 'vmat.bin']))
        self.assertTrue(tlog.treatment_type, TreatmentType.VMAT.value)

    def test_static_imrt_log(self):
        tlog = TrajectoryLog(get_file_from_cloud_test_repo(['mlc_logs', 'tlogs', 'static_imrt.bin']))
        self.assertTrue(tlog.treatment_type, TreatmentType.STATIC_IMRT.value)

    def test_dynamic_imrt_log(self):
        tlog = TrajectoryLog(get_file_from_cloud_test_repo(['mlc_logs', 'tlogs', 'dynamic_imrt.bin']))
        self.assertTrue(tlog.treatment_type, TreatmentType.DYNAMIC_IMRT.value)


class TestDynalogTreatmentTypes(TestCase):

    def test_vmat_log(self):
        get_folder_from_cloud_test_repo(['mlc_logs', 'dlogs'])
        dlog = Dynalog(get_file_from_cloud_test_repo(['mlc_logs', 'dlogs', 'A_vmat.dlg']))
        self.assertTrue(dlog.treatment_type, TreatmentType.VMAT)

    def test_static_imrt_log(self):
        get_folder_from_cloud_test_repo(['mlc_logs', 'dlogs'])
        dlog = Dynalog(get_file_from_cloud_test_repo(['mlc_logs', 'dlogs', 'A_static_imrt.dlg']))
        self.assertTrue(dlog.treatment_type, TreatmentType.STATIC_IMRT)

    def test_dynamic_imrt_log(self):
        pass  # need to find one


class TestLoadLog(TestCase):

    def test_load_trajectory_log_from_file_object(self):
        path = get_file_from_cloud_test_repo(['mlc_logs', 'tlogs', 'dynamic_imrt.bin'])
        ref_log = TrajectoryLog(path)
        with open(path, 'rb') as f:
            t = TrajectoryLog(f)
        self.assertIsInstance(t, TrajectoryLog)
        self.assertEqual(t.num_beamholds, ref_log.num_beamholds)

    def test_dynalog_file(self):
        dynalog = get_file_from_cloud_test_repo(['mlc_logs', 'dlogs', 'A_static_imrt.dlg'])
        self.assertIsInstance(load_log(dynalog), Dynalog)

    def test_tlog_file(self):
        tlog = get_file_from_cloud_test_repo(['mlc_logs', 'tlogs', 'dynamic_imrt.bin'])
        self.assertIsInstance(load_log(tlog), TrajectoryLog)

    def test_url(self):
        url = r'https://s3.amazonaws.com/pylinac/Tlog.bin'
        self.assertIsInstance(load_log(url), TrajectoryLog)

    def test_dir(self):
        dlog_dir = get_folder_from_cloud_test_repo(['mlc_logs', 'dlogs'])

        self.assertIsInstance(load_log(dlog_dir), MachineLogs)