Skip to content

API

Here the API is documented.

fbc_curation.frog

FROG schema definition.

BaseModel

Bases: BaseModel

Base model.

Source code in src/fbc_curation/frog.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
class BaseModel(PydanticBaseModel):
    """Base model."""
    model_config = ConfigDict(
        use_enum_values=True,
    )

    @model_validator(mode='before')
    def change_nan_to_none(cls, values: dict[str, Any]) -> dict[str, Any]:
        """Replace NaN with None for all fields."""
        for field, value in values.items():
            if isinstance(value, float) and np.isnan(value):
                values[field] = None
        return values

change_nan_to_none(values)

Replace NaN with None for all fields.

Source code in src/fbc_curation/frog.py 
34
35
36
37
38
39
40
@model_validator(mode='before')
def change_nan_to_none(cls, values: dict[str, Any]) -> dict[str, Any]:
    """Replace NaN with None for all fields."""
    for field, value in values.items():
        if isinstance(value, float) and np.isnan(value):
            values[field] = None
    return values

Creator

Bases: BaseModel

Creator/curator in ModelHistory and other COMBINE formats.

Extended by optional orcid.

Source code in src/fbc_curation/frog.py 
114
115
116
117
118
119
120
121
122
123
124
125
class Creator(BaseModel):
    """Creator/curator in ModelHistory and other COMBINE formats.

    Extended by optional orcid.
    """

    familyName: str
    givenName: str
    email: Optional[str] = None
    organization: Optional[str] = None
    site: Optional[str] = None
    orcid: Optional[str] = None

CuratorConstants

Class storing constants for curation and file format.

Source code in src/fbc_curation/frog.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
class CuratorConstants:
    """Class storing constants for curation and file format."""

    FROG_KEY = "frog"
    METADATA_KEY = "metadata"
    OBJECTIVE_KEY = "objective"
    FVA_KEY = "fva"
    GENE_DELETION_KEY = "gene_deletion"
    REACTION_DELETION_KEY = "reaction_deletion"

    # output filenames
    FROG_FILENAME = "frog.json"
    METADATA_FILENAME = "metadata.json"
    OBJECTIVE_FILENAME = f"01_{OBJECTIVE_KEY}.tsv"
    FVA_FILENAME = f"02_{FVA_KEY}.tsv"
    GENE_DELETION_FILENAME = f"03_{GENE_DELETION_KEY}.tsv"
    REACTION_DELETION_FILENAME = f"04_{REACTION_DELETION_KEY}.tsv"

    # special settings for comparison
    VALUE_INFEASIBLE = np.nan

FrogFVA

Bases: BaseModel

Definition of FROG FVA.

Source code in src/fbc_curation/frog.py 
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
class FrogFVA(BaseModel):
    """Definition of FROG FVA."""

    fva: List[FrogFVASingle]


    @staticmethod
    def from_df(df: pd.DataFrame) -> FrogFVA:
        """Parse FVA from DataFrame."""

        fva = []
        for item in df.to_dict(orient="records"):
            try:
                fva.append(FrogFVASingle(**item))
            except ValidationError as e:
                logger.error(item)
                logger.error(e.json())
        return FrogFVA(fva=fva)

    def to_df(self) -> pd.DataFrame:
        """Create fva DataFrame."""

        d: Dict[str, Any] = self.model_dump()
        item = list(d.values())[0]
        df = pd.DataFrame(item)
        if len(df) > 0:
            df.sort_values(by=["reaction"], inplace=True)
            df.index = range(len(df))
            df.loc[
                df.status == StatusCode.INFEASIBLE.value,
                ["flux", "minimum", "maximum"],
            ] = CuratorConstants.VALUE_INFEASIBLE

        return df

from_df(df) staticmethod

Parse FVA from DataFrame.

Source code in src/fbc_curation/frog.py 
226
227
228
229
230
231
232
233
234
235
236
237
@staticmethod
def from_df(df: pd.DataFrame) -> FrogFVA:
    """Parse FVA from DataFrame."""

    fva = []
    for item in df.to_dict(orient="records"):
        try:
            fva.append(FrogFVASingle(**item))
        except ValidationError as e:
            logger.error(item)
            logger.error(e.json())
    return FrogFVA(fva=fva)

to_df()

Create fva DataFrame.

Source code in src/fbc_curation/frog.py 
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
def to_df(self) -> pd.DataFrame:
    """Create fva DataFrame."""

    d: Dict[str, Any] = self.model_dump()
    item = list(d.values())[0]
    df = pd.DataFrame(item)
    if len(df) > 0:
        df.sort_values(by=["reaction"], inplace=True)
        df.index = range(len(df))
        df.loc[
            df.status == StatusCode.INFEASIBLE.value,
            ["flux", "minimum", "maximum"],
        ] = CuratorConstants.VALUE_INFEASIBLE

    return df

FrogFVASingle

Bases: BaseModel

Frog FVA.

Source code in src/fbc_curation/frog.py 
81
82
83
84
85
86
87
88
89
90
91
class FrogFVASingle(BaseModel):
    """Frog FVA."""

    model: str
    objective: str
    reaction: str
    flux: Optional[float]
    status: StatusCode
    minimum: Optional[float]
    maximum: Optional[float]
    fraction_optimum: float

FrogGeneDeletion

Bases: BaseModel

Frog gene deletion.

Source code in src/fbc_curation/frog.py 
104
105
106
107
108
109
110
111
class FrogGeneDeletion(BaseModel):
    """Frog gene deletion."""

    model: str
    objective: str
    gene: str
    status: StatusCode
    value: Optional[float]

FrogGeneDeletions

Bases: BaseModel

Definition of FROG Gene deletions.

Source code in src/fbc_curation/frog.py 
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
class FrogGeneDeletions(BaseModel):
    """Definition of FROG Gene deletions."""

    deletions: List[FrogGeneDeletion]

    @staticmethod
    def from_df(df: pd.DataFrame) -> FrogGeneDeletions:
        """Parse GeneDeletions from DataFrame."""

        json = df.to_dict(orient="records")
        deletions = []
        for item in json:
            try:
                deletions.append(FrogGeneDeletion(**item))
            except ValidationError as e:
                logger.error(item)
                logger.error(e.json())

        return FrogGeneDeletions(deletions=deletions)

    def to_df(self) -> pd.DataFrame:
        """Create gene deletions DataFrame."""

        d: Dict[str, Any] = self.model_dump()
        item = list(d.values())[0]
        df = pd.DataFrame(item)
        if len(df) > 0:
            df.sort_values(by=["gene"], inplace=True)
            df.index = range(len(df))
            df.loc[
                df.status == StatusCode.INFEASIBLE.value, "value"
            ] = CuratorConstants.VALUE_INFEASIBLE

        return df

from_df(df) staticmethod

Parse GeneDeletions from DataFrame.

Source code in src/fbc_curation/frog.py 
296
297
298
299
300
301
302
303
304
305
306
307
308
309
@staticmethod
def from_df(df: pd.DataFrame) -> FrogGeneDeletions:
    """Parse GeneDeletions from DataFrame."""

    json = df.to_dict(orient="records")
    deletions = []
    for item in json:
        try:
            deletions.append(FrogGeneDeletion(**item))
        except ValidationError as e:
            logger.error(item)
            logger.error(e.json())

    return FrogGeneDeletions(deletions=deletions)

to_df()

Create gene deletions DataFrame.

Source code in src/fbc_curation/frog.py 
311
312
313
314
315
316
317
318
319
320
321
322
323
324
def to_df(self) -> pd.DataFrame:
    """Create gene deletions DataFrame."""

    d: Dict[str, Any] = self.model_dump()
    item = list(d.values())[0]
    df = pd.DataFrame(item)
    if len(df) > 0:
        df.sort_values(by=["gene"], inplace=True)
        df.index = range(len(df))
        df.loc[
            df.status == StatusCode.INFEASIBLE.value, "value"
        ] = CuratorConstants.VALUE_INFEASIBLE

    return df

FrogMetaData

Bases: BaseModel

FROG metadata.

Source code in src/fbc_curation/frog.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
class FrogMetaData(BaseModel):
    """FROG metadata."""
    model_config = ConfigDict(
        use_enum_values=True,
        validate_by_name=True,
    )

    model_location: str = Field(
        alias="model.location",
        description="Location of the model in the COMBINE archive for which the FROG "
        "analysis was performed.",
    )
    model_md5: Optional[str] = Field(
        alias="model.md5",
        description="MD5 hash of model",
    )
    frog_id: str = Field(
        description="Id for the FROG analysis. All frog_ids within an archive must be "
        "unique."
    )
    frog_software: Tool = Field(
        alias="frog.software",
        description="Software used to run FROG (e.g. 'fbc_curation')",
    )
    curators: List[Creator] = Field(
        alias="frog.curators", description="Curators which executed the FROG analysis."
    )
    software: Tool = Field(
        description="Software used to run FBC (e.g. 'COBRA', 'cobrapy')."
    )
    solver: Tool = Field(
        description="Solver used to solve LP problem (e.g. 'CPLEX', 'GUROBI', 'GLPK')."
    )
    environment: Optional[str] = Field(
        description="Execution environment such as Linux."
    )

    @staticmethod
    def md5_for_path(path: Path) -> str:
        """Calculate MD5 of file content."""

        # Open,close, read file and calculate MD5 on its contents
        with open(path, "rb") as f_check:
            # read contents of the file
            data = f_check.read()
            # pipe contents of the file through
            return hashlib.md5(data).hexdigest()

md5_for_path(path) staticmethod

Calculate MD5 of file content.

Source code in src/fbc_curation/frog.py 
173
174
175
176
177
178
179
180
181
182
@staticmethod
def md5_for_path(path: Path) -> str:
    """Calculate MD5 of file content."""

    # Open,close, read file and calculate MD5 on its contents
    with open(path, "rb") as f_check:
        # read contents of the file
        data = f_check.read()
        # pipe contents of the file through
        return hashlib.md5(data).hexdigest()

FrogObjective

Bases: BaseModel

Frog Objective.

Source code in src/fbc_curation/frog.py 
72
73
74
75
76
77
78
class FrogObjective(BaseModel):
    """Frog Objective."""

    model: str
    objective: str
    status: StatusCode
    value: float

FrogObjectives

Bases: BaseModel

Definition of FROG Objectives.

Source code in src/fbc_curation/frog.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
class FrogObjectives(BaseModel):
    """Definition of FROG Objectives."""

    objectives: List[FrogObjective]

    @staticmethod
    def from_df(df: pd.DataFrame) -> FrogObjectives:
        """Parse Objectives from DataFrame."""
        json = df.to_dict(orient="records")
        objectives = []
        for item in json:
            try:
                objectives.append(FrogObjective(**item))
            except ValidationError as e:
                logger.error(item)
                logger.error(e.json())

        return FrogObjectives(objectives=objectives)

    def to_df(self) -> pd.DataFrame:
        """Create objectives DataFrame."""

        d: Dict[str, Any] = self.model_dump()
        item = list(d.values())[0]
        df = pd.DataFrame(item)
        if len(df) > 0:
            df.sort_values(by=["objective"], inplace=True)
            df.index = range(len(df))
            df.loc[
                df.status == StatusCode.INFEASIBLE.value, "value"
            ] = CuratorConstants.VALUE_INFEASIBLE

        return df

from_df(df) staticmethod

Parse Objectives from DataFrame.

Source code in src/fbc_curation/frog.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
@staticmethod
def from_df(df: pd.DataFrame) -> FrogObjectives:
    """Parse Objectives from DataFrame."""
    json = df.to_dict(orient="records")
    objectives = []
    for item in json:
        try:
            objectives.append(FrogObjective(**item))
        except ValidationError as e:
            logger.error(item)
            logger.error(e.json())

    return FrogObjectives(objectives=objectives)

to_df()

Create objectives DataFrame.

Source code in src/fbc_curation/frog.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
def to_df(self) -> pd.DataFrame:
    """Create objectives DataFrame."""

    d: Dict[str, Any] = self.model_dump()
    item = list(d.values())[0]
    df = pd.DataFrame(item)
    if len(df) > 0:
        df.sort_values(by=["objective"], inplace=True)
        df.index = range(len(df))
        df.loc[
            df.status == StatusCode.INFEASIBLE.value, "value"
        ] = CuratorConstants.VALUE_INFEASIBLE

    return df

FrogReactionDeletion

Bases: BaseModel

Frog reaction deletion.

Source code in src/fbc_curation/frog.py 
 94
 95
 96
 97
 98
 99
100
101
class FrogReactionDeletion(BaseModel):
    """Frog reaction deletion."""

    model: str
    objective: str
    reaction: str
    status: StatusCode
    value: Optional[float]

FrogReactionDeletions

Bases: BaseModel

Definition of FROG Reaction deletions.

Source code in src/fbc_curation/frog.py 
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
class FrogReactionDeletions(BaseModel):
    """Definition of FROG Reaction deletions."""

    deletions: List[FrogReactionDeletion]

    @staticmethod
    def from_df(df: pd.DataFrame) -> FrogReactionDeletions:
        """Parse FVA from DataFrame."""
        json = df.to_dict(orient="records")
        deletions = []
        for item in json:
            try:
                deletions.append(FrogReactionDeletion(**item))
            except ValidationError as e:
                logger.error(item)
                logger.error(e.json())

        return FrogReactionDeletions(deletions=deletions)

    def to_df(self) -> pd.DataFrame:
        """Create reaction deletions DataFrame."""

        d: Dict[str, Any] = self.model_dump()
        item = list(d.values())[0]
        df = pd.DataFrame(item)
        if len(df) > 0:
            df.sort_values(by=["reaction"], inplace=True)
            df.index = range(len(df))
            df.loc[
                df.status == StatusCode.INFEASIBLE.value, "value"
            ] = CuratorConstants.VALUE_INFEASIBLE

        return df

from_df(df) staticmethod

Parse FVA from DataFrame.

Source code in src/fbc_curation/frog.py 
261
262
263
264
265
266
267
268
269
270
271
272
273
@staticmethod
def from_df(df: pd.DataFrame) -> FrogReactionDeletions:
    """Parse FVA from DataFrame."""
    json = df.to_dict(orient="records")
    deletions = []
    for item in json:
        try:
            deletions.append(FrogReactionDeletion(**item))
        except ValidationError as e:
            logger.error(item)
            logger.error(e.json())

    return FrogReactionDeletions(deletions=deletions)

to_df()

Create reaction deletions DataFrame.

Source code in src/fbc_curation/frog.py 
275
276
277
278
279
280
281
282
283
284
285
286
287
288
def to_df(self) -> pd.DataFrame:
    """Create reaction deletions DataFrame."""

    d: Dict[str, Any] = self.model_dump()
    item = list(d.values())[0]
    df = pd.DataFrame(item)
    if len(df) > 0:
        df.sort_values(by=["reaction"], inplace=True)
        df.index = range(len(df))
        df.loc[
            df.status == StatusCode.INFEASIBLE.value, "value"
        ] = CuratorConstants.VALUE_INFEASIBLE

    return df

FrogReport

Bases: BaseModel

Definition of the FROG standard.

Source code in src/fbc_curation/frog.py 
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
class FrogReport(BaseModel):
    """Definition of the FROG standard."""

    metadata: FrogMetaData
    objectives: FrogObjectives
    fva: FrogFVA
    reaction_deletions: FrogReactionDeletions
    gene_deletions: FrogGeneDeletions

    def to_json(self, path: Path) -> None:
        """Write FrogReport to JSON format."""
        if not path.parent.exists():
            logger.warning(f"Creating results path: {path.parent}")
            path.mkdir(parents=True)

        # write FROG
        logger.debug(f"{path}")
        with open(path, "w+b") as f_json:
            json_bytes = orjson.dumps(self.model_dump(), option=orjson.OPT_INDENT_2)
            f_json.write(json_bytes)

    @staticmethod
    def from_json(path: Path) -> FrogReport:
        """Read FrogReport from JSON format.

        raises ValidationError

        :path: path to JSON report file
        """
        with open(path, "r+b") as f_json:
            s_json = f_json.read()
            d = orjson.loads(s_json)
            return FrogReport(**d)

    def to_dfs(self) -> Dict[str, pd.DataFrame]:
        """Create report DataFrames."""

        return {
            CuratorConstants.OBJECTIVE_KEY: self.objectives.to_df(),
            CuratorConstants.FVA_KEY: self.fva.to_df(),
            CuratorConstants.GENE_DELETION_KEY: self.gene_deletions.to_df(),
            CuratorConstants.REACTION_DELETION_KEY: self.reaction_deletions.to_df(),
        }

    def to_tsv(self, output_dir: Path) -> None:
        """Write Report TSV and metadata to directory."""
        if not output_dir.exists():
            logger.warning(f"Creating results path: {output_dir}")
            output_dir.mkdir(parents=True)

        # write metadata file
        logger.debug(f"{output_dir / CuratorConstants.METADATA_FILENAME}")
        with open(output_dir / CuratorConstants.METADATA_FILENAME, "w") as f_json:
            # make a copy
            metadata = FrogMetaData(**self.metadata.model_dump())
            metadata.frog_id = f"{metadata.frog_id}_tsv"
            f_json.write(metadata.model_dump_json(indent=2))

        # write reference files (TSV files)
        dfs_dict = self.to_dfs()
        for key, df in dfs_dict.items():
            if key == CuratorConstants.OBJECTIVE_KEY:
                filename = CuratorConstants.OBJECTIVE_FILENAME
            elif key == CuratorConstants.FVA_KEY:
                filename = CuratorConstants.FVA_FILENAME
            elif key == CuratorConstants.GENE_DELETION_KEY:
                filename = CuratorConstants.GENE_DELETION_FILENAME
            elif key == CuratorConstants.REACTION_DELETION_KEY:
                filename = CuratorConstants.REACTION_DELETION_FILENAME
            else:
                raise KeyError(f"Unsupported file: {key}")

            df.to_csv(output_dir / filename, sep="\t", index=False, na_rep="NaN")

    @classmethod
    def from_tsv(cls, path: Path) -> FrogReport:
        """Read fbc curation files from given directory."""

        path_metadata = path / CuratorConstants.METADATA_FILENAME
        path_objective = path / CuratorConstants.OBJECTIVE_FILENAME
        path_fva = path / CuratorConstants.FVA_FILENAME
        path_reaction_deletion = path / CuratorConstants.REACTION_DELETION_FILENAME
        path_gene_deletion = path / CuratorConstants.GENE_DELETION_FILENAME
        df_dict: Dict[str, pd.DataFrame] = dict()

        with open(path_metadata, "r+b") as f_json:
            json_bytes = f_json.read()
            df_dict[CuratorConstants.METADATA_KEY] = orjson.loads(json_bytes)

        for key, path in {
            CuratorConstants.OBJECTIVE_KEY: path_objective,
            CuratorConstants.FVA_KEY: path_fva,
            CuratorConstants.REACTION_DELETION_KEY: path_reaction_deletion,
            CuratorConstants.GENE_DELETION_KEY: path_gene_deletion,
        }.items():
            if not path.exists():
                logger.error(f"Required file for fbc curation does not exist: '{path}'")
            else:
                try:
                    df_dict[key] = pd.read_csv(path, sep="\t")
                except pd.errors.EmptyDataError:
                    df_dict[key] = pd.DataFrame()

        report = FrogReport(
            metadata=FrogMetaData(**df_dict[CuratorConstants.METADATA_KEY]),
            objectives=FrogObjectives.from_df(df_dict[CuratorConstants.OBJECTIVE_KEY]),
            fva=FrogFVA.from_df(df_dict[CuratorConstants.FVA_KEY]),
            reaction_deletions=FrogReactionDeletions.from_df(
                df_dict[CuratorConstants.REACTION_DELETION_KEY]
            ),
            gene_deletions=FrogGeneDeletions.from_df(
                df_dict[CuratorConstants.GENE_DELETION_KEY]
            ),
        )
        return report

    def add_to_omex(
        self, omex: Omex, location_prefix: str = f"./{FROG_PATH_PREFIX}/"
    ) -> None:
        """Add report to omex.

        :param omex: OMEX archive to add report to.
        :param location_prefix: prefix to where to write the FROG files in the OMEX
        """

        with tempfile.TemporaryDirectory() as f_tmp:
            tmp_path: Path = Path(f_tmp)

            # write json
            json_path = tmp_path / CuratorConstants.FROG_FILENAME
            self.to_json(json_path)
            omex.add_entry(
                entry_path=json_path,
                entry=ManifestEntry(
                    location=f"{location_prefix}{CuratorConstants.FROG_FILENAME}",
                    format=EntryFormat.FROG_JSON_V1,
                ),
            )

            # write tsvs with metadata
            self.to_tsv(tmp_path)
            for filename, format in [
                (
                    CuratorConstants.METADATA_FILENAME,
                    EntryFormat.FROG_METADATA_V1,
                ),
                (
                    CuratorConstants.OBJECTIVE_FILENAME,
                    EntryFormat.FROG_OBJECTIVE_V1,
                ),
                (CuratorConstants.FVA_FILENAME, EntryFormat.FROG_FVA_V1),
                (
                    CuratorConstants.REACTION_DELETION_FILENAME,
                    EntryFormat.FROG_REACTIONDELETION_V1,
                ),
                (
                    CuratorConstants.GENE_DELETION_FILENAME,
                    EntryFormat.FROG_GENEDELETION_V1,
                ),
            ]:
                omex.add_entry(
                    entry_path=tmp_path / filename,
                    entry=ManifestEntry(
                        location=f"{location_prefix}{filename}",
                        format=format,
                    ),
                )

add_to_omex(omex, location_prefix=f'./{FROG_PATH_PREFIX}/')

Add report to omex.

:param omex: OMEX archive to add report to. :param location_prefix: prefix to where to write the FROG files in the OMEX

Source code in src/fbc_curation/frog.py 
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
def add_to_omex(
    self, omex: Omex, location_prefix: str = f"./{FROG_PATH_PREFIX}/"
) -> None:
    """Add report to omex.

    :param omex: OMEX archive to add report to.
    :param location_prefix: prefix to where to write the FROG files in the OMEX
    """

    with tempfile.TemporaryDirectory() as f_tmp:
        tmp_path: Path = Path(f_tmp)

        # write json
        json_path = tmp_path / CuratorConstants.FROG_FILENAME
        self.to_json(json_path)
        omex.add_entry(
            entry_path=json_path,
            entry=ManifestEntry(
                location=f"{location_prefix}{CuratorConstants.FROG_FILENAME}",
                format=EntryFormat.FROG_JSON_V1,
            ),
        )

        # write tsvs with metadata
        self.to_tsv(tmp_path)
        for filename, format in [
            (
                CuratorConstants.METADATA_FILENAME,
                EntryFormat.FROG_METADATA_V1,
            ),
            (
                CuratorConstants.OBJECTIVE_FILENAME,
                EntryFormat.FROG_OBJECTIVE_V1,
            ),
            (CuratorConstants.FVA_FILENAME, EntryFormat.FROG_FVA_V1),
            (
                CuratorConstants.REACTION_DELETION_FILENAME,
                EntryFormat.FROG_REACTIONDELETION_V1,
            ),
            (
                CuratorConstants.GENE_DELETION_FILENAME,
                EntryFormat.FROG_GENEDELETION_V1,
            ),
        ]:
            omex.add_entry(
                entry_path=tmp_path / filename,
                entry=ManifestEntry(
                    location=f"{location_prefix}{filename}",
                    format=format,
                ),
            )

from_json(path) staticmethod

Read FrogReport from JSON format.

raises ValidationError

:path: path to JSON report file

Source code in src/fbc_curation/frog.py 
348
349
350
351
352
353
354
355
356
357
358
359
@staticmethod
def from_json(path: Path) -> FrogReport:
    """Read FrogReport from JSON format.

    raises ValidationError

    :path: path to JSON report file
    """
    with open(path, "r+b") as f_json:
        s_json = f_json.read()
        d = orjson.loads(s_json)
        return FrogReport(**d)

from_tsv(path) classmethod

Read fbc curation files from given directory.

Source code in src/fbc_curation/frog.py 
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
@classmethod
def from_tsv(cls, path: Path) -> FrogReport:
    """Read fbc curation files from given directory."""

    path_metadata = path / CuratorConstants.METADATA_FILENAME
    path_objective = path / CuratorConstants.OBJECTIVE_FILENAME
    path_fva = path / CuratorConstants.FVA_FILENAME
    path_reaction_deletion = path / CuratorConstants.REACTION_DELETION_FILENAME
    path_gene_deletion = path / CuratorConstants.GENE_DELETION_FILENAME
    df_dict: Dict[str, pd.DataFrame] = dict()

    with open(path_metadata, "r+b") as f_json:
        json_bytes = f_json.read()
        df_dict[CuratorConstants.METADATA_KEY] = orjson.loads(json_bytes)

    for key, path in {
        CuratorConstants.OBJECTIVE_KEY: path_objective,
        CuratorConstants.FVA_KEY: path_fva,
        CuratorConstants.REACTION_DELETION_KEY: path_reaction_deletion,
        CuratorConstants.GENE_DELETION_KEY: path_gene_deletion,
    }.items():
        if not path.exists():
            logger.error(f"Required file for fbc curation does not exist: '{path}'")
        else:
            try:
                df_dict[key] = pd.read_csv(path, sep="\t")
            except pd.errors.EmptyDataError:
                df_dict[key] = pd.DataFrame()

    report = FrogReport(
        metadata=FrogMetaData(**df_dict[CuratorConstants.METADATA_KEY]),
        objectives=FrogObjectives.from_df(df_dict[CuratorConstants.OBJECTIVE_KEY]),
        fva=FrogFVA.from_df(df_dict[CuratorConstants.FVA_KEY]),
        reaction_deletions=FrogReactionDeletions.from_df(
            df_dict[CuratorConstants.REACTION_DELETION_KEY]
        ),
        gene_deletions=FrogGeneDeletions.from_df(
            df_dict[CuratorConstants.GENE_DELETION_KEY]
        ),
    )
    return report

to_dfs()

Create report DataFrames.

Source code in src/fbc_curation/frog.py 
361
362
363
364
365
366
367
368
369
def to_dfs(self) -> Dict[str, pd.DataFrame]:
    """Create report DataFrames."""

    return {
        CuratorConstants.OBJECTIVE_KEY: self.objectives.to_df(),
        CuratorConstants.FVA_KEY: self.fva.to_df(),
        CuratorConstants.GENE_DELETION_KEY: self.gene_deletions.to_df(),
        CuratorConstants.REACTION_DELETION_KEY: self.reaction_deletions.to_df(),
    }

to_json(path)

Write FrogReport to JSON format.

Source code in src/fbc_curation/frog.py 
336
337
338
339
340
341
342
343
344
345
346
def to_json(self, path: Path) -> None:
    """Write FrogReport to JSON format."""
    if not path.parent.exists():
        logger.warning(f"Creating results path: {path.parent}")
        path.mkdir(parents=True)

    # write FROG
    logger.debug(f"{path}")
    with open(path, "w+b") as f_json:
        json_bytes = orjson.dumps(self.model_dump(), option=orjson.OPT_INDENT_2)
        f_json.write(json_bytes)

to_tsv(output_dir)

Write Report TSV and metadata to directory.

Source code in src/fbc_curation/frog.py 
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
def to_tsv(self, output_dir: Path) -> None:
    """Write Report TSV and metadata to directory."""
    if not output_dir.exists():
        logger.warning(f"Creating results path: {output_dir}")
        output_dir.mkdir(parents=True)

    # write metadata file
    logger.debug(f"{output_dir / CuratorConstants.METADATA_FILENAME}")
    with open(output_dir / CuratorConstants.METADATA_FILENAME, "w") as f_json:
        # make a copy
        metadata = FrogMetaData(**self.metadata.model_dump())
        metadata.frog_id = f"{metadata.frog_id}_tsv"
        f_json.write(metadata.model_dump_json(indent=2))

    # write reference files (TSV files)
    dfs_dict = self.to_dfs()
    for key, df in dfs_dict.items():
        if key == CuratorConstants.OBJECTIVE_KEY:
            filename = CuratorConstants.OBJECTIVE_FILENAME
        elif key == CuratorConstants.FVA_KEY:
            filename = CuratorConstants.FVA_FILENAME
        elif key == CuratorConstants.GENE_DELETION_KEY:
            filename = CuratorConstants.GENE_DELETION_FILENAME
        elif key == CuratorConstants.REACTION_DELETION_KEY:
            filename = CuratorConstants.REACTION_DELETION_FILENAME
        else:
            raise KeyError(f"Unsupported file: {key}")

        df.to_csv(output_dir / filename, sep="\t", index=False, na_rep="NaN")

StatusCode

Bases: str, Enum

Status code for simulation results.

Source code in src/fbc_curation/frog.py 
65
66
67
68
69
class StatusCode(str, Enum):
    """Status code for simulation results."""

    OPTIMAL = "optimal"
    INFEASIBLE = "infeasible"

Tool

Bases: BaseModel

Tool description.

Source code in src/fbc_curation/frog.py 
128
129
130
131
132
133
class Tool(BaseModel):
    """Tool description."""

    name: str = Field(description="Name of tool/software/library.")
    version: Optional[str] = Field(description="Version of tool/software/library.")
    url: Optional[str] = Field(description="URL of tool/software/library.")

fbc_curation.runfrog

Command line tool runfrog for creating FROG reports.

main()

Entry point which runs FROG report script.

The script is registered as runfrog command.

Example

runfrog --input resources/examples/models/e_coli_core.xml --output resources/examples/results/e_coli_core.omex python runfrog.py --input resources/examples/models/e_coli_core.xml --path resources/examples/results/e_coli_core.omex

Source code in src/fbc_curation/runfrog.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def main() -> None:
    """Entry point which runs FROG report script.

    The script is registered as `runfrog` command.

    Example:
        runfrog --input resources/examples/models/e_coli_core.xml
          --output resources/examples/results/e_coli_core.omex
        python runfrog.py --input resources/examples/models/e_coli_core.xml
          --path resources/examples/results/e_coli_core.omex

    """

    import optparse
    import sys

    parser = optparse.OptionParser()
    parser.add_option(
        "-i",
        "--input",
        action="store",
        dest="input_path",
        help="(required) path to COMBINE archive (OMEX) with SBML model or "
        "an SBML model",
    )
    parser.add_option(
        "-o",
        "--output",
        action="store",
        dest="output_path",
        help="(required) omex output path to write FROG",
    )
    # parser.add_option(
    #     "-r",
    #     "--reference",
    #     action="store",
    #     dest="reference_path",
    #     help="(optional) path to COMBINE archive (OMEX) with FROG results "
    #     "to include in comparison",
    # )

    console.rule(style="white")
    console.print(":frog: FBC CURATION FROG ANALYSIS :frog:")
    console.print(
        f"Version {__version__} (https://github.com/matthiaskoenig/fbc_curation)"
    )
    console.print(f"Citation {__citation__}")
    console.rule(style="white")

    options, args = parser.parse_args()

    def _parser_message(text: str) -> None:
        console.print(text)
        parser.print_help()
        console.rule(style="white")
        sys.exit(1)

    if not options.input_path:
        _parser_message("Required argument '--input' missing")
    if not options.output_path:
        _parser_message("Required argument '--output' missing")

    input_path = Path(options.input_path)
    if not input_path.exists():
        _parser_message(
            f"--input '{options.input_path}' does not exist, ensure valid model or "
            f"OMEX path."
        )

    output_path = Path(options.output_path)
    if not str(output_path).endswith(".omex"):
        _parser_message(
            f"--output '{options.input_path}' output path must end in '.omex'"
        )

    # reference_path: Optional[Path] = None
    # if options.reference_path:
    #     reference_path = Path(options.reference_path)
    #     if not reference_path.exists():
    #         _parser_message(
    #             f"--reference '{options.reference_path}' does not exist, ensure "
    #             f"valid reference path."
    #         )

    run_frog(
        source_path=input_path,
        omex_path=output_path,
    )

    model_reports = FrogComparison.read_reports_from_omex(omex_path=output_path)
    for _, reports in model_reports.items():
        FrogComparison.compare_reports(reports=reports)

fbc_curation.worker

Worker.

Here the tasks are defined.

frog_task(source_path_str, omex_path_str=None)

Run FROG task and create JSON for omex path.

This function should not be called directly. Use the run_frog function instead. Path can be either Omex or an SBML file.

:param omex_path_str: Path to OMEX for results.

Source code in src/fbc_curation/worker.py 
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def frog_task(
    source_path_str: str,
    omex_path_str: Optional[str] = None,
) -> Dict[str, Any]:
    """Run FROG task and create JSON for omex path.

    This function should not be called directly. Use the `run_frog` function
    instead. Path can be either Omex or an SBML file.

    :param omex_path_str: Path to OMEX for results.
    """
    logger.info(f"Loading '{source_path_str}'")

    omex_path = Path(source_path_str)
    if not omex_path.exists():
        raise IOError(f"Path does not exist: '{omex_path}'")
    if not omex_path.is_file():
        raise IOError(f"Path is not a file: '{omex_path}'")

    # move data for task

    if Omex.is_omex(omex_path):
        omex = Omex().from_omex(omex_path)
    else:
        # Path is SBML we create a new archive
        omex = Omex()
        omex.add_entry(
            entry_path=omex_path,
            entry=ManifestEntry(
                location=f"./{omex_path.name}", format=EntryFormat.SBML, master=True
            ),
        )

    content = {"manifest": omex.manifest.model_dump(), "frogs": {}}

    # Add FROG JSON for all SBML files
    entry: ManifestEntry
    for entry in omex.manifest.entries:
        if entry.is_sbml():
            # TODO: check that SBML model with FBC information

            report_dict = {}
            for curator_key in ["cobrapy", "cameo"]:
                sbml_path: Path = omex.get_path(entry.location)
                report: FrogReport = _frog_for_sbml(
                    source=sbml_path, curator_key=curator_key
                )

                # add FROG files to archive
                report.add_to_omex(
                    omex, location_prefix=f"./{FROG_PATH_PREFIX}/{curator_key}/"
                )

                # add JSON to response
                report_dict[curator_key] = report.model_dump()

            # store all reports for SBML entry
            content["frogs"][entry.location] = report_dict

    # save archive for download
    if not omex_path_str:
        raise ValueError(
            "The 'omex_path_str' argument must be set (if not executed "
            "within a celery Task)."
        )

    omex_path = Path(omex_path_str)
    console.rule("Write OMEX", style="white")
    omex.to_omex(omex_path=omex_path)

    return content

run_frog(source_path, omex_path)

Create FROG report for given SBML or OMEX source.

This function creates the FROG report and stores the results with the model in a COMBINE archive.

:param source_path: Path for SBML model to create FROG for, or a COMBINE archive (omex) which contains an SBML model. :param omex_path: Path for COMBINE archive (omex) with FROG results. The content of the file will be overwritten!

Source code in src/fbc_curation/worker.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def run_frog(source_path: Path, omex_path: Path) -> None:
    """Create FROG report for given SBML or OMEX source.

    This function creates the FROG report and stores the results with the
    model in a COMBINE archive.

    :param source_path: Path for SBML model to create FROG for, or a COMBINE archive
      (omex) which contains an SBML model.
    :param omex_path: Path for COMBINE archive (omex) with FROG results. The content
      of the file will be overwritten!
    """
    frog_task(
        source_path_str=str(source_path),
        omex_path_str=str(omex_path),
    )

fbc_curation.compare

Comparison of FROG results.

FrogComparison

FrogComparison.

Class for comparing FROG reports. Allows to check if multiple FROGs give the same results.

Source code in src/fbc_curation/compare.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
class FrogComparison:
    """FrogComparison.

    Class for comparing FROG reports. Allows to check if multiple FROGs
    give the same results.
    """

    absolute_tolerance: float = 1e-3
    relative_tolerance: float = 1e-3

    @staticmethod
    def read_reports_from_omex(omex_path: Path) -> Dict[str, Dict[str, FrogReport]]:
        """Read all reports from JSON and TSVs.

        Returns dictionary of {model_location: ...}

        """
        reports: List[FrogReport] = []
        omex = Omex.from_omex(omex_path)
        for entry in omex.manifest.entries:
            if entry.format == EntryFormat.FROG_JSON_V1:
                report = FrogReport.from_json(path=omex.get_path(entry.location))
                reports.append(report)
            elif entry.format == EntryFormat.FROG_METADATA_V1:
                path = omex.get_path(entry.location).parent
                report = FrogReport.from_tsv(path)
                reports.append(report)

        # get model reports per model
        model_reports: Dict[str, Dict[str, FrogReport]] = {}
        for report in reports:
            model_location = report.metadata.model_location
            d = model_reports.get(model_location, {})
            frog_id = report.metadata.frog_id
            if frog_id in d:
                logger.error(
                    f"duplicate FROG report: '{frog_id}' for '{model_location}'"
                )
            d[frog_id] = report
            model_reports[model_location] = d

        info = {loc: list(item.keys()) for loc, item in model_reports.items()}
        logger.info(f"Reports in omex:\n{info}")

        return model_reports

    # TODO: implement comparison result and return results

    @staticmethod
    def compare_reports(reports: Dict[str, FrogReport]) -> bool:
        """Compare results against each other.

        Compare all matrices pairwise, i.e., comparison matrix for
        - objective
        - FVA
        - gene deletions
        - reaction deletions
        """
        console.rule("Comparison of FROGReports", style="white")

        num_reports = len(reports)
        all_equal: bool = True
        # only comparing comparison between two data frames

        data: Dict[str, Dict[str, pd.DataFrame]] = {}

        # DataFrames for report
        report_keys = list(reports.keys())
        for report_key, report in reports.items():
            # all DataFrames for single report
            frog_dfs: Dict[str, pd.DataFrame] = report.to_dfs()
            data[report_key] = frog_dfs

        # Perform comparisons between all reports
        for key in [
            CuratorConstants.OBJECTIVE_KEY,
            CuratorConstants.FVA_KEY,
            CuratorConstants.REACTION_DELETION_KEY,
            CuratorConstants.GENE_DELETION_KEY,
        ]:
            mat_equal = np.zeros(shape=(num_reports, num_reports))

            # do all pairwise comparisons
            dfs: List[pd.DataFrame] = [
                data[report_key][key] for report_key in reports.keys()
            ]
            for p, df1 in enumerate(dfs):
                for q, df2 in enumerate(dfs):

                    fields: List[str]
                    equal = True
                    if key in [
                        CuratorConstants.OBJECTIVE_KEY,
                        CuratorConstants.REACTION_DELETION_KEY,
                        CuratorConstants.GENE_DELETION_KEY,
                    ]:
                        fields = ["value"]
                    elif key == CuratorConstants.FVA_KEY:
                        fields = ["flux", "minimum", "maximum"]

                    for field in fields:
                        if field in df1.columns and field in df2.columns:
                            equal_field = np.allclose(
                                df1[field].values,
                                df2[field].values,
                                atol=FrogComparison.absolute_tolerance,
                                rtol=FrogComparison.relative_tolerance,
                                equal_nan=True,
                            )
                            equal = equal and equal_field

                    mat_equal[p, q] = int(equal)

                    if not equal:
                        logger.warning(
                            f"difference: '{report_keys[p]}' vs '{report_keys[q]}'"
                        )
                        equal_vec = np.isclose(
                            df1[field].values,
                            df2[field].values,
                            atol=FrogComparison.absolute_tolerance,
                            rtol=FrogComparison.relative_tolerance,
                            equal_nan=True,
                        )
                        df_diff = pd.concat([df1[~equal_vec], df2[~equal_vec]])
                        if "reaction" in df_diff.columns:
                            df_diff.sort_values(by=["reaction"], inplace=True)
                        elif "gene" in df_diff.columns:
                            df_diff.sort_values(by=["gene"], inplace=True)
                        console.print(df_diff)

            df_equal = pd.DataFrame(
                mat_equal, columns=list(report_keys), index=list(report_keys), dtype=int
            )
            console.print(f"--- {key} ---")
            console.print(df_equal)
            all_equal = (
                all_equal
                and np.sum(np.sum(df_equal.values)) == num_reports * num_reports
            )

        console.rule(style="white")
        console.print(f"Equal: {all_equal}")
        console.rule(style="white")
        return bool(all_equal)

compare_reports(reports) staticmethod

Compare results against each other.

Compare all matrices pairwise, i.e., comparison matrix for - objective - FVA - gene deletions - reaction deletions

Source code in src/fbc_curation/compare.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
@staticmethod
def compare_reports(reports: Dict[str, FrogReport]) -> bool:
    """Compare results against each other.

    Compare all matrices pairwise, i.e., comparison matrix for
    - objective
    - FVA
    - gene deletions
    - reaction deletions
    """
    console.rule("Comparison of FROGReports", style="white")

    num_reports = len(reports)
    all_equal: bool = True
    # only comparing comparison between two data frames

    data: Dict[str, Dict[str, pd.DataFrame]] = {}

    # DataFrames for report
    report_keys = list(reports.keys())
    for report_key, report in reports.items():
        # all DataFrames for single report
        frog_dfs: Dict[str, pd.DataFrame] = report.to_dfs()
        data[report_key] = frog_dfs

    # Perform comparisons between all reports
    for key in [
        CuratorConstants.OBJECTIVE_KEY,
        CuratorConstants.FVA_KEY,
        CuratorConstants.REACTION_DELETION_KEY,
        CuratorConstants.GENE_DELETION_KEY,
    ]:
        mat_equal = np.zeros(shape=(num_reports, num_reports))

        # do all pairwise comparisons
        dfs: List[pd.DataFrame] = [
            data[report_key][key] for report_key in reports.keys()
        ]
        for p, df1 in enumerate(dfs):
            for q, df2 in enumerate(dfs):

                fields: List[str]
                equal = True
                if key in [
                    CuratorConstants.OBJECTIVE_KEY,
                    CuratorConstants.REACTION_DELETION_KEY,
                    CuratorConstants.GENE_DELETION_KEY,
                ]:
                    fields = ["value"]
                elif key == CuratorConstants.FVA_KEY:
                    fields = ["flux", "minimum", "maximum"]

                for field in fields:
                    if field in df1.columns and field in df2.columns:
                        equal_field = np.allclose(
                            df1[field].values,
                            df2[field].values,
                            atol=FrogComparison.absolute_tolerance,
                            rtol=FrogComparison.relative_tolerance,
                            equal_nan=True,
                        )
                        equal = equal and equal_field

                mat_equal[p, q] = int(equal)

                if not equal:
                    logger.warning(
                        f"difference: '{report_keys[p]}' vs '{report_keys[q]}'"
                    )
                    equal_vec = np.isclose(
                        df1[field].values,
                        df2[field].values,
                        atol=FrogComparison.absolute_tolerance,
                        rtol=FrogComparison.relative_tolerance,
                        equal_nan=True,
                    )
                    df_diff = pd.concat([df1[~equal_vec], df2[~equal_vec]])
                    if "reaction" in df_diff.columns:
                        df_diff.sort_values(by=["reaction"], inplace=True)
                    elif "gene" in df_diff.columns:
                        df_diff.sort_values(by=["gene"], inplace=True)
                    console.print(df_diff)

        df_equal = pd.DataFrame(
            mat_equal, columns=list(report_keys), index=list(report_keys), dtype=int
        )
        console.print(f"--- {key} ---")
        console.print(df_equal)
        all_equal = (
            all_equal
            and np.sum(np.sum(df_equal.values)) == num_reports * num_reports
        )

    console.rule(style="white")
    console.print(f"Equal: {all_equal}")
    console.rule(style="white")
    return bool(all_equal)

read_reports_from_omex(omex_path) staticmethod

Read all reports from JSON and TSVs.

Returns dictionary of {model_location: ...}

Source code in src/fbc_curation/compare.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
@staticmethod
def read_reports_from_omex(omex_path: Path) -> Dict[str, Dict[str, FrogReport]]:
    """Read all reports from JSON and TSVs.

    Returns dictionary of {model_location: ...}

    """
    reports: List[FrogReport] = []
    omex = Omex.from_omex(omex_path)
    for entry in omex.manifest.entries:
        if entry.format == EntryFormat.FROG_JSON_V1:
            report = FrogReport.from_json(path=omex.get_path(entry.location))
            reports.append(report)
        elif entry.format == EntryFormat.FROG_METADATA_V1:
            path = omex.get_path(entry.location).parent
            report = FrogReport.from_tsv(path)
            reports.append(report)

    # get model reports per model
    model_reports: Dict[str, Dict[str, FrogReport]] = {}
    for report in reports:
        model_location = report.metadata.model_location
        d = model_reports.get(model_location, {})
        frog_id = report.metadata.frog_id
        if frog_id in d:
            logger.error(
                f"duplicate FROG report: '{frog_id}' for '{model_location}'"
            )
        d[frog_id] = report
        model_reports[model_location] = d

    info = {loc: list(item.keys()) for loc, item in model_reports.items()}
    logger.info(f"Reports in omex:\n{info}")

    return model_reports